"""This is a library that works with mod_python to allow you to use
OpenID as access control for any resource that Apache serves. See
README for more details.
Session Keys
============
This module uses mod_python sessions. This is an overview of the data
that is stored in the session:
`cookied_user` (OpenID identifier (unicode or str))
If present, this is the currently logged in user. This is set
upon successful login and cleared when logout is called.
`target` (URL (str))
The URL that the login code will redirect to once authentication
has completed successfully. This is set by the `AccessHandler`
(`protect`) and cleared when there is a successful login or
replaced by another call to `protect`.
`logout` (bool)
Whether the last login action was a logout. This is used to know
whether or not to attempt auto-login. When this key is present
and true, auto-login is not attempted. Viewing the login page
clears this flag.
`message` (str)
A key into the messages dictionary. The login page displays this
message and clears this value. This is set whenever there is a
message that needs to be displayed to the user.
Cookies
=======
This module sets a session cookie, as well as the following cookies:
`mpopenid.last_user` (OpenID identifier (unicode or str))
If present, this is an advisory cookie that indicates to the
library that it should attempt to seamlessly authenticate that
identifier using 'immediate mode'. This cookie is set upon
successful login and cleared if the auto authentication attempt
fails.
"""
__copyright__ = """Copyright (C) 2005, 2006 JanRain, Inc."""
__license__ = """This is Free Software, under the terms of the GNU GPL.
See COPYING for details."""
__version__ = '1.2.0-pre4'
def protect(apache_request):
"""This is the function that protects a directory.
It should be used as a PythonAccessHandler. e.g.:
PythonAccessHandler mpopenid::protect
Options used by this handler:
# whitespace separated OpenID identifiers
PythonOption authorized-users ""
# a URL to a list of OpenID identifiers
PythonOption authorized-users-list-url ""
If both authorized-users and authorized-users-list-url are specified,
the lists will be concatenated. If neither is specified, no users
will have access. (this is not very useful)
"""
oid_req = OpenIDProtect(apache_request)
try:
return oid_req.protect()
finally:
oid_req.session.save()
def openid(apache_request):
"""This is the function that handles OpenID logins
It should be used as a PythonHandler. e.g.:
SetHandler mod_python
PythonHandler mpopenid::openid
Options used by this handler:
# (defaults to "stateless")
PythonOption store-type file
# Used only if it's a file store
PythonOption store-directory /somewhere/
"""
oid_req = OpenIDLogin(apache_request)
try:
return oid_req.openid()
finally:
oid_req.session.save()
##### The rest of this file is support for the functions above #####
import sys
import time
import os.path
import urllib
import urlparse
from cgi import escape
try:
from mod_python import apache, util, Cookie, Session
except ImportError:
print >>sys.stderr, ("Unable to import mod_python code. Continuing "
"anyway (for pychecker)")
sys.stderr.flush()
apache = util = Cookie = Session = None
try:
from openid.consumer import consumer
except ImportError:
print >>sys.stderr, (
'No JanRain OpenID library was found. Checked in:\n' + str(sys.path))
raise
try:
from openid.urinorm import urinorm
except ImportError:
# An OpenID library that does not yet have urinorm. Anything
# before 2.0 should be in this state. Not normalizing URLs just
# means that you have to be more careful about what you type when
# you're listing the authorized URLs.
from openid.oidutil import normalizeUrl as urinorm
try:
# 1.2, 2.0
from openid.yadis.discover import DiscoveryFailure
from openid.fetchers import HTTPFetchingError
except ImportError:
# 1.1
from yadis.discover import DiscoveryFailure
from urljr.fetchers import HTTPFetchingError
from openid.store import filestore, dumbstore
expected_discovery_exceptions = (DiscoveryFailure, HTTPFetchingError)
message_text = {
'empty':('message', 'Enter an OpenID URL to continue'),
'cancel':('message', 'Authorization cancelled'),
'http_failed':('error',
'There was an error communicating with the server'),
'failure':('error', 'The server reported an error'),
'denied':('error', 'That identity URL has not been granted '
'access to this resource'),
'discovery':('error', 'Failed to discover an OpenID server'),
}
login_page_tmpl = '''\
%(title)s
%(title)s
%(resource)s%(message)s
Enter your OpenID identity URL to continue.
'''
def parseAuthorizedURLs(data):
"""Parse and normalize a string containing a whitespace-separated
list of URLs"""
urls = []
for s in data.split():
url = s.strip()
if not url:
continue
parsed = urlparse.urlparse(url)
if not (parsed[0] and parsed[1]):
url = 'http://' + url
urls.append(urinorm(url))
return urls
class OpenIDAccessRequest(object):
# Cache for store instances
_file_stores = {}
_max_store_cache_size = 5
# Cache for authorized identities
_authorized_cache = {}
_max_authorized_cache_size = 100
# Time out sessions after a week of no access
session_timeout = 60 * 60 * 24 * 7
def __init__(self, apache_request):
self.apache_request = apache_request
self.options = apache_request.get_options()
self.session = Session.Session(
apache_request,
timeout=self.session_timeout,
lock=False)
def get_cookied_user(self):
"""Get the openid_identifier cookie for this request
mod_python.Request -> NoneType or str
"""
return self.session.get('cookied_user')
def set_cookied_user(self, openid_identifier):
"""Set the openid_identifier cookie for this user
(mod_python.Request, str, int) -> NoneType
"""
self.session['cookied_user'] = openid_identifier
cookied_user = property(get_cookied_user, set_cookied_user)
def getServerURL(self):
"""Return a URL to the root of the server that is serving this
request.
mod_python.Request -> str
"""
host = self.apache_request.hostname
port = self.apache_request.connection.local_addr[1]
# Don't include the default port number in the URL
if self.apache_request.subprocess_env.get('HTTPS', 'off') == 'on':
default_port = 443
proto = 'https'
else:
default_port = 80
proto = 'http'
if port == default_port:
server_url = '%s://%s/' % (proto, host)
else:
server_url = '%s://%s:%s/' % (proto, host, port)
return server_url
def getActionPath(self):
raise NotImplementedError
def loginRedirect(self, message=None, target=None, logout=False):
"""Issue a 302 redirect to the OpenID login page.
(mod_python.Request, str or NoneType, str or NoneType) ->
apache.SERVER_RETURN"""
if target:
self.session['target'] = target
self.session['message'] = message
self.session['logout'] = logout
self.redirect(self.actionURL('login'))
def redirect(self, url):
# This function raises an exception, so it will halt anything
# that calls it. This is probably what you want, but beware!
util.redirect(self.apache_request, url)
def actionURL(self, action):
"""Generate a URL that performs the given action. This depends
on knowing where the actions live.
"""
return urlparse.urljoin(self.getServerURL(), self.getActionPath() + action)
class OpenIDLogin(OpenIDAccessRequest):
# How long to keep the last_user cookie that is used to automate
# login
auto_login_lifetime = 24 * 60 * 60 * 365
def getConsumer(self):
"""Get the consumer that should be used for this request. This is
also responsible for creating stores. If you want to use a
different kind of store, change this function to support creating
it.
mod_python.Request -> consumer.OpenIDConsumer or apache.SERVER_RETURN
"""
store_type = self.options.get('store-type', 'stateless')
if store_type == 'file':
store_dir = self.options['store-directory']
store_dir = os.path.normpath(store_dir)
store = self._file_stores.get(store_dir)
if store is None:
if len(self._file_stores) > self._max_store_cache_size:
self._file_stores.clear()
store = filestore.FileOpenIDStore(store_dir)
self._file_stores[store_dir] = store
elif store_type == 'stateless':
store = dumbstore.DumbStore("XXX unused")
else:
# XXX: implement other store types (SQL...)
self.apache_request.log_error(
'Unknown OpenID store type: %r' % (store_type,))
raise apache.SERVER_RETURN, apache.HTTP_INTERNAL_SERVER_ERROR
return consumer.Consumer(self.session, store)
def getLastUser(self):
cookies = Cookie.get_cookies(self.apache_request)
cookie = cookies.get('mpopenid.last_user')
if cookie is None:
self.apache_request.log_error('No last user')
return None
self.apache_request.log_error('Got last user: %r' % (cookie.value,))
return cookie.value
def setLastUser(self, username):
assert isinstance(username, basestring)
expires = time.time() + self.auto_login_lifetime
Cookie.add_cookie(self.apache_request,
'mpopenid.last_user', username, expires=expires)
self.apache_request.log_error('Set last user: %r' % (username,))
def delLastUser(self):
Cookie.add_cookie(
self.apache_request, 'mpopenid.last_user', '', expires=0)
self.apache_request.log_error('Deleted last user')
def fillLoginPage(self, openid_identifier, messages):
"""Generate the HTML for the login page
"""
message_chunks = []
for name in messages:
message_info = message_text.get(name)
if message_info is None:
message_info = ('error', 'An error occurred')
chunk = "%s
" % message_info
message_chunks.append(chunk)
if self.cookied_user:
chunk = ("You are currently logged "
"in as %s. (
logout)
Authorization is required to access '
'%s
') % (escape(target),)
else:
resource = ''
if not openid_identifier:
openid_identifier = ''
return login_page_tmpl % dict(
action=escape(self.actionURL('login'), True),
resource=resource,
title='OpenID Authentication Required',
message=message_html,
openid_identifier=escape(openid_identifier, True),
)
def getActionPath(self):
"""Find the URL to the actions
"""
# First check configuration
action_path = self.options.get('action-path')
if action_path is None:
# Default to path for this request (since this handler *is*
# the action path)
#
# Too bad there is no way to just get the Apache path to
# this handler. It depends on whether this handler appears
# in a , , or section. (see
# )
path_info = self.apache_request.path_info
if path_info:
assert self.apache_request.uri.endswith(path_info)
action_path = self.apache_request.uri[:-len(path_info)]
else:
action_path = self.apache_request.uri
# XXX: if this is '/' or '' there is probably something
# wrong with the config because it'd be pretty silly to
# have this handler at the root of a server. We should
# tell the user, but how?
# Make sure that there is a trailing slash on the action path
if not action_path or action_path[-1] != '/':
action_path += '/'
return action_path
def openid(self):
"""Dispatch to the appropriate access control action.
"""
action_path = self.getActionPath()
if self.apache_request.uri.startswith(action_path):
action = self.apache_request.uri[len(action_path):]
else:
self.apache_request.log_error(
'Action path does not match my URL. Configuration problem? '
'(action_path=%r, uri=%r)' %
(action_path, self.apache_request.uri))
self.loginRedirect()
try:
handler = getattr(self, 'do_' + action)
except AttributeError:
# An action we don't know about was called.
self.apache_request.log_error(
"Unknown OpenID access control action: %r" % (action,))
self.loginRedirect()
else:
return handler()
def do_logout(self):
try:
del self.session['cookied_user']
except KeyError:
pass
self.loginRedirect(logout=True)
def do_login(self):
"""Show a login page for setting the OpenID cookie.
"""
form = util.FieldStorage(self.apache_request)
openid_identifier = form.getfirst(
'openid_identifier', self.cookied_user)
immediate = False
if self.session.get('logout'):
del self.session['logout']
else:
last_user = self.getLastUser()
if not openid_identifier and last_user:
openid_identifier = last_user
immediate = True
message = self.session.get('message', None)
if message is None:
messages = []
else:
messages = [message]
if self.apache_request.method == 'POST' or immediate:
if openid_identifier:
consumer = self.getConsumer()
try:
auth_request = consumer.begin(openid_identifier)
except expected_discovery_exceptions:
# Print traceback to Apache log (stderr goes there)
sys.excepthook(*sys.exc_info())
sys.stderr.flush()
# Note that this was an exception in discovery
messages.append('discovery')
else:
# Do the redirect to the OpenID server
redirect_url = auth_request.redirectURL(
self.getServerURL(),
self.actionURL('return'),
immediate)
self.redirect(redirect_url)
else:
messages.append('empty')
text = self.fillLoginPage(openid_identifier, messages)
self.apache_request.content_type = 'text/html; charset=UTF-8'
self.apache_request.set_content_length(len(text))
self.apache_request.write(text)
self.session['message'] = None
return apache.OK
def do_return(self):
"""Handle a response from the OpenID server. Always redirects.
mod_python.Request -> apache.SERVER_RETURN
"""
form = util.FieldStorage(self.apache_request)
query = {}
for k in form.keys():
query[k] = form.getfirst(k)
consumer = self.getConsumer()
response = consumer.complete(query)
if response is None:
self.loginRedirect(message='failure')
elif response.status == consumer.SUCCESS:
# Set the cookie and then redirect back to the target
self.cookied_user = response.identity_url
self.setLastUser(response.identity_url)
target = self.session.get('target')
if not target:
target = self.actionURL('login')
self.session['target'] = None
self.redirect(target)
elif response.status == consumer.CANCEL:
self.loginRedirect(message='cancel')
elif response.status == consumer.FAILURE:
self.loginRedirect(message='failure')
elif response.status == consumer.SETUP_NEEDED:
# Silently redirect to login
self.delLastUser()
self.loginRedirect()
else:
assert False, response.status
class OpenIDProtect(OpenIDAccessRequest):
def getAuthorizedUsers(self):
"""Get all authorized identifiers for this request
"""
authorized_s = self.options.get('authorized-users', None)
if authorized_s:
try:
authorized_list = self._authorized_cache[authorized_s]
except KeyError:
if (len(self._authorized_cache) ==
self._max_authorized_cache_size):
self._authorized_cache.clear()
authorized_list = parseAuthorizedURLs(authorized_s)
self._authorized_cache[authorized_s] = authorized_list
else:
authorized_list = []
authorized_list_url = self.options.get(
'authorized-users-list-url', None)
if authorized_list_url:
try:
url_handle = urllib.urlopen(authorized_list_url)
except IOError:
self.apache_request.log_error(
'Failed to fetch authorized list URL %r'
% (authorized_list_url,))
# Write traceback to Apache log
sys.excepthook(*sys.exc_info())
sys.stderr.flush()
else:
data = url_handle.read()
url_handle.close()
authorized_list.extend(parseAuthorizedURLs(data))
return authorized_list
def getActionPath(self):
"""Find the URL to the actions
"""
action_path = self.options.get('action-path')
if action_path is None:
# the path where *Handler directive was specified
protected_path = self.apache_request.hlist.directory
if protected_path:
docroot = self.apache_request.document_root()
protected_path = protected_path[len(docroot):]
if not protected_path or protected_path[-1] != '/':
protected_path += '/'
else:
self.apache_request.log_error(
'No action-path specified and the protect directive is '
'in a block. Defaulting to /openid for login '
'actions.')
protected_path = '/'
action_path = protected_path + 'openid/'
elif not action_path or action_path[-1] != '/':
action_path += '/'
return action_path
def protect(self):
"""Only allow the request to proceed if the openid_identifier
has authenticated with OpenID and is in the authorized list.
The list can be in the authorized-users setting, with a
supplemental list specified by the authorized-users-list-url,
which should be text/plain and be a list of OpenIDs, one per
line.
mod_python.Request -> int or apache.SERVER_RETURN
"""
# Do not apply rule if this is handled by one of the OpenID
# access control actions.
action_path = self.getActionPath()
if self.apache_request.uri.startswith(action_path):
return apache.OK
request_uri = urlparse.urljoin(
self.getServerURL(), self.apache_request.uri)
# Check to see if cookied openid_identifier is authorized
if self.cookied_user:
authorized_users = self.getAuthorizedUsers()
if self.cookied_user in authorized_users:
self.apache_request.user = self.cookied_user
# Do not cache this access-controlled page
self.apache_request.headers_out['Cache-Control'] = 'no-cache'
return apache.OK
else:
self.apache_request.log_error(
'Unauthorized access attempt from %r for %r' %
(self.cookied_user, request_uri))
message = 'denied'
else:
# Initial request with no openid_identifier cookie, so no message.
message = None
# The redirects only work for GET, so just return FORBIDDEN if
# it's any other method.
if self.apache_request.method != 'GET':
raise apache.SERVER_RETURN, apache.HTTP_FORBIDDEN
# cookied_user not authorized or not set, so redirect to login
self.loginRedirect(message, request_uri)