from codecs import utf_8_decode
from codecs import utf_8_encode
import datetime
import re
import time
from paste.auth import auth_tkt
from paste.request import get_cookies
from zope.interface import implements
from pyramid.interfaces import IAuthenticationPolicy
from pyramid.request import add_global_response_headers
from pyramid.security import Authenticated
from pyramid.security import Everyone
VALID_TOKEN = re.compile(r"^[A-Za-z][A-Za-z0-9+_-]*$")
class CallbackAuthenticationPolicy(object):
""" Abstract class """
def authenticated_userid(self, request):
userid = self.unauthenticated_userid(request)
if userid is None:
return None
if self.callback is None:
return userid
if self.callback(userid, request) is not None: # is not None!
return userid
def effective_principals(self, request):
effective_principals = [Everyone]
userid = self.unauthenticated_userid(request)
if userid is None:
return effective_principals
if self.callback is None:
groups = []
else:
groups = self.callback(userid, request)
if groups is None: # is None!
return effective_principals
effective_principals.append(Authenticated)
effective_principals.append(userid)
effective_principals.extend(groups)
return effective_principals
[docs]class RepozeWho1AuthenticationPolicy(CallbackAuthenticationPolicy):
""" A :app:`Pyramid` :term:`authentication policy` which
obtains data from the :mod:`repoze.who` 1.X WSGI 'API' (the
``repoze.who.identity`` key in the WSGI environment).
Constructor Arguments
``identifier_name``
Default: ``auth_tkt``. The :mod:`repoze.who` plugin name that
performs remember/forget. Optional.
``callback``
Default: ``None``. A callback passed the :mod:`repoze.who` identity
and the :term:`request`, expected to return ``None`` if the user
represented by the identity doesn't exist or a sequence of principal
identifiers (possibly empty) representing groups if the user does
exist. If ``callback`` is None, the userid will be assumed to exist
with no group principals.
Objects of this class implement the interface described by
:class:`pyramid.interfaces.IAuthenticationPolicy`.
"""
implements(IAuthenticationPolicy)
def __init__(self, identifier_name='auth_tkt', callback=None):
self.identifier_name = identifier_name
self.callback = callback
def _get_identity(self, request):
return request.environ.get('repoze.who.identity')
def _get_identifier(self, request):
plugins = request.environ.get('repoze.who.plugins')
if plugins is None:
return None
identifier = plugins[self.identifier_name]
return identifier
def authenticated_userid(self, request):
identity = self._get_identity(request)
if identity is None:
return None
if self.callback is None:
return identity['repoze.who.userid']
if self.callback(identity, request) is not None: # is not None!
return identity['repoze.who.userid']
def unauthenticated_userid(self, request):
identity = self._get_identity(request)
if identity is None:
return None
return identity['repoze.who.userid']
def effective_principals(self, request):
effective_principals = [Everyone]
identity = self._get_identity(request)
if identity is None:
return effective_principals
if self.callback is None:
groups = []
else:
groups = self.callback(identity, request)
if groups is None: # is None!
return effective_principals
userid = identity['repoze.who.userid']
effective_principals.append(Authenticated)
effective_principals.append(userid)
effective_principals.extend(groups)
return effective_principals
def remember(self, request, principal, **kw):
identifier = self._get_identifier(request)
if identifier is None:
return []
environ = request.environ
identity = {'repoze.who.userid':principal}
return identifier.remember(environ, identity)
def forget(self, request):
identifier = self._get_identifier(request)
if identifier is None:
return []
identity = self._get_identity(request)
return identifier.forget(request.environ, identity)
[docs]class RemoteUserAuthenticationPolicy(CallbackAuthenticationPolicy):
""" A :app:`Pyramid` :term:`authentication policy` which
obtains data from the ``REMOTE_USER`` WSGI environment variable.
Constructor Arguments
``environ_key``
Default: ``REMOTE_USER``. The key in the WSGI environ which
provides the userid.
``callback``
Default: ``None``. A callback passed the userid and the request,
expected to return None if the userid doesn't exist or a sequence of
principal identifiers (possibly empty) representing groups if the
user does exist. If ``callback`` is None, the userid will be assumed
to exist with no group principals.
Objects of this class implement the interface described by
:class:`pyramid.interfaces.IAuthenticationPolicy`.
"""
implements(IAuthenticationPolicy)
def __init__(self, environ_key='REMOTE_USER', callback=None):
self.environ_key = environ_key
self.callback = callback
def unauthenticated_userid(self, request):
return request.environ.get(self.environ_key)
def remember(self, request, principal, **kw):
return []
def forget(self, request):
return []
[docs]class AuthTktAuthenticationPolicy(CallbackAuthenticationPolicy):
""" A :app:`Pyramid` :term:`authentication policy` which
obtains data from an :class:`paste.auth.auth_tkt` cookie.
Constructor Arguments
``secret``
The secret (a string) used for auth_tkt cookie signing.
Required.
``callback``
Default: ``None``. A callback passed the userid and the
request, expected to return ``None`` if the userid doesn't
exist or a sequence of principal identifiers (possibly empty) if
the user does exist. If ``callback`` is ``None``, the userid
will be assumed to exist with no principals. Optional.
``cookie_name``
Default: ``auth_tkt``. The cookie name used
(string). Optional.
``secure``
Default: ``False``. Only send the cookie back over a secure
conn. Optional.
``include_ip``
Default: ``False``. Make the requesting IP address part of
the authentication data in the cookie. Optional.
``timeout``
Default: ``None``. Maximum number of seconds which a newly
issued ticket will be considered valid. After this amount of
time, the ticket will expire (effectively logging the user
out). If this value is ``None``, the ticket never expires.
Optional.
``reissue_time``
Default: ``None``. If this parameter is set, it represents the number
of seconds that must pass before an authentication token cookie is
automatically reissued as the result of a request which requires
authentication. The duration is measured as the number of seconds
since the last auth_tkt cookie was issued and 'now'. If this value is
``0``, a new ticket cookie will be reissued on every request which
requires authentication.
A good rule of thumb: if you want auto-expired cookies based on
inactivity: set the ``timeout`` value to 1200 (20 mins) and set the
``reissue_time`` value to perhaps a tenth of the ``timeout`` value
(120 or 2 mins). It's nonsensical to set the ``timeout`` value lower
than the ``reissue_time`` value, as the ticket will never be reissued
if so. However, such a configuration is not explicitly prevented.
Optional.
``max_age``
Default: ``None``. The max age of the auth_tkt cookie, in
seconds. This differs from ``timeout`` inasmuch as ``timeout``
represents the lifetime of the ticket contained in the cookie,
while this value represents the lifetime of the cookie itself.
When this value is set, the cookie's ``Max-Age`` and
``Expires`` settings will be set, allowing the auth_tkt cookie
to last between browser sessions. It is typically nonsensical
to set this to a value that is lower than ``timeout`` or
``reissue_time``, although it is not explicitly prevented.
Optional.
``path``
Default: ``/``. The path for which the auth_tkt cookie is valid.
May be desirable if the application only serves part of a domain.
Optional.
``http_only``
Default: ``False``. Hide cookie from JavaScript by setting the
HttpOnly flag. Not honored by all browsers.
Optional.
``wild_domain``
Default: ``True``. An auth_tkt cookie will be generated for the
wildcard domain.
Optional.
Objects of this class implement the interface described by
:class:`pyramid.interfaces.IAuthenticationPolicy`.
"""
implements(IAuthenticationPolicy)
def __init__(self,
secret,
callback=None,
cookie_name='auth_tkt',
secure=False,
include_ip=False,
timeout=None,
reissue_time=None,
max_age=None,
path="/",
http_only=False,
wild_domain=True,
):
self.cookie = AuthTktCookieHelper(
secret,
cookie_name=cookie_name,
secure=secure,
include_ip=include_ip,
timeout=timeout,
reissue_time=reissue_time,
max_age=max_age,
http_only=http_only,
path=path,
wild_domain=wild_domain,
)
self.callback = callback
def unauthenticated_userid(self, request):
result = self.cookie.identify(request)
if result:
return result['userid']
def remember(self, request, principal, **kw):
""" Accepts the following kw args: ``max_age``."""
return self.cookie.remember(request, principal, **kw)
def forget(self, request):
return self.cookie.forget(request)
def b64encode(v):
return v.encode('base64').strip().replace('\n', '')
def b64decode(v):
return v.decode('base64')
EXPIRE = object()
[docs]class AuthTktCookieHelper(object):
"""
A helper class for use in third-party authentication policy
implementations. See
:class:`pyramid.authentication.AuthTktAuthenticationPolicy` for the
meanings of the constructor arguments.
"""
auth_tkt = auth_tkt # for tests
now = None # for tests
userid_type_decoders = {
'int':int,
'unicode':lambda x: utf_8_decode(x)[0], # bw compat for old cookies
'b64unicode': lambda x: utf_8_decode(b64decode(x))[0],
'b64str': lambda x: b64decode(x),
}
userid_type_encoders = {
int: ('int', str),
long: ('int', str),
unicode: ('b64unicode', lambda x: b64encode(utf_8_encode(x)[0])),
str: ('b64str', lambda x: b64encode(x)),
}
def __init__(self, secret, cookie_name='auth_tkt', secure=False,
include_ip=False, timeout=None, reissue_time=None,
max_age=None, http_only=False, path="/", wild_domain=True):
self.secret = secret
self.cookie_name = cookie_name
self.include_ip = include_ip
self.secure = secure
self.timeout = timeout
self.reissue_time = reissue_time
self.max_age = max_age
self.http_only = http_only
self.path = path
self.wild_domain = wild_domain
static_flags = []
if self.secure:
static_flags.append('; Secure')
if self.http_only:
static_flags.append('; HttpOnly')
self.static_flags = "".join(static_flags)
def _get_cookies(self, environ, value, max_age=None):
if max_age is EXPIRE:
max_age = "; Max-Age=0; Expires=Wed, 31-Dec-97 23:59:59 GMT"
elif max_age is not None:
later = datetime.datetime.utcnow() + datetime.timedelta(
seconds=int(max_age))
# Wdy, DD-Mon-YY HH:MM:SS GMT
expires = later.strftime('%a, %d %b %Y %H:%M:%S GMT')
# the Expires header is *required* at least for IE7 (IE7 does
# not respect Max-Age)
max_age = "; Max-Age=%s; Expires=%s" % (max_age, expires)
else:
max_age = ''
cur_domain = environ.get('HTTP_HOST', environ.get('SERVER_NAME'))
# While Chrome, IE, and Firefox can cope, Opera (at least) cannot
# cope with a port number in the cookie domain when the URL it
# receives the cookie from does not also have that port number in it
# (e.g via a proxy). In the meantime, HTTP_HOST is sent with port
# number, and neither Firefox nor Chrome do anything with the
# information when it's provided in a cookie domain except strip it
# out. So we strip out any port number from the cookie domain
# aggressively to avoid problems. See also
# https://github.com/Pylons/pyramid/issues/131
if ':' in cur_domain:
cur_domain = cur_domain.split(':', 1)[0]
cookies = [
('Set-Cookie', '%s="%s"; Path=%s%s%s' % (
self.cookie_name, value, self.path, max_age, self.static_flags)),
('Set-Cookie', '%s="%s"; Path=%s; Domain=%s%s%s' % (
self.cookie_name, value, self.path, cur_domain, max_age,
self.static_flags)),
]
if self.wild_domain:
wild_domain = '.' + cur_domain
cookies.append(('Set-Cookie', '%s="%s"; Path=%s; Domain=%s%s%s' % (
self.cookie_name, value, self.path, wild_domain, max_age,
self.static_flags)))
return cookies
[docs] def identify(self, request):
""" Return a dictionary with authentication information, or ``None``
if no valid auth_tkt is attached to ``request``"""
environ = request.environ
cookies = get_cookies(environ)
cookie = cookies.get(self.cookie_name)
if cookie is None or not cookie.value:
return None
if self.include_ip:
remote_addr = environ['REMOTE_ADDR']
else:
remote_addr = '0.0.0.0'
try:
timestamp, userid, tokens, user_data = self.auth_tkt.parse_ticket(
self.secret, cookie.value, remote_addr)
except self.auth_tkt.BadTicket:
return None
now = self.now # service tests
if now is None:
now = time.time()
if self.timeout and ( (timestamp + self.timeout) < now ):
# the auth_tkt data has expired
return None
userid_typename = 'userid_type:'
user_data_info = user_data.split('|')
for datum in filter(None, user_data_info):
if datum.startswith(userid_typename):
userid_type = datum[len(userid_typename):]
decoder = self.userid_type_decoders.get(userid_type)
if decoder:
userid = decoder(userid)
reissue = self.reissue_time is not None
if reissue and not hasattr(request, '_authtkt_reissued'):
if ( (now - timestamp) > self.reissue_time ):
# work around https://github.com/Pylons/pyramid/issues#issue/108
tokens = filter(None, tokens)
headers = self.remember(request, userid, max_age=self.max_age,
tokens=tokens)
add_global_response_headers(request, headers)
request._authtkt_reissued = True
environ['REMOTE_USER_TOKENS'] = tokens
environ['REMOTE_USER_DATA'] = user_data
environ['AUTH_TYPE'] = 'cookie'
identity = {}
identity['timestamp'] = timestamp
identity['userid'] = userid
identity['tokens'] = tokens
identity['userdata'] = user_data
return identity
[docs] def forget(self, request):
""" Return a set of expires Set-Cookie headers, which will destroy
any existing auth_tkt cookie when attached to a response"""
environ = request.environ
return self._get_cookies(environ, '', max_age=EXPIRE)
[docs] def remember(self, request, userid, max_age=None, tokens=()):
""" Return a set of Set-Cookie headers; when set into a response,
these headers will represent a valid authentication ticket.
``max_age``
The max age of the auth_tkt cookie, in seconds. When this value is
set, the cookie's ``Max-Age`` and ``Expires`` settings will be set,
allowing the auth_tkt cookie to last between browser sessions. If
this value is ``None``, the ``max_age`` value provided to the
helper itself will be used as the ``max_age`` value. Default:
``None``.
``tokens``
A sequence of strings that will be placed into the auth_tkt tokens
field. Each string in the sequence must be of the Python ``str``
type and must match the regex ``^[A-Za-z][A-Za-z0-9+_-]*$``.
Tokens are available in the returned identity when an auth_tkt is
found in the request and unpacked. Default: ``()``.
"""
if max_age is None:
max_age = self.max_age
environ = request.environ
if self.include_ip:
remote_addr = environ['REMOTE_ADDR']
else:
remote_addr = '0.0.0.0'
user_data = ''
encoding_data = self.userid_type_encoders.get(type(userid))
if encoding_data:
encoding, encoder = encoding_data
userid = encoder(userid)
user_data = 'userid_type:%s' % encoding
for token in tokens:
if not (isinstance(token, str) and VALID_TOKEN.match(token)):
raise ValueError("Invalid token %r" % (token,))
ticket = self.auth_tkt.AuthTicket(
self.secret,
userid,
remote_addr,
tokens=tokens,
user_data=user_data,
cookie_name=self.cookie_name,
secure=self.secure)
cookie_value = ticket.cookie_value()
return self._get_cookies(environ, cookie_value, max_age)
[docs]class SessionAuthenticationPolicy(CallbackAuthenticationPolicy):
""" A :app:`Pyramid` authentication policy which gets its data from the
configured :term:`session`. For this authentication policy to work, you
will have to follow the instructions in the :ref:`sessions_chapter` to
configure a :term:`session factory`.
Constructor Arguments
``prefix``
A prefix used when storing the authentication parameters in the
session. Defaults to 'auth.'. Optional.
``callback``
Default: ``None``. A callback passed the userid and the
request, expected to return ``None`` if the userid doesn't
exist or a sequence of principal identifiers (possibly empty) if
the user does exist. If ``callback`` is ``None``, the userid
will be assumed to exist with no principals. Optional.
"""
implements(IAuthenticationPolicy)
def __init__(self, prefix='auth.', callback=None):
self.callback = callback
self.prefix = prefix or ''
self.userid_key = prefix + 'userid'
def remember(self, request, principal, **kw):
""" Store a principal in the session."""
request.session[self.userid_key] = principal
return []
def forget(self, request):
""" Remove the stored principal from the session."""
if self.userid_key in request.session:
del request.session[self.userid_key]
return []
def unauthenticated_userid(self, request):
return request.session.get(self.userid_key)
