1# -*- coding: utf-8 -*- 

2""" 

3@file 

4@brief Starts an application. 

5""" 

6import hashlib 

7from starlette.responses import RedirectResponse 

8from itsdangerous import URLSafeTimedSerializer 

9import ujson 

10 

11 

12class AuthentificationAnswers: 

13 """ 

14 Defines answers for an application with authentification. 

15 It stores a cookie with only the user alias. 

16 The method `authentify_user <mathenjeu.apps.common.auth_app.AuthentificationAnswers.authentify_user>`_ 

17 must be overwritten. The method 

18 `page_context <mathenjeu.apps.qcm.acm_app.ACMApp.page_context>`_ 

19 returns additional information to add before applying any template. 

20 """ 

21 

22 def __init__(self, app, 

23 login_page="login.html", 

24 notauth_page="notauthorized.html", 

25 auth_page="authorized.html", 

26 redirect_logout="/", max_age=14 * 24 * 60 * 60, 

27 cookie_key=None, cookie_name="mathenjeu", 

28 cookie_domain="127.0.0.1", cookie_path="/", 

29 secure=False, page_context=None, userpwd=None): 

30 """ 

31 @param app :epkg:`starlette` application 

32 @param login_page name of the login page 

33 @param notauth_page page displayed when a user is not authorized 

34 @param auth_page page displayed when a user is authorized 

35 @param redirect_logout a not authorized used is redirected to this page 

36 @param max_age cookie's duration in seconds 

37 @param cookie_key to encrypt information in the cookie (cannot be None) 

38 @param cookie_name name of the session cookie 

39 @param cookie_domain cookie is valid for this path only 

40 @param cookie_path path of the cookie once storeds 

41 @param secure use secured connection for cookies 

42 @param page_context to retrieve additional context 

43 before rendering the pages (as a function 

44 which returns a dictionary) 

45 @param userpwd users are authentified with any alias but a common password 

46 """ 

47 if cookie_key is None: 

48 raise ValueError("cookie_key cannot be None") 

49 self.app = app 

50 self.login_page = login_page 

51 self.notauth_page = notauth_page 

52 self.auth_page = auth_page 

53 self.redirect_logout = redirect_logout 

54 self.cookie_name = cookie_name 

55 self.cookie_domain = cookie_domain 

56 self.cookie_path = cookie_path 

57 self.cookie_key = cookie_key 

58 self.max_age = max_age 

59 self.secure = secure 

60 self.signer = URLSafeTimedSerializer(self.cookie_key) 

61 self.userpwd = userpwd 

62 self.hashed_userpwd = None if userpwd is None else self.hash_pwd( 

63 userpwd) 

64 self._get_page_context = page_context 

65 app._get_session = self.get_session 

66 for method in ['log_event', 'log_any']: 

67 if hasattr(self, method): 

68 setattr(app, '_' + method, getattr(self, method)) 

69 

70 async def login(self, request): 

71 """ 

72 Login page. If paramater *returnto* is specified in the url, 

73 the user will go to this page after being logged. 

74 """ 

75 ps = request.query_params 

76 context = {'request': request, 'returnto': ps.get('returnto', '/')} 

77 context.update(self._get_page_context()) 

78 return self.templates.TemplateResponse(self.login_page, context) # pylint: disable=E1101 

79 

80 def hash_pwd(self, pwd): 

81 """ 

82 Hashes a password. 

83 

84 @param pwd password 

85 @return hashed password in hexadecimal format 

86 """ 

87 m = hashlib.sha256() 

88 m.update(pwd.encode("utf-8")) 

89 return m.hexdigest() 

90 

91 async def authenticate(self, request): 

92 """ 

93 Authentification. 

94 

95 @param request request 

96 @return response 

97 """ 

98 try: 

99 fo = await request.form() 

100 except Exception as e: 

101 raise RuntimeError( # pylint: disable=W0707 

102 "Unable to read login and password due to '{0}'".format(e)) 

