]> git.openstreetmap.org Git - osqa.git/blob - forum_modules/openidauth/consumer.py
6ed5b2b111e3545a1eac1d21cfb789c0b7c6d040
[osqa.git] / forum_modules / openidauth / consumer.py
1 import re
2
3 from django.utils.html import escape
4 from django.http import get_host
5
6 from forum.authentication.base import AuthenticationConsumer, InvalidAuthentication
7 import settings
8
9 from openid.yadis import xri
10 from openid.consumer.consumer import Consumer, SUCCESS, CANCEL, FAILURE, SETUP_NEEDED
11 from openid.consumer.discover import DiscoveryFailure
12 from openid.extensions.sreg import SRegRequest, SRegResponse
13 from openid.extensions.ax import FetchRequest as AXFetchRequest, AttrInfo, FetchResponse as AXFetchResponse
14 from django.utils.translation import ugettext as _
15
16 from store import OsqaOpenIDStore
17
18 class OpenIdAbstractAuthConsumer(AuthenticationConsumer):
19
20     dataype2ax_schema = {
21         'username': 'http://axschema.org/namePerson/friendly',
22         'email': 'http://axschema.org/contact/email',
23         #'web': 'http://axschema.org/contact/web/default',
24         #'firstname': 'http://axschema.org/namePerson/first',
25         #'lastname': 'http://axschema.org/namePerson/last',
26         #'birthdate': 'http://axschema.org/birthDate',
27     }
28
29     sreg_attributes = {
30         "required": {
31             "email": "email",
32             "nickname": "username"
33         }
34     }
35
36     def get_user_url(self, request):
37         try:
38             return request.POST['openid_identifier']
39         except:
40             raise NotImplementedError()
41
42     def prepare_authentication_request(self, request, redirect_to):
43         if not redirect_to.startswith('http://') or redirect_to.startswith('https://'):
44                     redirect_to =  get_url_host(request) + redirect_to
45
46         user_url = self.get_user_url(request)
47
48         if xri.identifierScheme(user_url) == 'XRI' and getattr(
49             settings, 'OPENID_DISALLOW_INAMES', False
50         ):
51             raise InvalidAuthentication('i-names are not supported')
52
53         consumer = Consumer(request.session, OsqaOpenIDStore())
54
55         try:
56             auth_request = consumer.begin(user_url)
57         except DiscoveryFailure:
58             raise InvalidAuthentication(_('Sorry, but your input is not a valid OpenId'))
59
60         sreg = getattr(self, 'sreg_attributes', False)
61
62         if sreg:
63             s = SRegRequest()
64
65             for k, attr_dic in sreg.items():
66                 if k == "policy_url":
67                     s.policy_url = attr_dic
68                     continue
69
70                 for attr_name in attr_dic.keys():
71                     s.requestField(field_name=attr_name, required=(k == "required"))
72
73             auth_request.addExtension(s)
74
75         ax_schema = getattr(self, 'dataype2ax_schema', False)
76
77         if ax_schema and request.session.get('force_email_request', True):
78             axr = AXFetchRequest()
79             for data_type, schema in ax_schema.items():
80                 if isinstance(schema, tuple):
81                     axr.add(AttrInfo(schema[0], 1, True, schema[1]))
82                 else:
83                     axr.add(AttrInfo(schema, 1, True, data_type))
84
85             auth_request.addExtension(axr)
86
87         trust_root = getattr(
88             settings, 'OPENID_TRUST_ROOT', get_url_host(request) + '/'
89         )
90
91         return auth_request.redirectURL(trust_root, redirect_to)
92
93     def process_authentication_request(self, request):
94         consumer = Consumer(request.session, OsqaOpenIDStore())
95
96         query_dict = dict([
97             (k.encode('utf8'), v.encode('utf8')) for k, v in request.GET.items()
98         ])
99
100         #for i in query_dict.items():
101             #print "%s : %s" % i
102
103         url = get_url_host(request) + request.path
104         openid_response = consumer.complete(query_dict, url)
105
106         if openid_response.status == SUCCESS:
107
108             consumer_data = {}
109
110             sreg_attrs = getattr(self, 'sreg_attributes', False)
111
112             if sreg_attrs:
113                 sreg_response = SRegResponse.fromSuccessResponse(openid_response)
114
115                 if sreg_response:
116                     all_attrs = {}
117                     [all_attrs.update(d) for k,d in sreg_attrs.items() if k != "policy_url"]
118
119                     for attr_name, local_name in all_attrs.items():
120                         if attr_name in sreg_response:
121                             consumer_data[local_name] = sreg_response[attr_name]
122
123             ax_schema = getattr(self, 'dataype2ax_schema', False)
124
125             if ax_schema:
126                 ax = AXFetchResponse.fromSuccessResponse(openid_response)
127
128                 if ax:
129                     axargs = ax.getExtensionArgs()
130
131                     ax_schema2data_type = dict([(s, t) for t, s in ax_schema.items()])
132
133                     available_types = dict([
134                         (ax_schema2data_type[s], re.sub('^type\.', '', n))
135                         for n, s in axargs.items() if s in ax_schema2data_type
136                     ])
137
138                     for t, s in available_types.items():
139                         if not t in consumer_data:
140                             if axargs.get("value.%s.1" % s, None):
141                                 consumer_data[t] = axargs["value.%s.1" % s]
142                     
143             request.session['auth_consumer_data'] = consumer_data
144
145
146             return request.GET['openid.identity']
147         elif openid_response.status == CANCEL:
148             raise InvalidAuthentication(_('The OpenId authentication request was canceled'))
149         elif openid_response.status == FAILURE:
150             raise InvalidAuthentication(_('The OpenId authentication failed: ') + openid_response.message)
151         elif openid_response.status == SETUP_NEEDED:
152             raise InvalidAuthentication(_('Setup needed'))
153         else:
154             raise InvalidAuthentication(_('The OpenId authentication failed with an unknown status: ') + openid_response.status)
155
156     def get_user_data(self, key):
157         return {}
158
159 def get_url_host(request):
160     if request.is_secure():
161         protocol = 'https'
162     else:
163         protocol = 'http'
164     host = escape(get_host(request))
165     return '%s://%s' % (protocol, host)
166
167 def get_full_url(request):
168     return get_url_host(request) + request.get_full_path()