repoze.who.plugins.form

1 import urlparse 2 import urllib 3 import cgi 4 5 from paste.httpheaders import CONTENT_LENGTH 6 from paste.httpheaders import CONTENT_TYPE 7 from paste.httpheaders import LOCATION 8 from paste.httpexceptions import HTTPFound 9 from paste.httpexceptions import HTTPUnauthorized 10 11 from paste.request import parse_dict_querystring 12 from paste.request import parse_formvars 13 from paste.request import construct_url 14 15 from paste.response import header_value 16 17 from zope.interface import implements 18 19 from repoze.who.config import _resolve 20 from repoze.who.interfaces import IChallenger 21 from repoze.who.interfaces import IIdentifier 22 23 _DEFAULT_FORM = """ 24 <html> 25 <head> 26 <title>Log In</title> 27 </head> 28 <body> 29 <div> 30 <b>Log In</b> 31 </div> 32 <br/> 33 <form method="POST" action="?__do_login=true"> 34 <table border="0"> 35 <tr> 36 <td>User Name</td> 37 <td><input type="text" name="login"></input></td> 38 </tr> 39 <tr> 40 <td>Password</td> 41 <td><input type="password" name="password"></input></td> 42 </tr> 43 <tr> 44 <td></td> 45 <td><input type="submit" name="submit" value="Log In"/></td> 46 </tr> 47 </table> 48 </form> 49 <pre> 50 </pre> 51 </body> 52 </html> 53 """ 54

55 -class FormPluginBase(object):

56 - def _get_rememberer(self, environ):

57 rememberer = environ['repoze.who.plugins'][self.rememberer_name] 58 return rememberer

59 60 # IIdentifier

61 - def remember(self, environ, identity):

62 rememberer = self._get_rememberer(environ) 63 return rememberer.remember(environ, identity)

64 65 # IIdentifier

66 - def forget(self, environ, identity):

67 rememberer = self._get_rememberer(environ) 68 return rememberer.forget(environ, identity)

69

70 - def __repr__(self):

71 return '<%s %s>' % (self.__class__.__name__, 72 id(self)) #pragma NO COVERAGE

73

74 -class FormPlugin(FormPluginBase):

75 76 implements(IChallenger, IIdentifier) 77

78 - def __init__(self, login_form_qs, rememberer_name, formbody=None, 79 formcallable=None):

80 self.login_form_qs = login_form_qs 81 # rememberer_name is the name of another configured plugin which 82 # implements IIdentifier, to handle remember and forget duties 83 # (ala a cookie plugin or a session plugin) 84 self.rememberer_name = rememberer_name 85 self.formbody = formbody 86 self.formcallable = formcallable

87 88 # IIdentifier

89 - def identify(self, environ):

90 query = parse_dict_querystring(environ) 91 # If the extractor finds a special query string on any request, 92 # it will attempt to find the values in the input body. 93 if query.get(self.login_form_qs): 94 form = parse_formvars(environ) 95 from StringIO import StringIO 96 # XXX we need to replace wsgi.input because we've read it 97 # this smells funny 98 environ['wsgi.input'] = StringIO() 99 form.update(query) 100 try: 101 login = form['login'] 102 password = form['password'] 103 except KeyError: 104 return None 105 del query[self.login_form_qs] 106 environ['QUERY_STRING'] = urllib.urlencode(query) 107 environ['repoze.who.application'] = HTTPFound( 108 construct_url(environ)) 109 credentials = {'login':login, 'password':password} 110 max_age = form.get('max_age', None) 111 if max_age is not None: 112 credentials['max_age'] = max_age 113 return credentials 114 115 return None

116 117 # IChallenger

118 - def challenge(self, environ, status, app_headers, forget_headers):

119 if app_headers: 120 location = LOCATION(app_headers) 121 if location: 122 headers = list(app_headers) + list(forget_headers) 123 return HTTPFound(headers = headers) 124 125 form = self.formbody or _DEFAULT_FORM 126 if self.formcallable is not None: 127 form = self.formcallable(environ) 128 def auth_form(environ, start_response): 129 content_length = CONTENT_LENGTH.tuples(str(len(form))) 130 content_type = CONTENT_TYPE.tuples('text/html') 131 headers = content_length + content_type + forget_headers 132 start_response('200 OK', headers) 133 return [form]

