# Copyright
"""Utilities for building WSGI commands.
See Also
--------
:py:mod:`libbe.command.serve_storage` and
:py:mod:`libbe.command.serve_commands`.
"""
import copy
import hashlib
import logging
import re
import StringIO
import sys
import time
import traceback
import types
import urllib
import urlparse
import wsgiref.simple_server
try:
import cherrypy
import cherrypy.wsgiserver
except ImportError:
cherrypy = None
if cherrypy != None:
try: # CherryPy >= 3.2
import cherrypy.wsgiserver.ssl_builtin
except ImportError: # CherryPy <= 3.1.X
cherrypy.wsgiserver.ssl_builtin = None
try:
import OpenSSL
except ImportError:
OpenSSL = None
import libbe.util.encoding
import libbe.command
import libbe.command.base
import libbe.storage
if libbe.TESTING == True:
import doctest
import unittest
import wsgiref.validate
try:
import cherrypy.test.webtest
cherrypy_test_webtest = True
except ImportError:
cherrypy_test_webtest = None
class HandlerError (Exception):
def __init__(self, code, msg, headers=[]):
super(HandlerError, self).__init__('{} {}'.format(code, msg))
self.code = code
self.msg = msg
self.headers = headers
class Unauthenticated (HandlerError):
def __init__(self, realm, msg='User Not Authenticated', headers=[]):
super(Unauthenticated, self).__init__(401, msg, headers+[
('WWW-Authenticate','Basic realm="{}"'.format(realm))])
class Unauthorized (HandlerError):
def __init__(self, msg='User Not Authorized', headers=[]):
super(Unauthorized, self).__init__(403, msg, headers)
class User (object):
def __init__(self, uname=None, name=None, passhash=None, password=None):
self.uname = uname
self.name = name
self.passhash = passhash
if passhash is None:
if password is not None:
self.passhash = self.hash(password)
else:
assert password is None, (
'Redundant password {} with passhash {}'.format(
password, passhash))
self.users = None
def from_string(self, string):
string = string.strip()
fields = string.split(':')
if len(fields) != 3:
raise ValueError, '{}!=3 fields in "{}"'.format(
len(fields), string)
self.uname,self.name,self.passhash = fields
def __str__(self):
return ':'.join([self.uname, self.name, self.passhash])
def __cmp__(self, other):
return cmp(self.uname, other.uname)
def hash(self, password):
return hashlib.sha1(password).hexdigest()
def valid_login(self, password):
if self.hash(password) == self.passhash:
return True
return False
def set_name(self, name):
self._set_property('name', name)
def set_password(self, password):
self._set_property('passhash', self.hash(password))
def _set_property(self, property, value):
if self.uname == 'guest':
raise Unauthorized(
'guest user not allowed to change {}'.format(property))
if (getattr(self, property) != value and
self.users is not None):
self.users.changed = True
setattr(self, property, value)
class Users (dict):
def __init__(self, filename=None):
super(Users, self).__init__()
self.filename = filename
self.changed = False
def load(self):
if self.filename is None:
return
user_file = libbe.util.encoding.get_file_contents(
self.filename, decode=True)
self.clear()
for line in user_file.splitlines():
user = User()
user.from_string(line)
self.add_user(user)
def save(self):
if self.filename is not None and self.changed:
lines = []
for user in sorted(self.users):
lines.append(str(user))
libbe.util.encoding.set_file_contents(self.filename)
self.changed = False
def add_user(self, user):
assert user.users is None, user.users
user.users = self
self[user.uname] = user
def valid_login(self, uname, password):
return (uname in self and
self[uname].valid_login(password))
class WSGI_Object (object):
"""Utility class for WGSI clients and middleware.
For details on WGSI, see `PEP 333`_
.. _PEP 333: http://www.python.org/dev/peps/pep-0333/
"""
def __init__(self, logger=None, log_level=logging.INFO, log_format=None):
self.logger = logger
self.log_level = log_level
if log_format is None:
self.log_format = (
'{REMOTE_ADDR} - {REMOTE_USER} [{time}] '
'"{REQUEST_METHOD} {REQUEST_URI} {HTTP_VERSION}" '
'{status} {bytes} "{HTTP_REFERER}" "{HTTP_USER_AGENT}"')
else:
self.log_format = log_format
def __call__(self, environ, start_response):
if self.logger is not None:
self.logger.log(
logging.DEBUG, 'entering {}'.format(self.__class__.__name__))
ret = self._call(environ, start_response)
if self.logger is not None:
self.logger.log(
logging.DEBUG, 'leaving {}'.format(self.__class__.__name__))
return ret
def _call(self, environ, start_response):
"""The main WSGI entry point."""
raise NotImplementedError
# start_response() is a callback for setting response headers
# start_response(status, response_headers, exc_info=None)
# status is an HTTP status string (e.g., "200 OK").
# response_headers is a list of 2-tuples, the HTTP headers in
# key-value format.
# exc_info is used in exception handling.
#
# The application function then returns an iterable of body chunks.
def error(self, environ, start_response, error, message, headers=[]):
"""Make it easy to call start_response for errors."""
response = '{} {}'.format(error, message)
self.log_request(environ, status=response, bytes=len(message))
start_response(response,
[('Content-Type', 'text/plain')]+headers)
return [message]
def log_request(self, environ, status='-1 OK', bytes=-1):
if self.logger is None or self.logger.level > self.log_level:
return
req_uri = urllib.quote(environ.get('SCRIPT_NAME', '')
+ environ.get('PATH_INFO', ''))
if environ.get('QUERY_STRING'):
req_uri += '?' + environ['QUERY_STRING']
start = time.localtime()
if time.daylight:
offset = time.altzone / 60 / 60 * -100
else:
offset = time.timezone / 60 / 60 * -100
if offset >= 0:
offset = '+{:04d}'.format(offset)
elif offset < 0:
offset = '{:04d}'.format(offset)
d = {
'REMOTE_ADDR': environ.get('REMOTE_ADDR', '-'),
'REMOTE_USER': environ.get('REMOTE_USER', '-'),
'REQUEST_METHOD': environ['REQUEST_METHOD'],
'REQUEST_URI': req_uri,
'HTTP_VERSION': environ.get('SERVER_PROTOCOL'),
'time': time.strftime('%d/%b/%Y:%H:%M:%S ', start) + offset,
'status': status.split(None, 1)[0],
'bytes': bytes,
'HTTP_REFERER': environ.get('HTTP_REFERER', '-'),
'HTTP_USER_AGENT': environ.get('HTTP_USER_AGENT', '-'),
}
self.logger.log(self.log_level, self.log_format.format(**d))
class WSGI_Middleware (WSGI_Object):
"""Utility class for WGSI middleware.
"""
def __init__(self, app, *args, **kwargs):
super(WSGI_Middleware, self).__init__(*args, **kwargs)
self.app = app
def _call(self, environ, start_response):
return self.app(environ, start_response)
class ExceptionApp (WSGI_Middleware):
"""Some servers (e.g. cherrypy) eat app-raised exceptions.
Work around that by logging tracebacks by hand.
"""
def _call(self, environ, start_response):
try:
return self.app(environ, start_response)
except Exception, e:
etype,value,tb = sys.exc_info()
trace = ''.join(
traceback.format_exception(etype, value, tb, None))
self.logger.log(self.log_level, trace)
raise
class HandlerErrorApp (WSGI_Middleware):
"""Catch HandlerErrors and return HTTP error pages.
"""
def _call(self, environ, start_response):
try:
return self.app(environ, start_response)
except HandlerError, e:
self.log_request(environ, status=str(e), bytes=0)
start_response('{} {}'.format(e.code, e.msg), e.headers)
return []
class BEExceptionApp (WSGI_Middleware):
"""Translate BE-specific exceptions
"""
def __init__(self, *args, **kwargs):
super(BEExceptionApp, self).__init__(*args, **kwargs)
self.http_user_error = 418
def _call(self, environ, start_response):
try:
return self.app(environ, start_response)
except libbe.storage.NotReadable as e:
raise libbe.util.wsgi.HandlerError(403, 'Read permission denied')
except libbe.storage.NotWriteable as e:
raise libbe.util.wsgi.HandlerError(403, 'Write permission denied')
except libbe.storage.InvalidID as e:
raise libbe.util.wsgi.HandlerError(
self.http_user_error, 'InvalidID {}'.format(e))
class UppercaseHeaderApp (WSGI_Middleware):
"""WSGI middleware that uppercases incoming HTTP headers.
From PEP 333, `The start_response() Callable`_ :
A reminder for server/gateway authors: HTTP
header names are case-insensitive, so be sure
to take that into consideration when examining
application-supplied headers!
.. _The start_response() Callable:
http://www.python.org/dev/peps/pep-0333/#id20
"""
def _call(self, environ, start_response):
for key,value in environ.items():
if key.startswith('HTTP_'):
uppercase = key.upper()
if uppercase != key:
environ[uppercase] = environ.pop(key)
return self.app(environ, start_response)
class AuthenticationApp (WSGI_Middleware):
"""WSGI middleware for handling user authentication.
"""
def __init__(self, realm, setting='be-auth', users=None, *args, **kwargs):
super(AuthenticationApp, self).__init__(*args, **kwargs)
self.realm = realm
self.setting = setting
self.users = users
def _call(self, environ, start_response):
environ['{}.realm'.format(self.setting)] = self.realm
try:
username = self.authenticate(environ)
environ['{}.user'.format(self.setting)] = username
environ['{}.user.name'.format(self.setting)] = self.users[username].name
return self.app(environ, start_response)
except Unauthorized, e:
return self.error(environ, start_response,
e.code, e.msg, e.headers)
def authenticate(self, environ):
"""Handle user-authentication sent in the "Authorization" header.
This function implements ``Basic`` authentication as described in
HTTP/1.0 specification [1]_ . Do not use this module unless you
are using SSL, as it transmits unencrypted passwords.
.. [1] http://www.w3.org/Protocols/HTTP/1.0/draft-ietf-http-spec.html#BasicAA
Examples
--------
>>> users = Users()
>>> users.add_user(User('Aladdin', 'Big Al', password='open sesame'))
>>> app = AuthenticationApp(app=None, realm='Dummy Realm', users=users)
>>> app.authenticate({'HTTP_AUTHORIZATION':'Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ=='})
'Aladdin'
>>> app.authenticate({'HTTP_AUTHORIZATION':'Basic AAAAAAAAAAAAAAAAAAAAAAAAAA=='})
Notes
-----
Code based on authkit/authenticate/basic.py
(c) 2005 Clark C. Evans.
Released under the MIT License:
http://www.opensource.org/licenses/mit-license.php
"""
authorization = environ.get('HTTP_AUTHORIZATION', None)
if authorization is None:
raise Unauthorized('Authorization required')
try:
authmeth,auth = authorization.split(' ', 1)
except ValueError:
return None
if 'basic' != authmeth.lower():
return None # non-basic HTTP authorization not implemented
auth = auth.strip().decode('base64')
try:
username,password = auth.split(':', 1)
except ValueError:
return None
if self.authfunc(environ, username, password):
return username
def authfunc(self, environ, username, password):
if not username in self.users:
return False
if self.users[username].valid_login(password):
if self.logger is not None:
self.logger.log(self.log_level,
'Authenticated {}'.format(self.users[username].name))
return True
return False
class WSGI_DataObject (WSGI_Object):
"""Useful WSGI utilities for handling data (POST, QUERY) and
returning responses.
"""
def __init__(self, *args, **kwargs):
super(WSGI_DataObject, self).__init__(*args, **kwargs)
# Maximum input we will accept when REQUEST_METHOD is POST
# 0 ==> unlimited input
self.maxlen = 0
def ok_response(self, environ, start_response, content,
content_type='application/octet-stream',
headers=[]):
if content is None:
start_response('200 OK', [])
return []
if type(content) is types.UnicodeType:
content = content.encode('utf-8')
for i,header in enumerate(headers):
header_name,header_value = header
if type(header_value) == types.UnicodeType:
headers[i] = (header_name, header_value.encode('ISO-8859-1'))
response = '200 OK'
content_length = len(content)
self.log_request(environ, status=response, bytes=content_length)
start_response(response, [
('Content-Type', content_type),
('Content-Length', str(content_length)),
]+headers)
if self.is_head(environ):
return []
return [content]
def query_data(self, environ):
if not environ['REQUEST_METHOD'] in ['GET', 'HEAD']:
raise HandlerError(404, 'Not Found')
return self._parse_query(environ.get('QUERY_STRING', ''))
def _parse_query(self, query):
if len(query) == 0:
return {}
data = urlparse.parse_qs(
query, keep_blank_values=True, strict_parsing=True)
for k,v in data.items():
if len(v) == 1:
data[k] = v[0]
return data
def post_data(self, environ):
if environ['REQUEST_METHOD'] != 'POST':
raise HandlerError(404, 'Not Found')
post_data = self._read_post_data(environ)
return self._parse_post(post_data)
def _parse_post(self, post):
return self._parse_query(post)
def _read_post_data(self, environ):
try:
clen = int(environ.get('CONTENT_LENGTH', '0'))
except ValueError:
clen = 0
if clen != 0:
if self.maxlen > 0 and clen > self.maxlen:
raise ValueError, 'Maximum content length exceeded'
return environ['wsgi.input'].read(clen)
return ''
def data_get_string(self, data, key, default=None, source='query'):
if not key in data or data[key] in [None, 'None']:
if default == HandlerError:
raise HandlerError(
406, 'Missing {} key {}'.format(source, key))
return default
return data[key]
def data_get_id(self, data, key='id', default=HandlerError,
source='query'):
return self.data_get_string(data, key, default, source)
def data_get_boolean(self, data, key, default=False, source='query'):
val = self.data_get_string(data, key, default, source)
if val == 'True':
return True
elif val == 'False':
return False
return val
def is_head(self, environ):
return environ['REQUEST_METHOD'] == 'HEAD'
class WSGI_AppObject (WSGI_Object):
"""Useful WSGI utilities for handling URL delegation.
"""
def __init__(self, urls=tuple(), default_handler=None, setting='be-server',
*args, **kwargs):
super(WSGI_AppObject, self).__init__(*args, **kwargs)
self.urls = [(re.compile(regexp),callback) for regexp,callback in urls]
self.default_handler = default_handler
self.setting = setting
def _call(self, environ, start_response):
path = environ.get('PATH_INFO', '').lstrip('/')
for regexp,callback in self.urls:
match = regexp.match(path)
if match is not None:
setting = '{}.url_args'.format(self.setting)
environ[setting] = match.groups()
return callback(environ, start_response)
if self.default_handler is None:
raise HandlerError(404, 'Not Found')
return self.default_handler(environ, start_response)
class AdminApp (WSGI_AppObject, WSGI_DataObject, WSGI_Middleware):
"""WSGI middleware for managing users
Changing passwords, usernames, etc.
"""
def __init__(self, users=None, setting='be-auth', *args, **kwargs):
handler = ('^admin/?', self.admin)
if 'urls' not in kwargs:
kwargs['urls'] = [handler]
else:
kwargs.urls.append(handler)
super(AdminApp, self).__init__(*args, **kwargs)
self.users = users
self.setting = setting
def admin(self, environ, start_response):
if not '{}.user'.format(self.setting) in environ:
realm = envirion.get('{}.realm'.format(self.setting))
raise Unauthenticated(realm=realm)
uname = environ.get('{}.user'.format(self.setting))
user = self.users[uname]
data = self.post_data(environ)
source = 'post'
name = self.data_get_string(
data, 'name', default=None, source=source)
if name is not None:
self.users[uname].set_name(name)
password = self.data_get_string(
data, 'password', default=None, source=source)
if password is not None:
self.users[uname].set_password(password)
self.users.save()
return self.ok_response(environ, start_response, None)
class SilentRequestHandler (wsgiref.simple_server.WSGIRequestHandler):
def log_message(self, format, *args):
pass
class ServerCommand (libbe.command.base.Command):
"""Serve something over HTTP.
Use this as a base class to build commands that serve a web interface.
"""
def __init__(self, *args, **kwargs):
super(ServerCommand, self).__init__(*args, **kwargs)
self.options.extend([
libbe.command.Option(name='port',
help='Bind server to port (%default)',
arg=libbe.command.Argument(
name='port', metavar='INT', type='int', default=8000)),
libbe.command.Option(name='host',
help='Set host string (blank for localhost, %default)',
arg=libbe.command.Argument(
name='host', metavar='HOST', default='localhost')),
libbe.command.Option(name='read-only', short_name='r',
help='Dissable operations that require writing'),
libbe.command.Option(name='notify', short_name='n',
help='Send notification emails for changes.',
arg=libbe.command.Argument(
name='notify', metavar='EMAIL-COMMAND', default=None)),
libbe.command.Option(name='ssl', short_name='s',
help='Use CherryPy to serve HTTPS (HTTP over SSL/TLS)'),
libbe.command.Option(name='auth', short_name='a',
help=('Require authentication. FILE should be a file '
'containing colon-separated '
'UNAME:USER:sha1(PASSWORD) lines, for example: '
'"jdoe:John Doe <jdoe@example.com>:'
'd99f8e5a4b02dc25f49da2ea67c0034f61779e72"'),
arg=libbe.command.Argument(
name='auth', metavar='FILE', default=None,
completion_callback=libbe.command.util.complete_path)),
])
def _run(self, **params):
self._setup_logging()
storage = self._get_storage()
if params['read-only']:
writeable = storage.writeable
storage.writeable = False
if params['auth']:
self._check_restricted_access(storage, params['auth'])
users = Users(params['auth'])
users.load()
app = self._get_app(logger=self.logger, storage=storage, **params)
if params['auth']:
app = AdminApp(app, users=users, logger=self.logger)
app = AuthenticationApp(app, realm=storage.repo,
users=users, logger=self.logger)
app = UppercaseHeaderApp(app, logger=self.logger)
server,details = self._get_server(params, app)
details['repo'] = storage.repo
try:
self._start_server(params, server, details)
except KeyboardInterrupt:
pass
self._stop_server(params, server)
if params['read-only']:
storage.writeable = writeable
def _get_app(self, logger, storage, **kwargs):
raise NotImplementedError()
def _setup_logging(self, log_level=logging.INFO):
self.logger = logging.getLogger('be-{}'.format(self.name))
self.log_level = logging.INFO
console = logging.StreamHandler(self.stdout)
console.setFormatter(logging.Formatter('%(message)s'))
self.logger.addHandler(console)
self.logger.propagate = False
if log_level is not None:
console.setLevel(log_level)
self.logger.setLevel(log_level)
def _get_server(self, params, app):
details = {
'socket-name':params['host'],
'port':params['port'],
}
app = BEExceptionApp(app, logger=self.logger)
app = HandlerErrorApp(app, logger=self.logger)
app = ExceptionApp(app, logger=self.logger)
if params['ssl'] == True:
details['protocol'] = 'HTTPS'
if cherrypy == None:
raise libbe.command.UserError(
'--ssl requires the cherrypy module')
server = cherrypy.wsgiserver.CherryPyWSGIServer(
(params['host'], params['port']), app)
#server.throw_errors = True
#server.show_tracebacks = True
private_key,certificate = _get_cert_filenames(
'be-server', logger=self.logger)
if cherrypy.wsgiserver.ssl_builtin == None:
server.ssl_module = 'builtin'
server.ssl_private_key = private_key
server.ssl_certificate = certificate
else:
server.ssl_adapter = (
cherrypy.wsgiserver.ssl_builtin.BuiltinSSLAdapter(
certificate=certificate, private_key=private_key))
else:
details['protocol'] = 'HTTP'
server = wsgiref.simple_server.make_server(
params['host'], params['port'], app,
handler_class=SilentRequestHandler)
return (server, details)
def _start_server(self, params, server, details):
self.logger.log(self.log_level,
('Serving {protocol} on {socket-name} port {port} ...\n'
'BE repository {repo}').format(**details))
if params['ssl']:
server.start()
else:
server.serve_forever()
def _stop_server(self, params, server):
self.logger.log(self.log_level, 'Clossing server')
if params['ssl'] == True:
server.stop()
else:
server.server_close()
def _long_help(self):
raise NotImplementedError()
class WSGICaller (object):
"""Call into WSGI apps programmatically
"""
def __init__(self, *args, **kwargs):
super(WSGICaller, self).__init__(*args, **kwargs)
self.default_environ = { # required by PEP 333
'REQUEST_METHOD': 'GET', # 'POST', 'HEAD'
'REMOTE_ADDR': '192.168.0.123',
'SCRIPT_NAME':'',
'PATH_INFO': '',
#'QUERY_STRING':'', # may be empty or absent
#'CONTENT_TYPE':'', # may be empty or absent
#'CONTENT_LENGTH':'', # may be empty or absent
'SERVER_NAME':'example.com',
'SERVER_PORT':'80',
'SERVER_PROTOCOL':'HTTP/1.1',
'wsgi.version':(1,0),
'wsgi.url_scheme':'http',
'wsgi.input':StringIO.StringIO(),
'wsgi.errors':StringIO.StringIO(),
'wsgi.multithread':False,
'wsgi.multiprocess':False,
'wsgi.run_once':False,
}
def getURL(self, app, path='/', method='GET', data=None,
data_dict=None, scheme='http', environ={}):
env = copy.copy(self.default_environ)
env['PATH_INFO'] = path
env['REQUEST_METHOD'] = method
env['scheme'] = scheme
if data_dict is not None:
assert data is None, (data, data_dict)
data = urllib.urlencode(data_dict)
if data is not None:
if data_dict is None:
assert method == 'POST', (method, data)
if method == 'POST':
env['CONTENT_LENGTH'] = len(data)
env['wsgi.input'] = StringIO.StringIO(data)
else:
assert method in ['GET', 'HEAD'], method
env['QUERY_STRING'] = data
for key,value in environ.items():
env[key] = value
return ''.join(app(env, self.start_response))
def start_response(self, status, response_headers, exc_info=None):
self.status = status
self.response_headers = response_headers
self.exc_info = exc_info
if libbe.TESTING:
class WSGITestCase (unittest.TestCase):
def setUp(self):
self.logstream = StringIO.StringIO()
self.logger = logging.getLogger('be-wsgi-test')
console = logging.StreamHandler(self.logstream)
console.setFormatter(logging.Formatter('%(message)s'))
self.logger.addHandler(console)
self.logger.propagate = False
console.setLevel(logging.INFO)
self.logger.setLevel(logging.INFO)
self.caller = WSGICaller()
def getURL(self, *args, **kwargs):
content = self.caller.getURL(*args, **kwargs)
self.status = self.caller.status
self.response_headers = self.caller.response_headers
self.exc_info = self.caller.exc_info
return content
class WSGI_ObjectTestCase (WSGITestCase):
def setUp(self):
WSGITestCase.setUp(self)
self.app = WSGI_Object(self.logger)
def test_error(self):
contents = self.app.error(
environ=self.caller.default_environ,
start_response=self.caller.start_response,
error=123,
message='Dummy Error',
headers=[('X-Dummy-Header','Dummy Value')])
self.failUnless(contents == ['Dummy Error'], contents)
self.failUnless(
self.caller.status == '123 Dummy Error', self.caller.status)
self.failUnless(self.caller.response_headers == [
('Content-Type','text/plain'),
('X-Dummy-Header','Dummy Value')],
self.caller.response_headers)
self.failUnless(self.caller.exc_info == None, self.caller.exc_info)
def test_log_request(self):
self.app.log_request(
environ=self.caller.default_environ, status='-1 OK', bytes=123)
log = self.logstream.getvalue()
self.failUnless(log.startswith('192.168.0.123 -'), log)
class ExceptionAppTestCase (WSGITestCase):
def setUp(self):
WSGITestCase.setUp(self)
def child_app(environ, start_response):
raise ValueError('Dummy Error')
self.app = ExceptionApp(child_app, self.logger)
def test_traceback(self):
try:
self.getURL(self.app)
except ValueError, e:
pass
log = self.logstream.getvalue()
self.failUnless(log.startswith('Traceback'), log)
self.failUnless('child_app' in log, log)
self.failUnless('ValueError: Dummy Error' in log, log)
class AdminAppTestCase (WSGITestCase):
def setUp(self):
WSGITestCase.setUp(self)
self.users = Users()
self.users.add_user(
User('Aladdin', 'Big Al', password='open sesame'))
self.users.add_user(
User('guest', 'Guest', password='guestpass'))
def child_app(environ, start_response):
pass
app = AdminApp(
app=child_app, users=self.users, logger=self.logger)
app = AuthenticationApp(
app=app, realm='Dummy Realm', users=self.users,
logger=self.logger)
self.app = UppercaseHeaderApp(app=app, logger=self.logger)
def basic_auth(self, uname, password):
"""HTTP basic authorization string"""
return 'Basic {}'.format(
'{}:{}'.format(uname, password).encode('base64'))
def test_new_name(self):
self.getURL(
self.app, '/admin/', method='POST',
data_dict={'name':'Prince Al'},
environ={'HTTP_Authorization':
self.basic_auth('Aladdin', 'open sesame')})
self.failUnless(self.status == '200 OK', self.status)
self.failUnless(self.response_headers == [],
self.response_headers)
self.failUnless(self.exc_info == None, self.exc_info)
self.failUnless(self.users['Aladdin'].name == 'Prince Al',
self.users['Aladdin'].name)
self.failUnless(self.users.changed == True,
self.users.changed)
def test_new_password(self):
self.getURL(
self.app, '/admin/', method='POST',
data_dict={'password':'New Pass'},
environ={'HTTP_Authorization':
self.basic_auth('Aladdin', 'open sesame')})
self.failUnless(self.status == '200 OK', self.status)
self.failUnless(self.response_headers == [],
self.response_headers)
self.failUnless(self.exc_info == None, self.exc_info)
self.failUnless((self.users['Aladdin'].passhash ==
self.users['Aladdin'].hash('New Pass')),
self.users['Aladdin'].passhash)
self.failUnless(self.users.changed == True,
self.users.changed)
def test_guest_name(self):
self.getURL(
self.app, '/admin/', method='POST',
data_dict={'name':'SPAM'},
environ={'HTTP_Authorization':
self.basic_auth('guest', 'guestpass')})
self.failUnless(self.status.startswith('403 '), self.status)
self.failUnless(self.response_headers == [
('Content-Type', 'text/plain')],
self.response_headers)
self.failUnless(self.exc_info == None, self.exc_info)
self.failUnless(self.users['guest'].name == 'Guest',
self.users['guest'].name)
self.failUnless(self.users.changed == False,
self.users.changed)
def test_guest_password(self):
self.getURL(
self.app, '/admin/', method='POST',
data_dict={'password':'SPAM'},
environ={'HTTP_Authorization':
self.basic_auth('guest', 'guestpass')})
self.failUnless(self.status.startswith('403 '), self.status)
self.failUnless(self.response_headers == [
('Content-Type', 'text/plain')],
self.response_headers)
self.failUnless(self.exc_info == None, self.exc_info)
self.failUnless(self.users['guest'].name == 'Guest',
self.users['guest'].name)
self.failUnless(self.users.changed == False,
self.users.changed)
unitsuite =unittest.TestLoader().loadTestsFromModule(sys.modules[__name__])
suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])
# The following certificate-creation code is adapted From pyOpenSSL's
# examples.
def _get_cert_filenames(server_name, autogenerate=True, logger=None):
"""
Generate private key and certification filenames.
get_cert_filenames(server_name) -> (pkey_filename, cert_filename)
"""
pkey_file = '{}.pkey'.format(server_name)
cert_file = '{}.cert'.format(server_name)
if autogenerate:
for file in [pkey_file, cert_file]:
if not os.path.exists(file):
_make_certs(server_name, logger)
return (pkey_file, cert_file)
def _create_key_pair(type, bits):
"""Create a public/private key pair.
Returns the public/private key pair in a PKey object.
Parameters
----------
type : TYPE_RSA or TYPE_DSA
Key type.
bits : int
Number of bits to use in the key.
"""
pkey = OpenSSL.crypto.PKey()
pkey.generate_key(type, bits)
return pkey
def _create_cert_request(pkey, digest="md5", **name):
"""Create a certificate request.
Returns the certificate request in an X509Req object.
Parameters
----------
pkey : PKey
The key to associate with the request.
digest : "md5" or ?
Digestion method to use for signing, default is "md5",
`**name` :
The name of the subject of the request, possible.
Arguments are:
============ ========================
C Country name
ST State or province name
L Locality name
O Organization name
OU Organizational unit name
CN Common name
emailAddress E-mail address
============ ========================
"""
req = OpenSSL.crypto.X509Req()
subj = req.get_subject()
for (key,value) in name.items():
setattr(subj, key, value)
req.set_pubkey(pkey)
req.sign(pkey, digest)
return req
def _create_certificate(req, (issuerCert, issuerKey), serial,
(notBefore, notAfter), digest='md5'):
"""Generate a certificate given a certificate request.
Returns the signed certificate in an X509 object.
Parameters
----------
req :
Certificate reqeust to use
issuerCert :
The certificate of the issuer
issuerKey :
The private key of the issuer
serial :
Serial number for the certificate
notBefore :
Timestamp (relative to now) when the certificate
starts being valid
notAfter :
Timestamp (relative to now) when the certificate
stops being valid
digest :
Digest method to use for signing, default is md5
"""
cert = OpenSSL.crypto.X509()
cert.set_serial_number(serial)
cert.gmtime_adj_notBefore(notBefore)
cert.gmtime_adj_notAfter(notAfter)
cert.set_issuer(issuerCert.get_subject())
cert.set_subject(req.get_subject())
cert.set_pubkey(req.get_pubkey())
cert.sign(issuerKey, digest)
return cert
def _make_certs(server_name, logger=None) :
"""Generate private key and certification files.
`mk_certs(server_name) -> (pkey_filename, cert_filename)`
"""
if OpenSSL == None:
raise libbe.command.UserError(
'SSL certificate generation requires the OpenSSL module')
pkey_file,cert_file = get_cert_filenames(
server_name, autogenerate=False)
if logger != None:
logger.log(logger._server_level,
'Generating certificates', pkey_file, cert_file)
cakey = _create_key_pair(OpenSSL.crypto.TYPE_RSA, 1024)
careq = _create_cert_request(cakey, CN='Certificate Authority')
cacert = _create_certificate(
careq, (careq, cakey), 0, (0, 60*60*24*365*5)) # five years
open(pkey_file, 'w').write(OpenSSL.crypto.dump_privatekey(
OpenSSL.crypto.FILETYPE_PEM, cakey))
open(cert_file, 'w').write(OpenSSL.crypto.dump_certificate(
OpenSSL.crypto.FILETYPE_PEM, cacert))