103 if 'alias' not in fo: 

104 return self.is_allowed( 

105 alias=None, pwd=None, request=request) 

106 

107 ps = request.query_params 

108 loge = getattr(self, 'logevent', None) 

109 if loge: 

110 loge("authenticate", request, session={}, # pylint: disable=E1102 

111 alias=fo['alias']) 

112 res = self.is_allowed(alias=fo['alias'], pwd=fo['pwd'], 

113 request=request) 

114 if res is not None: 

115 return res 

116 data = dict(alias=fo['alias'], hashpwd=self.hash_pwd(fo['pwd'])) 

117 returnto = ps.get('returnto', '/') 

118 context = {'request': request, 

119 'alias': fo['alias'], 'returnto': returnto} 

120 context.update(self._get_page_context()) 

121 response = self.templates.TemplateResponse( # pylint: disable=E1101 

122 'authorized.html', context) 

123 self.save_session(response, data) 

124 return response 

125 # response = RedirectResponse(url=returnto) 

126 # return response 

127 

128 async def logout(self, request): 

129 """ 

130 Logout page. 

131 """ 

132 response = RedirectResponse(url=self.redirect_logout) 

133 response.delete_cookie(self.cookie_name, domain=self.cookie_domain, 

134 path=self.cookie_path) 

135 return response 

136 

137 def save_session(self, response, data): 

138 """ 

139 Saves the session to the response in a secure cookie. 

140 

141 @param response response 

142 @param data data 

143 """ 

144 data = ujson.dumps(data) # pylint: disable=E1101 

145 signed_data = self.signer.dumps([data]) # pylint: disable=E1101 

146 response.set_cookie(self.cookie_name, signed_data, 

147 max_age=self.max_age, 

148 httponly=True, domain=self.cookie_domain, 

149 path=self.cookie_path, secure=self.secure) 

150 

151 def get_session(self, request, notnone=False): 

152 """ 

153 Retrieves the session. 

154 

155 @param request request 

156 @param notnone None or empty dictionary 

157 @return session 

158 """ 

159 cook = request.cookies.get(self.cookie_name) 

160 if cook is not None: 

161 unsigned = self.signer.loads(cook) 

162 data = unsigned[0] 

163 jsdata = ujson.loads(data) # pylint: disable=E1101 

164 # We check the hashed password is still good. 

165 hashpwd = jsdata.get('hashpwd', '') 

166 if not self.authentify_user(jsdata.get('alias', ''), hashpwd, False): 

167 # We cancel the authentification. 

168 return {} 

169 return jsdata 

170 return {} if notnone else None 

171 

172 def is_allowed(self, alias, pwd, request): 

173 """ 

174 Checks that a user is allowed. Returns None if it is allowed, 

175 otherwise an page with an error message. 

176 

177 @param alias alias or iser 

178 @param pwd password 

179 @param request received request 

180 @return None if allowed, *HTMLResponse* otherwise 

181 """ 

182 if not self.authentify_user(alias, pwd): 

183 context = {'request': request, 'alias': alias} 

184 context.update(self._get_page_context()) 

185 return self.templates.TemplateResponse('notauthorized.html', context) # pylint: disable=E1101 

186 return None 

187 

188 def authentify_user(self, alias, pwd, hash_before=True): 

189 """ 

190 Overwrites this method to allow or reject users. 

191 

192 @param alias alias or user 

193 @param pwd password 

194 @param hash_before hashes the password before comparing, otherwise, 

195 the function assumes it is already hashed 

196 @return boolean 

197 

198 The current behavior is to allow anybody if the alias is longer 

199 than 3 characters. 

200 """ 

201 if alias is None or len(alias.strip()) <= 3: 

202 return False 

203 if self.hashed_userpwd is None: 

204 return True 

205 if hash_before: 

206 hashed_pwd = self.hash_pwd(pwd) 

207 return hashed_pwd == self.hashed_userpwd 

208 return pwd == self.hashed_userpwd