source: main/waeup.cas/trunk/waeup/cas/server.py @ 10476

"""A WSGI app for serving CAS.
"""
import datetime
import os
import random
import time
from webob import exc, Response
from webob.dec import wsgify
from waeup.cas.authenticators import get_authenticator
from waeup.cas.db import (
    DB, DBSessionContext, LoginTicket, ServiceTicket, TicketGrantingCookie)

template_dir = os.path.join(os.path.dirname(__file__), 'templates')

RANDOM = random.SystemRandom(os.urandom(1024))

#: The chars allowed by protocol specification for tickets and cookie
#: values.
ALPHABET = ('abcdefghijklmnopqrstuvwxyz'
            'ABCDEFGHIJKLMNOPQRSTUVWXYZ'
            '01234567789-')


def get_random_string(length):
    """Get a random string of length `length`.

    The returned string should be hard to guess but is not
    neccessarily unique.
    """
    return ''.join([RANDOM.choice(ALPHABET) for x in range(length)])


def get_unique_string():
    """Get a unique string based on current time.

    The returned string contains only chars from `ALPHABET`.

    We try to be unique by using a timestamp in high resolution, so
    that even tickets created shortly after another should differ. On
    very fast machines, however, this might be not enough (currently
    we use 16 decimal places).

    This is fast because we don't have to fetch foreign data sources
    nor have to do database lookups.

    The returned string will be unique but it won't be hard to guess
    for people able to read a clock.
    """
    return ('%.16f' % time.time()).replace('.', '-')


def create_service_ticket(user, service=None, sso=True):
    """Get a service ticket.

    Ticket length will be 32 chars, randomly picked from `ALPHABET`.
    """
    t_id = 'ST-' + get_random_string(29)
    return ServiceTicket(t_id, user, service, sso)


def check_service_ticket(db, ticket, service, renew=False):
    """Check whether (`ticket`, `service`) represents a valid service
    ticket in `db`.

    Returns a database set or ``None``.
    """
    if None in (ticket, service):
        return None
    ticket, service = str(ticket), str(service)
    q = db.query(ServiceTicket).filter(
        ServiceTicket.ticket == ticket).filter(
        ServiceTicket.service == service).first()
    if renew and q.sso:
        return None
    return q


def create_login_ticket():
    """Create a unique login ticket.

    Login tickets are required to be unique (but not neccessarily hard
    to guess), according to protocol specification.
    """
    t_id = 'LT-%s' % get_unique_string()
    return LoginTicket(t_id)


def check_login_ticket(db, lt_string):
    """Check whether `lt_string` represents a valid login ticket in `db`.
    """
    if lt_string is None:
        return False
    q = db.query(LoginTicket).filter(LoginTicket.ticket == str(lt_string))
    result = [x for x in q]
    if result:
        db.delete(result[0])
    return len(result) > 0


def create_tgc_value():
    """Get a ticket granting cookie value.
    """
    value = 'TGC-' + get_random_string(128)
    return TicketGrantingCookie(value)


def set_session_cookie(db, response):
    """Create a session cookie (ticket granting cookie) on `response`.

    The `db` database is used to make the created cookie value
    persistent.
    """
    tgc = create_tgc_value()
    db.add(tgc)
    response.set_cookie(
        'cas-tgc', tgc.value, path='/', secure=True, httponly=True)
    return response


def delete_session_cookie(db, response, old_value=None):
    """Delete session cookie.

    Sets cookie with expiration date in past and deletes respective
    entry from database.
    """
    if old_value is not None:
        # delete old tgc from db
        q = db.query(TicketGrantingCookie).filter(
            TicketGrantingCookie.value == old_value)
        result = list(q)
        if len(result) == 1:
            db.delete(result[0])
    response.set_cookie(
        'cas-tgc', '', path='/', secure=True, httponly=True,
        expires=datetime.datetime(1970, 1, 1, 0, 0, 0))
    return response


def check_session_cookie(db, cookie_value):
    """Check whether `cookie_value` represents a valid ticket granting
    ticket in `db`.

    `cookie_value` is a string representing a ticket granting ticket
    maybe stored in `db`.

    If a respective cookie can be found, a
    :class:`waeup.cas.db.TicketGrantingCookie` is returend. Else
    ``None`` is returned.
    """
    if cookie_value is None:
        return None
    try:
        # turn value into unicode (py2.x) / str (py3.x)
        cookie_value = cookie_value.decode('utf-8')
    except AttributeError:                         # pragma: no cover
        pass
    q = db.query(TicketGrantingCookie).filter(
        TicketGrantingCookie.value == cookie_value)
    result = [x for x in q]
    if len(result):
        return result[0]
    return None


def get_template(name):
    path = os.path.join(template_dir, name)
    if os.path.isfile(path):
        return open(path, 'r').read()
    return None


