ipsilon/providers/persona/auth.py

   1 # Copyright (C) 2014  Ipsilon project Contributors, for licensee see COPYING
   2
   3 from ipsilon.providers.common import ProviderPageBase
   4 from ipsilon.util.user import UserSession
   5
   6 import base64
   7 import cherrypy
   8 import time
   9 import json
  10 import M2Crypto
  11
  12
  13 class AuthenticateRequest(ProviderPageBase):
  14
  15     def __init__(self, *args, **kwargs):
  16         super(AuthenticateRequest, self).__init__(*args, **kwargs)
  17         self.trans = None
  18
  19     def _preop(self, *args, **kwargs):
  20         self.trans = self.get_valid_transaction('persona', **kwargs)
  21
  22     def pre_GET(self, *args, **kwargs):
  23         self._preop(*args, **kwargs)
  24
  25     def pre_POST(self, *args, **kwargs):
  26         self._preop(*args, **kwargs)
  27
  28
  29 class Sign(AuthenticateRequest):
  30
  31     def _base64_url_decode(self, inp):
  32         inp += '=' * (4 - (len(inp) % 4))
  33         return base64.urlsafe_b64decode(inp)
  34
  35     def _base64_url_encode(self, inp):
  36         return base64.urlsafe_b64encode(inp).replace('=', '')
  37
  38     def _persona_sign(self, email, publicKey, certDuration):
  39         self.debug('Signing for %s with duration of %s' % (email,
  40                                                            certDuration))
  41         header = {'alg': 'RS256'}
  42         header = json.dumps(header)
  43         header = self._base64_url_encode(header)
  44
  45         claim = {}
  46         # Valid from 10 seconds before now to account for clock skew
  47         claim['iat'] = 1000 * int(time.time() - 10)
  48         # Validity of at most 24 hours
  49         claim['exp'] = 1000 * int(time.time() +
  50                                   min(certDuration, 24 * 60 * 60))
  51
  52         claim['iss'] = self.cfg.issuer_domain
  53         claim['public-key'] = json.loads(publicKey)
  54         claim['principal'] = {'email': email}
  55
  56         claim = json.dumps(claim)
  57         claim = self._base64_url_encode(claim)
  58
  59         certificate = '%s.%s' % (header, claim)
  60         digest = M2Crypto.EVP.MessageDigest('sha256')
  61         digest.update(certificate)
  62         signature = self.cfg.key.sign(digest.digest(), 'sha256')
  63         signature = self._base64_url_encode(signature)
  64         signed_certificate = '%s.%s' % (certificate, signature)
  65
  66         return signed_certificate
  67
  68     def _willing_to_sign(self, email, username):
  69         for domain in self.cfg.allowed_domains:
  70             if email == ('%s@%s' % (username, domain)):
  71                 return True
  72         return False
  73
  74     def POST(self, *args, **kwargs):
  75         if 'email' not in kwargs or 'publicKey' not in kwargs \
  76                 or 'certDuration' not in kwargs or '@' not in kwargs['email']:
  77             cherrypy.response.status = 400
  78             raise Exception('Invalid request: %s' % kwargs)
  79
  80         us = UserSession()
  81         user = us.get_user()
  82
  83         if user.is_anonymous:
  84             raise cherrypy.HTTPError(401, 'Not signed in')
  85
  86         if not self._willing_to_sign(kwargs['email'], user.name):
  87             self.log('Not willing to sign for %s, logged in as %s' % (
  88                 kwargs['email'], user.name))
  89             raise cherrypy.HTTPError(403, 'Incorrect user')
  90
  91         return self._persona_sign(kwargs['email'], kwargs['publicKey'],
  92                                   kwargs['certDuration'])
  93
  94
  95 class SignInResult(AuthenticateRequest):
  96     def GET(self, *args, **kwargs):
  97         user = UserSession().get_user()
  98
  99         return self._template('persona/signin_result.html',
 100                               loggedin=not user.is_anonymous)
 101
 102
 103 class SignIn(AuthenticateRequest):
 104     def __init__(self, *args, **kwargs):
 105         super(SignIn, self).__init__(*args, **kwargs)
 106         self.result = SignInResult(*args, **kwargs)
 107         self.trans = None
 108
 109     def GET(self, *args, **kwargs):
 110         username = None
 111         domain = None
 112         if 'email' in kwargs:
 113             if '@' in kwargs['email']:
 114                 username, domain = kwargs['email'].split('@', 2)
 115                 self.debug('Persona SignIn requested for: %s@%s' % (username,
 116                                                                     domain))
 117
 118         returl = '%s/persona/SignIn/result?%s' % (
 119             self.basepath, self.trans.get_GET_arg())
 120         data = {'login_return': returl,
 121                 'login_target': 'Persona',
 122                 'login_username': username}
 123         self.trans.store(data)
 124         redirect = '%s/login?%s' % (self.basepath,
 125                                     self.trans.get_GET_arg())
 126         self.debug('Redirecting: %s' % redirect)
 127         raise cherrypy.HTTPRedirect(redirect)
 128
 129
 130 class Persona(AuthenticateRequest):
 131
 132     def __init__(self, *args, **kwargs):
 133         super(Persona, self).__init__(*args, **kwargs)
 134         self.Sign = Sign(*args, **kwargs)
 135         self.SignIn = SignIn(*args, **kwargs)
 136         self.trans = None
 137
 138     def GET(self, *args, **kwargs):
 139         user = UserSession().get_user()
 140         return self._template('persona/provisioning.html',
 141                               loggedin=not user.is_anonymous)