Coverage for benefits/core/session.py: 99%

1""" 

2The core application: helpers to work with request sessions. 

3""" 

4 

5from datetime import datetime, timedelta, timezone 

6import hashlib 

7import logging 

8import time 

9import uuid 

10 

11from django.urls import reverse 

12 

13from benefits.routes import routes 

14from . import models 

15 

16 

17logger = logging.getLogger(__name__) 

18 

19 

20_AGENCY = "agency" 

21_DEBUG = "debug" 

22_DID = "did" 

23_ELIGIBLE = "eligibility" 

24_ENROLLMENT_TOKEN = "enrollment_token" 

25_ENROLLMENT_TOKEN_EXP = "enrollment_token_expiry" 

26_ENROLLMENT_EXP = "enrollment_expiry" 

27_FLOW = "flow" 

28_LANG = "lang" 

29_OAUTH_CLAIM = "oauth_claim" 

30_OAUTH_TOKEN = "oauth_token" 

31_ORIGIN = "origin" 

32_START = "start" 

33_UID = "uid" 

34 

35 

36def agency(request): 

37 """Get the agency from the request's session, or None""" 

38 try: 

39 return models.TransitAgency.by_id(request.session[_AGENCY]) 

40 except (KeyError, models.TransitAgency.DoesNotExist): 

41 return None 

42 

43 

44def active_agency(request): 

45 """True if the request's session is configured with an active agency. False otherwise.""" 

46 a = agency(request) 

47 return a and a.active 

48 

49 

50def context_dict(request): 

51 """The request's session context as a dict.""" 

52 return { 

53 _AGENCY: agency(request).slug if active_agency(request) else None, 

54 _DEBUG: debug(request), 

55 _DID: did(request), 

56 _FLOW: flow(request), 

57 _ELIGIBLE: eligible(request), 

58 _ENROLLMENT_EXP: enrollment_expiry(request), 

59 _ENROLLMENT_TOKEN: enrollment_token(request), 

60 _ENROLLMENT_TOKEN_EXP: enrollment_token_expiry(request), 

61 _LANG: language(request), 

62 _OAUTH_TOKEN: oauth_token(request), 

63 _OAUTH_CLAIM: oauth_claim(request), 

64 _ORIGIN: origin(request), 

65 _START: start(request), 

66 _UID: uid(request), 

67 } 

68 

69 

70def debug(request): 

71 """Get the DEBUG flag from the request's session.""" 

72 return bool(request.session.get(_DEBUG, False)) 

73 

74 

75def did(request): 

76 """ 

77 Get the session's device ID, a hashed version of the unique ID. If unset, 

78 the session is reset to initialize a value. 

79 

80 This value, like UID, is randomly generated per session and is needed for 

81 Amplitude to accurately track that a sequence of events came from a unique 

82 user. 

83 

84 See more: https://help.amplitude.com/hc/en-us/articles/115003135607-Track-unique-users-in-Amplitude 

85 """ 

86 d = request.session.get(_DID) 

87 if not d: 

88 reset(request) 

89 d = request.session.get(_DID) 

90 return str(d) 

91 

92 

93def eligible(request): 

94 """True if the request's session has confirmed eligibility. False otherwise.""" 

95 return request.session.get(_ELIGIBLE) 

96 

97 

98def enrollment_expiry(request): 

99 """Get the expiry date for a user's enrollment from session, or None.""" 

100 expiry = request.session.get(_ENROLLMENT_EXP) 

101 if expiry: 

102 return datetime.fromtimestamp(expiry, tz=timezone.utc) 

103 else: 

104 return None 

105 

106 

107def enrollment_reenrollment(request): 

108 """Get the reenrollment date for a user's enrollment from session, or None.""" 

109 expiry = enrollment_expiry(request) 

110 enrollment_flow = flow(request) 

111 

112 if enrollment_flow and enrollment_flow.supports_expiration and expiry: 

113 return expiry - timedelta(days=enrollment_flow.expiration_reenrollment_days) 

114 else: 

115 return None 

116 

117 

118def enrollment_token(request): 

119 """Get the enrollment token from the request's session, or None.""" 

120 return request.session.get(_ENROLLMENT_TOKEN) 

121 

122 

123def enrollment_token_expiry(request): 

124 """Get the enrollment token's expiry time from the request's session, or None.""" 

125 return request.session.get(_ENROLLMENT_TOKEN_EXP) 

126 

127 

128def enrollment_token_valid(request): 

129 """True if the request's session is configured with a valid token. False otherwise.""" 

130 if bool(enrollment_token(request)): 

131 exp = enrollment_token_expiry(request) 

132 # ensure token does not expire in the next 5 seconds 

133 valid = exp is None or exp > (time.time() + 5) 

134 return valid 

135 else: 

136 return False 

137 

138 

139def language(request): 

140 """Get the language configured for the request.""" 

141 return request.LANGUAGE_CODE 

142 

143 

144def logged_in(request): 

145 """Check if the current session has an OAuth token.""" 

146 return bool(oauth_token(request)) 

147 

148 

149def logout(request): 

150 """Reset the session claims and tokens.""" 

151 update(request, oauth_claim=False, oauth_token=False, enrollment_token=False) 

152 

153 

154def oauth_token(request): 

