tests/blobs/openid_app.py

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