tests/blobs/openid_app.py

   1 # Copyright (C) 2015 Ipsilon project Contributors, for license see COPYING
   2
   3 import sys
   4 sys.stdout = sys.stderr
   5
   6 import cherrypy
   7 import os
   8 import pwd
   9
  10 from openid.consumer import consumer
  11 from openid.extensions import sreg, ax
  12 from openid_teams import teams
  13
  14
  15 class OpenIDApp(object):
  16     def index(self, extensions):
  17         self.extensions = extensions == 'YES'
  18         oidconsumer = consumer.Consumer(dict(), None)
  19         try:
  20             request = oidconsumer.begin('http://127.0.0.10:45080/idp1/')
  21         except Exception as ex:
  22             return 'ERROR: %s' % ex
  23
  24         if request is None:
  25             return 'ERROR: No request'
  26
  27         # Attach extensions here
  28         if self.extensions:
  29             request.addExtension(sreg.SRegRequest(
  30                 required=['nickname', 'email', 'timezone']))
  31             ax_req = ax.FetchRequest()
  32             ax_req_name = ax.AttrInfo('http://schema.openid.net/namePerson')
  33             ax_req.add(ax_req_name)
  34             request.addExtension(ax_req)
  35             username = pwd.getpwuid(os.getuid())[0]
  36             request.addExtension(teams.TeamsRequest(requested=[username]))
  37
  38         # Build and send final request
  39         trust_root = cherrypy.url()
  40         return_to = trust_root + 'finish'
  41         if request.shouldSendRedirect():
  42             redirect_url = request.redirectURL(
  43                 trust_root, return_to)
  44             raise cherrypy.HTTPRedirect(redirect_url)
  45         else:
  46             return request.htmlMarkup(
  47                 trust_root, return_to)
  48     index.exposed = True
  49
  50     def finish(self, **args):
  51         oidconsumer = consumer.Consumer(dict(), None)
  52         info = oidconsumer.complete(cherrypy.request.params, cherrypy.url())
  53         display_identifier = info.getDisplayIdentifier()
  54
  55         if info.status == consumer.FAILURE and display_identifier:
  56             return 'ERROR:Verification of %s failed: %s' % (
  57                 display_identifier, info.message)
  58         elif info.status == consumer.CANCEL:
  59             return 'ERROR: Cancelled'
  60         elif info.status == consumer.SUCCESS:
  61             username = pwd.getpwuid(os.getuid())[0]
  62             expected_identifier = 'http://127.0.0.10:45080/idp1/openid/id/%s/'\
  63                 % username
  64             if expected_identifier != display_identifier:
  65                 return 'ERROR: Wrong id returned: %s != %s' % (
  66                     expected_identifier,
  67                     display_identifier)
  68
  69             if self.extensions:
  70                 sreg_resp = sreg.SRegResponse.fromSuccessResponse(info)
  71                 teams_resp = teams.TeamsResponse.fromSuccessResponse(info)
  72                 ax_resp = ax.FetchResponse.fromSuccessResponse(info)
  73
  74                 if sreg_resp is None:
  75                     return 'ERROR: No sreg!'
  76                 elif teams_resp is None:
  77                     return 'ERROR: No teams!'
  78                 elif ax_resp is None:
  79                     return 'ERROR: No AX!'
  80
  81                 # Check values
  82                 expected_name = 'Test User %s' % username
  83                 expected_email = '%s@example.com' % username
  84
  85                 ax_name = ax_resp.data[
  86                     'http://schema.openid.net/namePerson'][0]
  87                 sreg_email = sreg_resp.data['email']
  88
  89                 if ax_name != expected_name:
  90                     return 'ERROR: Wrong name returned: %s != %s' % (
  91                         expected_name,
  92                         ax_name)
  93
  94                 if sreg_email != expected_email:
  95                     return 'ERROR: Wrong email returned: %s != %s' % (
  96                         expected_email,
  97                         sreg_email)
  98
  99                 if username not in teams_resp.teams:
 100                     return 'ERROR: User not in self-named group (%s not in %s)' %\
 101                         (username, teams_resp.teams)
 102
 103             if self.extensions:
 104                 return 'SUCCESS, WITH EXTENSIONS'
 105             else:
 106                 return 'SUCCESS, WITHOUT EXTENSIONS'
 107         else:
 108             return 'ERROR: Strange error: %s' % info.message
 109     finish.exposed = True
 110
 111
 112 cherrypy.config['environment'] = 'embedded'
 113
 114 application = cherrypy.Application(OpenIDApp(),
 115                                    script_name=None, config=None)