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

1""" 

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

3""" 

4 

5import hashlib 

6import logging 

7import time 

8import uuid 

9from datetime import datetime, timedelta, timezone 

10 

11from cdt_identity.claims import ClaimsResult 

12from cdt_identity.session import Session as OAuthSession 

13from django.urls import reverse 

14 

15from benefits.enrollment_littlepay.session import Session as LittlepaySession 

16from benefits.enrollment_switchio.session import Session as SwitchioSession 

17from benefits.routes import routes 

18 

19from . import models 

20 

21logger = logging.getLogger(__name__) 

22 

23 

24_AGENCY = "agency" 

25_DEBUG = "debug" 

26_DID = "did" 

27_ELIGIBLE = "eligibility" 

28_ENROLLMENT_EXP = "enrollment_expiry" 

29_FLOW = "flow" 

30_GROUP = "group" # EnrollmentGroup, not django.auth Group 

31_LANG = "lang" 

32_LOGGED_IN = "logged_in" 

33_ORIGIN = "origin" 

34_START = "start" 

35_UID = "uid" 

36 

37 

38def agency(request): 

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

40 agency_id = request.session.get(_AGENCY) 

41 if not agency_id: 

42 return None 

43 

44 if getattr(request, "_cached_agency", None) and request._cached_agency.id == agency_id: 

45 return request._cached_agency 

46 try: 

47 agency = models.TransitAgency.by_id(agency_id) 

48 request._cached_agency = agency 

49 return agency 

50 except models.TransitAgency.DoesNotExist: 

51 return None 

52 

53 

54def active_agency(request): 

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

56 a = agency(request) 

57 return a and a.active 

58 

59 

60def context_dict(request): 

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

62 littlepay_session = LittlepaySession(request) 

63 switchio_session = SwitchioSession(request) 

64 return { 

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

66 _DEBUG: debug(request), 

67 _DID: did(request), 

68 _FLOW: flow(request), 

69 _GROUP: group(request), 

70 _ELIGIBLE: eligible(request), 

71 _ENROLLMENT_EXP: enrollment_expiry(request), 

72 littlepay_session._keys_access_token: littlepay_session.access_token, 

73 littlepay_session._keys_access_token_expiry: littlepay_session.access_token_expiry, 

74 switchio_session._keys_registration_id: switchio_session.registration_id, 

75 _LANG: language(request), 

76 _LOGGED_IN: logged_in(request), 

77 _ORIGIN: origin(request), 

78 _START: start(request), 

79 _UID: uid(request), 

80 } 

81 

82 

83def debug(request): 

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

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

86 

87 

88def did(request): 

89 """ 

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

91 the session is reset to initialize a value. 

92 

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

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

95 user. 

96 

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

98 """ 

99 d = request.session.get(_DID) 

100 if not d: 

101 reset(request) 

102 d = request.session.get(_DID) 

103 return str(d) 

104 

105 

106def eligible(request): 

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

108 return request.session.get(_ELIGIBLE) 

109 

110 

111def enrollment_expiry(request): 

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

113 expiry = request.session.get(_ENROLLMENT_EXP) 

114 if expiry: 

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

116 else: 

117 return None 

118 

119 

120def enrollment_reenrollment(request): 

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

122 expiry = enrollment_expiry(request) 

123 enrollment_flow = flow(request) 

124 

125 if enrollment_flow and enrollment_flow.supports_expiration and expiry: 

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

127 else: 

128 return None 

129 

130 

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

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

133 flow_id = request.session.get(_FLOW) 

134 if not flow_id: 

135 return None 

136 

137 if getattr(request, "_cached_flow", None) and request._cached_flow.id == flow_id: 

138 return request._cached_flow 

139 try: 

140 flow = models.EnrollmentFlow.by_id(flow_id) 

141 request._cached_flow = flow 

142 return flow 

143 except models.EnrollmentFlow.DoesNotExist: 

144 return None 

145 

146 

147def group(request) -> models.EnrollmentGroup | None: 

148 """Get the EnrollmentGroup from the request's session, or None""" 

149 

150 if agency(request) and agency(request).transit_processor: 

151 group_model = agency(request).transit_processor.group_model 

152 

153 try: 

154 return group_model.by_id(request.session[_GROUP]) 

155 except (KeyError, group_model.DoesNotExist): 

156 return None 

157 

158 return None 

159 

160 

161def language(request): 

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

163 return request.LANGUAGE_CODE 

164 

165 

166def logged_in(request): 

167 """Get the user's status of having logged in with OAuth from the request's session, or None""" 

168 return bool(request.session.get(_LOGGED_IN)) 

169 

170 

171def logout(request): 

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

173 LittlepaySession(request, reset=True) 

174 SwitchioSession(request, reset=True) 

175 OAuthSession(request, claims_result=ClaimsResult()) 

176 update(request, logged_in=False) 

177 

178 

179def oauth_extra_claims(request): 

180 """Get the extra oauth claims from the request's session, or None""" 

181 claims = [claim for claim, value in OAuthSession(request).claims_result.verified.items() if value] 

182 

183 if claims: 

184 f = flow(request) 

185 if f and f.uses_claims_verification: 

186 claims.remove(f.claims_request.eligibility_claim) 

187 return claims or None 

188 raise Exception("Oauth claims but no flow") 

189 else: 

190 return None 

191 

192 

193def origin(request): 

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

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

196 

197 

198def reset(request): 

199 """Reset the session for the request.""" 

200 logger.debug("Reset session") 

201 request.session[_AGENCY] = None 

202 request.session[_FLOW] = None 

203 request.session[_GROUP] = None 

204 request.session[_ELIGIBLE] = False 

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

206 request.session[_ENROLLMENT_EXP] = None 

207 request.session[_LOGGED_IN] = False 

208 LittlepaySession(request, reset=True) 

209 SwitchioSession(request, reset=True) 

210 OAuthSession(request, reset=True) 

211 

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

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

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

215 u = str(uuid.uuid4()) 

216 request.session[_UID] = u 

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

218 

219 

220def start(request): 

221 """ 

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

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

224 

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

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

227 sessions. 

228 

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

230 """ 

231 s = request.session.get(_START) 

232 if not s: 

233 reset(request) 

234 s = request.session.get(_START) 

235 return s 

236 

237 

238def uid(request): 

239 """ 

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

241 the session is reset to initialize a value. 

242 

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

244 sequence of events came from a unique user. 

245 

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

247 

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

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

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

251 """ 

252 u = request.session.get(_UID) 

253 if not u: 

254 reset(request) 

255 u = request.session.get(_UID) 

256 return u 

257 

258 

259def update( 

260 request, 

261 agency=None, 

262 debug=None, 

263 flow=None, 

264 group=None, 

265 eligible=None, 

266 enrollment_expiry=None, 

267 logged_in=None, 

268 origin=None, 

269): 

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

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

272 request.session[_AGENCY] = agency.id 

273 if debug is not None: 

274 request.session[_DEBUG] = debug 

275 if eligible is not None: 

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

277 if isinstance(enrollment_expiry, datetime): 

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

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

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

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

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

283 # > timestamp by supplying tzinfo=timezone.utc 

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

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

286 if logged_in is not None: 

287 request.session[_LOGGED_IN] = logged_in 

288 if origin is not None: 

289 request.session[_ORIGIN] = origin 

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

291 request.session[_FLOW] = flow.id 

292 oauth_session = OAuthSession(request) 

293 oauth_session.client_config = flow.oauth_config 

294 oauth_session.claims_request = flow.claims_request 

295 if group is not None and isinstance(group, models.EnrollmentGroup): 

296 request.session[_GROUP] = group.id