', re.DOTALL) 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): """Read template named `name`. Templates are looked up in the local `templates` dir. In the result any 'PART_HEADER' and 'PART_FOOTER' parts are replaced by the respective templates. Returns the HTML template. """ path = os.path.join(template_dir, name) if os.path.isfile(path): html = open(path, 'r').read() html = html.replace('PART_HEADER', SHARED_HEADER) html = html.replace('PART_FOOTER', SHARED_FOOTER) return html return None def update_url(url, params_dict): """Update query params of an url. The `url` is modified to have the query parameters set to keys/values in `params_dict`, preserving any different existing keys/values and overwriting any existing keys/values that are also in `params_dict`. Thus, ``'http://sample?a=1', dict(b='1')`` will result in ``'http://sample?a=1&b=1`` and similar. """ parts = [x for x in urlparse(url)] old_params = dict(parse_qsl(parts[4])) old_params.update(params_dict) query_string = urlencode(old_params) parts[4] = query_string return urlunparse(parts) def login_redirect_service(db, user, 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(user, service, sso) db.add(st) service = update_url(service, dict(ticket=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 = set_message(msg, html) resp = Response(html) if not sso: resp = set_session_cookie(db, resp) return resp def set_message(msg, html): """Insert a message box in html template. If the message is empty, not only the string `MSG_TEXT` is removed from `html`, but also any encapsulating ``

`` tag with id ``msg``. This makes it possible to give message boxes certain additional styles that will not show up if there is no message to display. """ if not msg: if not '

' % ( 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 = set_message(msg, html) 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') html = set_message('', html) if url is not None: html = self._get_template('logout_url.html') html = html.replace('URL_HREF', url) html = set_message('', html) 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)