134 135 return auth_form

136

137 -class RedirectingFormPlugin(FormPluginBase):

138 139 implements(IChallenger, IIdentifier) 140

141 - def __init__(self, login_form_url, login_handler_path, logout_handler_path, 142 rememberer_name, reason_param='reason'):

143 self.login_form_url = login_form_url 144 self.login_handler_path = login_handler_path 145 self.logout_handler_path = logout_handler_path 146 # rememberer_name is the name of another configured plugin which 147 # implements IIdentifier, to handle remember and forget duties 148 # (ala a cookie plugin or a session plugin) 149 self.rememberer_name = rememberer_name 150 self.reason_param = reason_param

151 152 # IIdentifier

153 - def identify(self, environ):

154 path_info = environ['PATH_INFO'] 155 query = parse_dict_querystring(environ) 156 157 if path_info == self.logout_handler_path: 158 # we've been asked to perform a logout 159 form = parse_formvars(environ) 160 form.update(query) 161 referer = environ.get('HTTP_REFERER', '/') 162 came_from = form.get('came_from', referer) 163 # set in environ for self.challenge() to find later 164 environ['came_from'] = came_from 165 environ['repoze.who.application'] = HTTPUnauthorized() 166 return None 167 168 elif path_info == self.login_handler_path: 169 # we've been asked to perform a login 170 form = parse_formvars(environ) 171 form.update(query) 172 try: 173 login = form['login'] 174 password = form['password'] 175 max_age = form.get('max_age', None) 176 credentials = { 177 'login':form['login'], 178 'password':form['password'], 179 } 180 except KeyError: 181 credentials = None 182 183 if credentials is not None: 184 max_age = form.get('max_age', None) 185 if max_age is not None: 186 credentials['max_age'] = max_age 187 188 referer = environ.get('HTTP_REFERER', '/') 189 came_from = form.get('came_from', referer) 190 environ['repoze.who.application'] = HTTPFound(came_from) 191 return credentials

192 193 # IChallenger

194 - def challenge(self, environ, status, app_headers, forget_headers):

195 reason = header_value(app_headers, 'X-Authorization-Failure-Reason') 196 url_parts = list(urlparse.urlparse(self.login_form_url)) 197 query = url_parts[4] 198 query_elements = cgi.parse_qs(query) 199 came_from = environ.get('came_from', construct_url(environ)) 200 query_elements['came_from'] = came_from 201 if reason: 202 query_elements[self.reason_param] = reason 203 url_parts[4] = urllib.urlencode(query_elements, doseq=True) 204 login_form_url = urlparse.urlunparse(url_parts) 205 headers = [ ('Location', login_form_url) ] 206 cookies = [(h,v) for (h,v) in app_headers if h.lower() == 'set-cookie'] 207 headers = headers + forget_headers + cookies 208 return HTTPFound(headers=headers)

209

210 -def make_plugin(login_form_qs='__do_login', 211 rememberer_name=None, 212 form=None, 213 formcallable=None, 214 ):

215 if rememberer_name is None: 216 raise ValueError( 217 'must include rememberer key (name of another IIdentifier plugin)') 218 if form is not None: 219 form = open(form).read() 220 if isinstance(formcallable, str): 221 formcallable = _resolve(formcallable) 222 plugin = FormPlugin(login_form_qs, rememberer_name, form, formcallable) 223 return plugin

224

225 -def make_redirecting_plugin(login_form_url=None, 226 login_handler_path='/login_handler', 227 logout_handler_path='/logout_handler', 228 rememberer_name=None):

229 if login_form_url is None: 230 raise ValueError( 231 'must include login_form_url in configuration') 232 if login_handler_path is None: 233 raise ValueError( 234 'login_handler_path must not be None') 235 if logout_handler_path is None: 236 raise ValueError( 237 'logout_handler_path must not be None') 238 if rememberer_name is None: 239 raise ValueError( 240 'must include rememberer key (name of another IIdentifier plugin)') 241 plugin = RedirectingFormPlugin(login_form_url, 242 login_handler_path, 243 logout_handler_path, 244 rememberer_name) 245 return plugin

246

Source Code for Module repoze.who.plugins.form