def login_redirect_service(db, service, sso=True, create_ticket=True,
                           warn=False):
    """Return a response redirecting to a service via HTTP 303 See Other.
    """
    if create_ticket:
        st = create_service_ticket(service, sso)
        db.add(st)
        service = '%s?ticket=%s' % (service, st.ticket)
    html = get_template('login_service_redirect.html')
    if warn:
        html = get_template('login_service_confirm.html')
    html = html.replace('SERVICE_URL', service)
    resp = exc.HTTPSeeOther(location=service)
    if warn:
        resp = Response()
    # try to forbid caching of any type
    resp.cache_control = 'no-store'
    resp.pragma = 'no-cache'
    # some arbitrary date in the past
    resp.expires = 'Thu, 01 Dec 1994 16:00:00 GMT'
    resp.text = html
    if not sso:
        resp = set_session_cookie(db, resp)
    return resp


def login_success_no_service(db, msg='', sso=False):
    """Show logged-in screen after successful auth.

    `sso` must be a boolean indicating whether login happened via
    credentials (``False``) or via cookie (``True``).

    Returns a response.
    """
    # show logged-in screen
    html = get_template('login_successful.html')
    html = html.replace('MSG_TEXT', msg)
    resp = Response(html)
    if not sso:
        resp = set_session_cookie(db, resp)
    return resp


class CASServer(object):
    """A WSGI CAS server.

    This CAS server stores credential data (tickets, etc.) in a local
    sqlite3 database file.

    `db_path` -
       The filesystem path to the database to use. If none is given
       CAS server will create a new one in some new, temporary
       directory. Please note that credentials will be lost after a
       CAS server restart.

       If the path is given and the file exists already, it will be
       used.

       If the database file does not exist, it will be created.
    """
    def __init__(self, db='sqlite:///:memory:', auth=None):
        self.db_connection_string = db
        self.db = DB(self.db_connection_string)
        self.auth = auth

    @wsgify
    def __call__(self, req):
        with DBSessionContext():
            if req.path in ['/login', '/validate', '/logout']:
                return getattr(self, req.path[1:])(req)
        return exc.HTTPNotFound()

    def _get_template(self, name):
        path = os.path.join(template_dir, name)
        if os.path.isfile(path):
            return open(path, 'r').read()
        return None

    def login(self, req):
        service = req.POST.get('service', req.GET.get('service', None))
        renew = req.POST.get('renew', req.GET.get('renew', None))
        warn = req.POST.get('warn', req.GET.get('warn', False))
        gateway = req.POST.get('gateway', req.GET.get('gateway', None))
        if renew is not None and gateway is not None:
            gateway = None
        service_field = ''
        msg = ''
        username = req.POST.get('username', None)
        password = req.POST.get('password', None)
        valid_lt = check_login_ticket(self.db, req.POST.get('lt'))
        tgc = check_session_cookie(self.db, req.cookies.get('cas-tgc', None))
        if gateway and (not tgc) and service:
            return login_redirect_service(
                self.db, service, sso=True, create_ticket=False)
        if tgc and (renew is None):
            if service:
                return login_redirect_service(
                    self.db, service, sso=True, warn=warn)
            else:
                return login_success_no_service(
                    self.db, 'You logged in already.', True)
        if username and password and valid_lt:
            # act as credentials acceptor
            cred_ok, reason = self.auth.check_credentials(
                username, password)
            if cred_ok:
                if service is None:
                    # show logged-in screen
                    return login_success_no_service(self.db, msg, False)
                else:
                    # safely redirect to service given
                    return login_redirect_service(
                        self.db, service, sso=False, warn=warn)
            else:
                # login failed
                msg = '<i>Login failed</i><br />Reason: %s' % reason
        if service is not None:
            service_field = (
                '<input type="hidden" name="service" value="%s" />' % (
                    service)
                )
        lt = create_login_ticket()
        self.db.add(lt)
        html = self._get_template('login.html')
        html = html.replace('LT_VALUE', lt.ticket)
        html = html.replace('SERVICE_FIELD_VALUE', service_field)
        html = html.replace('MSG_TEXT', msg)
        return Response(html)

    def validate(self, req):
        service = req.POST.get('service', req.GET.get('service', None))
        ticket = req.POST.get('ticket', req.GET.get('ticket', None))
        renew = req.POST.get('renew', req.GET.get('renew', None))
        renew = renew is not None
        st = check_service_ticket(self.db, ticket, service, renew)
        if st is not None:
            return Response('yes' + chr(0x0a) + st.user + chr(0x0a))
        return Response('no' + chr(0x0a) + chr(0x0a))

    def logout(self, req):
        url = req.GET.get('url', req.POST.get('url', None))
        old_val = req.cookies.get('cas-tgc', None)
        html = self._get_template('logout.html')
        if url is not None:
            html = self._get_template('logout_url.html')
            html = html.replace('URL_HREF', url)
        resp = Response(html)
        delete_session_cookie(self.db, resp, old_val)
        return resp


cas_server = CASServer


def make_cas_server(global_conf, **local_conf):
    local_conf = get_authenticator(local_conf)
    return CASServer(**local_conf)