155 """Get the oauth token from the request's session, or None""" 

156 return request.session.get(_OAUTH_TOKEN) 

157 

158 

159def oauth_claim(request): 

160 """Get the oauth claim from the request's session, or None""" 

161 return request.session.get(_OAUTH_CLAIM) 

162 

163 

164def origin(request): 

165 """Get the origin for the request's session, or default to the index route.""" 

166 return request.session.get(_ORIGIN, reverse(routes.INDEX)) 

167 

168 

169def reset(request): 

170 """Reset the session for the request.""" 

171 logger.debug("Reset session") 

172 request.session[_AGENCY] = None 

173 request.session[_FLOW] = None 

174 request.session[_ELIGIBLE] = False 

175 request.session[_ORIGIN] = reverse(routes.INDEX) 

176 request.session[_ENROLLMENT_EXP] = None 

177 request.session[_ENROLLMENT_TOKEN] = None 

178 request.session[_ENROLLMENT_TOKEN_EXP] = None 

179 request.session[_OAUTH_TOKEN] = None 

180 request.session[_OAUTH_CLAIM] = None 

181 

182 if _UID not in request.session or not request.session[_UID]: 

183 logger.debug("Reset session time and uid") 

184 request.session[_START] = int(time.time() * 1000) 

185 u = str(uuid.uuid4()) 

186 request.session[_UID] = u 

187 request.session[_DID] = str(uuid.UUID(hashlib.sha512(bytes(u, "utf8")).hexdigest()[:32])) 

188 

189 

190def start(request): 

191 """ 

192 Get the start time from the request's session, as integer milliseconds since 

193 Epoch. If unset, the session is reset to initialize a value. 

194 

195 Once started, does not reset after subsequent calls to session.reset() or 

196 session.start(). This value is needed for Amplitude to accurately track 

197 sessions. 

198 

199 See more: https://help.amplitude.com/hc/en-us/articles/115002323627-Tracking-Sessions 

200 """ 

201 s = request.session.get(_START) 

202 if not s: 

203 reset(request) 

204 s = request.session.get(_START) 

205 return s 

206 

207 

208def uid(request): 

209 """ 

210 Get the session's unique ID, a randomly generated UUID4 string. If unset, 

211 the session is reset to initialize a value. 

212 

213 This value, like DID, is needed for Amplitude to accurately track that a 

214 sequence of events came from a unique user. 

215 

216 See more: https://help.amplitude.com/hc/en-us/articles/115003135607-Track-unique-users-in-Amplitude 

217 

218 Although Amplitude advises *against* setting user_id for anonymous users, 

219 here a value is set on anonymous users anyway, as the users never sign-in 

220 and become de-anonymized to this app / Amplitude. 

221 """ 

222 u = request.session.get(_UID) 

223 if not u: 

224 reset(request) 

225 u = request.session.get(_UID) 

226 return u 

227 

228 

229def update( 

230 request, 

231 agency=None, 

232 debug=None, 

233 flow=None, 

234 eligible=None, 

235 enrollment_expiry=None, 

236 enrollment_token=None, 

237 enrollment_token_exp=None, 

238 oauth_token=None, 

239 oauth_claim=None, 

240 origin=None, 

241): 

242 """Update the request's session with non-null values.""" 

243 if agency is not None and isinstance(agency, models.TransitAgency): 

244 request.session[_AGENCY] = agency.id 

245 if debug is not None: 

246 request.session[_DEBUG] = debug 

247 if eligible is not None: 

248 request.session[_ELIGIBLE] = bool(eligible) 

249 if isinstance(enrollment_expiry, datetime): 

250 if enrollment_expiry.tzinfo is None or enrollment_expiry.tzinfo.utcoffset(enrollment_expiry) is None: 

251 # this is a naive datetime instance, update tzinfo for UTC 

252 # see notes under https://docs.python.org/3/library/datetime.html#datetime.datetime.timestamp 

253 # > There is no method to obtain the POSIX timestamp directly from a naive datetime instance representing UTC time. 

254 # > If your application uses this convention and your system timezone is not set to UTC, you can obtain the POSIX 

255 # > timestamp by supplying tzinfo=timezone.utc 

256 enrollment_expiry = enrollment_expiry.replace(tzinfo=timezone.utc) 

257 request.session[_ENROLLMENT_EXP] = enrollment_expiry.timestamp() 

258 if enrollment_token is not None: 

259 request.session[_ENROLLMENT_TOKEN] = enrollment_token 

260 request.session[_ENROLLMENT_TOKEN_EXP] = enrollment_token_exp 

261 if oauth_token is not None: 

262 request.session[_OAUTH_TOKEN] = oauth_token 

263 if oauth_claim is not None: 

264 request.session[_OAUTH_CLAIM] = oauth_claim 

265 if origin is not None: 

266 request.session[_ORIGIN] = origin 

267 if flow is not None and isinstance(flow, models.EnrollmentFlow): 

268 request.session[_FLOW] = flow.id 

269 

270 

271def flow(request) -> models.EnrollmentFlow | None: 

272 """Get the EnrollmentFlow from the request's session, or None""" 

273 try: 

274 return models.EnrollmentFlow.by_id(request.session[_FLOW]) 

275 except (KeyError, models.EnrollmentFlow.DoesNotExist): 

276 return None