Coverage for chalicelib/core/users.py: 75%

1import json 

2import secrets 

3from typing import Optional 

4 

5from decouple import config 

6from fastapi import BackgroundTasks 

7from pydantic import BaseModel, model_validator 

8 

9import schemas 

10from chalicelib.core import authorizers 

11from chalicelib.core import tenants, spot 

12from chalicelib.utils import email_helper 

13from chalicelib.utils import helper 

14from chalicelib.utils import pg_client 

15from chalicelib.utils.TimeUTC import TimeUTC 

16from cachetools import TTLCache, cached 

17 

18AUDIENCE = "front:OpenReplay" 

19 

20 

21def __generate_invitation_token(): 

22 return secrets.token_urlsafe(64) 

23 

24 

25def create_new_member(email, invitation_token, admin, name, owner=False): 

26 with pg_client.PostgresClient() as cur: 

27 query = cur.mogrify(f"""\ 

28 WITH u AS (INSERT INTO public.users (email, role, name, data) 

29 VALUES (%(email)s, %(role)s, %(name)s, %(data)s) 

30 RETURNING user_id,email,role,name,created_at 

31 ), 

32 au AS (INSERT INTO public.basic_authentication (user_id, invitation_token, invited_at) 

33 VALUES ((SELECT user_id FROM u), %(invitation_token)s, timezone('utc'::text, now())) 

34 RETURNING invitation_token 

35 ) 

36 SELECT u.user_id, 

37 u.email, 

38 u.role, 

39 u.name, 

40 u.created_at, 

41 (CASE WHEN u.role = 'owner' THEN TRUE ELSE FALSE END) AS super_admin, 

42 (CASE WHEN u.role = 'admin' THEN TRUE ELSE FALSE END) AS admin, 

43 (CASE WHEN u.role = 'member' THEN TRUE ELSE FALSE END) AS member, 

44 au.invitation_token 

45 FROM u,au;""", 

46 {"email": email, "role": "owner" if owner else "admin" if admin else "member", "name": name, 

47 "data": json.dumps({"lastAnnouncementView": TimeUTC.now()}), 

48 "invitation_token": invitation_token}) 

49 cur.execute(query) 

50 row = helper.dict_to_camel_case(cur.fetchone()) 

51 if row: 51 ↛ 53line 51 didn't jump to line 53 because the condition on line 51 was always true

52 row["createdAt"] = TimeUTC.datetime_to_timestamp(row["createdAt"]) 

53 return row 

54 

55 

56def restore_member(user_id, email, invitation_token, admin, name, owner=False): 

57 with pg_client.PostgresClient() as cur: 

58 query = cur.mogrify(f"""\ 

59 WITH ua AS (UPDATE public.basic_authentication 

60 SET invitation_token = %(invitation_token)s, 

61 invited_at = timezone('utc'::text, now()), 

62 change_pwd_expire_at = NULL, 

63 change_pwd_token = NULL 

64 WHERE user_id=%(user_id)s 

65 RETURNING invitation_token) 

66 UPDATE public.users 

67 SET name= %(name)s, 

68 role = %(role)s, 

69 deleted_at= NULL, 

70 created_at = timezone('utc'::text, now()), 

71 api_key= generate_api_key(20) 

72 WHERE user_id=%(user_id)s 

73 RETURNING  

74 user_id, 

75 email, 

76 role, 

77 name, 

78 (CASE WHEN role = 'owner' THEN TRUE ELSE FALSE END) AS super_admin, 

79 (CASE WHEN role = 'admin' THEN TRUE ELSE FALSE END) AS admin, 

80 (CASE WHEN role = 'member' THEN TRUE ELSE FALSE END) AS member, 

81 created_at, 

82 (SELECT invitation_token FROM ua) AS invitation_token;""", 

83 {"user_id": user_id, "email": email, 

84 "role": "owner" if owner else "admin" if admin else "member", 

85 "name": name, "invitation_token": invitation_token}) 

86 cur.execute(query) 

87 result = cur.fetchone() 

88 result["created_at"] = TimeUTC.datetime_to_timestamp(result["created_at"]) 

89 return helper.dict_to_camel_case(result) 

90 

91 

92def generate_new_invitation(user_id): 

93 invitation_token = __generate_invitation_token() 

94 with pg_client.PostgresClient() as cur: 

95 query = cur.mogrify(""" \ 

96 UPDATE public.basic_authentication 

97 SET invitation_token = %(invitation_token)s, 

98 invited_at = timezone('utc'::text, now()), 

99 change_pwd_expire_at = NULL, 

100 change_pwd_token = NULL 

101 WHERE user_id = %(user_id)s RETURNING invitation_token;""", 

102 {"user_id": user_id, "invitation_token": invitation_token}) 

103 cur.execute( 

104 query 

105 ) 

106 return __get_invitation_link(cur.fetchone().pop("invitation_token")) 

107 

108 

109def reset_member(tenant_id, editor_id, user_id_to_update): 

110 admin = get_user(tenant_id=tenant_id, user_id=editor_id) 

111 if not admin["admin"] and not admin["superAdmin"]: 111 ↛ 112line 111 didn't jump to line 112 because the condition on line 111 was never true

112 return {"errors": ["unauthorized"]} 

113 user = get_user(tenant_id=tenant_id, user_id=user_id_to_update) 

114 if not user: 

115 return {"errors": ["user not found"]} 

116 return {"data": {"invitationLink": generate_new_invitation(user_id_to_update)}} 

117 

118 

119def update(tenant_id, user_id, changes, output=True): 

120 AUTH_KEYS = ["password", "invitationToken", "invitedAt", "changePwdExpireAt", "changePwdToken"] 

121 if len(changes.keys()) == 0: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true

122 return None 

123 

124 sub_query_users = [] 

125 sub_query_bauth = [] 

126 for key in changes.keys(): 

127 if key in AUTH_KEYS: 127 ↛ 128line 127 didn't jump to line 128 because the condition on line 127 was never true

128 if key == "password": 

129 sub_query_bauth.append("password = crypt(%(password)s, gen_salt('bf', 12))") 

130 sub_query_bauth.append("changed_at = timezone('utc'::text, now())") 

131 else: 

132 sub_query_bauth.append(f"{helper.key_to_snake_case(key)} = %({key})s") 

133 else: 

134 sub_query_users.append(f"{helper.key_to_snake_case(key)} = %({key})s") 

135 

136 with pg_client.PostgresClient() as cur: 

137 if len(sub_query_users) > 0: 137 ↛ 144line 137 didn't jump to line 144 because the condition on line 137 was always true

138 query = cur.mogrify(f"""\ 

139 UPDATE public.users 

140 SET {" ,".join(sub_query_users)} 

141 WHERE users.user_id = %(user_id)s;""", 

142 {"user_id": user_id, **changes}) 

143 cur.execute(query) 

144 if len(sub_query_bauth) > 0: 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true

145 query = cur.mogrify(f"""\ 

146 UPDATE public.basic_authentication 

147 SET {" ,".join(sub_query_bauth)} 

148 WHERE basic_authentication.user_id = %(user_id)s;""", 

149 {"user_id": user_id, **changes}) 

150 cur.execute(query) 

151 if not output: 

152 return None 

153 return get_user(user_id=user_id, tenant_id=tenant_id) 

154 

155 

156def create_member(tenant_id, user_id, data: schemas.CreateMemberSchema, background_tasks: BackgroundTasks): 

157 admin = get_user(tenant_id=tenant_id, user_id=user_id) 

158 if not admin["admin"] and not admin["superAdmin"]: 158 ↛ 159line 158 didn't jump to line 159 because the condition on line 158 was never true

159 return {"errors": ["unauthorized"]} 

160 if data.user_id is not None: 

161 return {"errors": ["please use POST/PUT /client/members/{memberId} for update"]} 

162 user = get_by_email_only(email=data.email) 

163 if user: 

164 return {"errors": ["user already exists"]} 

165 

166 if data.name is None or len(data.name) == 0: 

167 data.name = data.email 

168 invitation_token = __generate_invitation_token() 

169 user = get_deleted_user_by_email(email=data.email) 

170 if user is not None: 

171 new_member = restore_member(email=data.email, invitation_token=invitation_token, 

172 admin=data.admin, name=data.name, user_id=user["userId"]) 

173 else: 

174 new_member = create_new_member(email=data.email, invitation_token=invitation_token, 

175 admin=data.admin, name=data.name) 

176 new_member["invitationLink"] = __get_invitation_link(new_member.pop("invitationToken")) 

177 background_tasks.add_task(email_helper.send_team_invitation, **{ 

178 "recipient": data.email, 

179 "invitation_link": new_member["invitationLink"], 

180 "client_id": tenants.get_by_tenant_id(tenant_id)["name"], 

181 "sender_name": admin["name"] 

182 }) 

183 return {"data": new_member} 

184 

185 

186def __get_invitation_link(invitation_token): 

187 return config("SITE_URL") + config("invitation_link") % invitation_token 

188 

189 

190def allow_password_change(user_id, delta_min=10): 

191 pass_token = secrets.token_urlsafe(8) 

192 with pg_client.PostgresClient() as cur: 

193 query = cur.mogrify(f"""UPDATE public.basic_authentication  

194 SET change_pwd_expire_at = timezone('utc'::text, now()+INTERVAL '%(delta)s MINUTES'), 

195 change_pwd_token = %(pass_token)s 

196 WHERE user_id = %(user_id)s""", 

197 {"user_id": user_id, "delta": delta_min, "pass_token": pass_token}) 

198 cur.execute( 

199 query 

200 ) 

201 return pass_token 

202 

203 

204cache = TTLCache(maxsize=5000, ttl=config("USERS_CACHE_TTL_S", cast=int, default=60)) 

205 

206 

207@cached(cache) 

208def get_user(user_id, tenant_id): 

209 with pg_client.PostgresClient() as cur: 

210 cur.execute( 

211 cur.mogrify( 

212 f"""SELECT  

213 users.user_id, 

214 email,  

215 role,  

216 users.name, 

217 (CASE WHEN role = 'owner' THEN TRUE ELSE FALSE END) AS super_admin, 

218 (CASE WHEN role = 'admin' THEN TRUE ELSE FALSE END) AS admin, 

219 (CASE WHEN role = 'member' THEN TRUE ELSE FALSE END) AS member, 

220 TRUE AS has_password, 

221 settings 

222 FROM public.users  

223 WHERE users.user_id = %(userId)s 

224 AND deleted_at IS NULL 

225 LIMIT 1;""", 

226 {"userId": user_id}) 

227 ) 

228 r = cur.fetchone() 

229 result = helper.dict_to_camel_case(r) 

230 if result and isinstance(result, dict): 

231 if result.get("settings") is None or not isinstance(result.get("settings"), dict): 231 ↛ 232line 231 didn't jump to line 232 because the condition on line 231 was never true

232 result["settings"] = {} 

233 if not result["settings"].get("modules"): 

234 result["settings"]["modules"] = [] 

235 return result 

236 

237 

238def generate_new_api_key(user_id): 

239 with pg_client.PostgresClient() as cur: 

240 cur.execute( 

241 cur.mogrify( 

242 f"""UPDATE public.users 

243 SET api_key=generate_api_key(20) 

244 WHERE users.user_id = %(userId)s 

245 AND deleted_at IS NULL 

246 RETURNING api_key;""", 

247 {"userId": user_id}) 

248 ) 

249 r = cur.fetchone() 

250 return helper.dict_to_camel_case(r) 

251 

252 

253def __get_account_info(tenant_id, user_id): 

254 with pg_client.PostgresClient() as cur: 

255 cur.execute( 

256 cur.mogrify( 

257 f"""SELECT users.name,  

258 tenants.name AS tenant_name,  

259 tenants.opt_out 

260 FROM public.users INNER JOIN public.tenants ON(TRUE) 

261 WHERE users.user_id = %(userId)s 

262 AND users.deleted_at IS NULL;""", 

263 {"tenantId": tenant_id, "userId": user_id}) 

264 ) 

265 r = cur.fetchone() 

266 return helper.dict_to_camel_case(r) 

267 

268 

269def edit_account(user_id, tenant_id, changes: schemas.EditAccountSchema): 

270 if changes.opt_out is not None or changes.tenantName is not None and len(changes.tenantName) > 0: 

271 user = get_user(user_id=user_id, tenant_id=tenant_id) 

272 if not user["superAdmin"] and not user["admin"]: 272 ↛ 273line 272 didn't jump to line 273 because the condition on line 272 was never true

273 return {"errors": ["unauthorized"]} 

274 

275 if changes.name is not None and len(changes.name) > 0: 

276 update(tenant_id=tenant_id, user_id=user_id, changes={"name": changes.name}) 

277 

278 _tenant_changes = {} 

279 if changes.tenantName is not None and len(changes.tenantName) > 0: 

280 _tenant_changes["name"] = changes.tenantName 

281 

282 if changes.opt_out is not None: 

283 _tenant_changes["opt_out"] = changes.opt_out 

284 if len(_tenant_changes.keys()) > 0: 

285 tenants.edit_tenant(tenant_id=tenant_id, changes=_tenant_changes) 

286 

287 return {"data": __get_account_info(tenant_id=tenant_id, user_id=user_id)} 

288 

289 

290def edit_member(user_id_to_update, tenant_id, changes: schemas.EditMemberSchema, editor_id): 

291 user = get_member(user_id=user_id_to_update, tenant_id=tenant_id) 

292 _changes = {} 

293 if editor_id != user_id_to_update: 

294 admin = get_user_role(tenant_id=tenant_id, user_id=editor_id) 

295 if not admin["superAdmin"] and not admin["admin"]: 295 ↛ 296line 295 didn't jump to line 296 because the condition on line 295 was never true

296 return {"errors": ["unauthorized, you must have admin privileges"]} 

297 if admin["admin"] and user["superAdmin"]: 297 ↛ 298line 297 didn't jump to line 298 because the condition on line 297 was never true

298 return {"errors": ["only the owner can edit his own details"]} 

299 else: 

300 if user["superAdmin"]: 300 ↛ 302line 300 didn't jump to line 302 because the condition on line 300 was always true

301 changes.admin = None 

302 elif changes.admin != user["admin"]: 

303 return {"errors": ["cannot change your own admin privileges"]} 

304 

305 if changes.name and len(changes.name) > 0: 305 ↛ 308line 305 didn't jump to line 308 because the condition on line 305 was always true

306 _changes["name"] = changes.name 

307 

308 if changes.admin is not None: 

309 _changes["role"] = "admin" if changes.admin else "member" 

310 

311 if len(_changes.keys()) > 0: 311 ↛ 314line 311 didn't jump to line 314 because the condition on line 311 was always true

312 update(tenant_id=tenant_id, user_id=user_id_to_update, changes=_changes, output=False) 

313 return {"data": get_member(user_id=user_id_to_update, tenant_id=tenant_id)} 

314 return {"data": user} 

315 

316 

317def get_by_email_only(email): 

318 with pg_client.PostgresClient() as cur: 

319 cur.execute( 

320 cur.mogrify( 

321 f"""SELECT  

322 users.user_id, 

323 1 AS tenant_id, 

324 users.email,  

325 users.role,  

326 users.name, 

327 (CASE WHEN users.role = 'owner' THEN TRUE ELSE FALSE END) AS super_admin, 

328 (CASE WHEN users.role = 'admin' THEN TRUE ELSE FALSE END) AS admin, 

329 (CASE WHEN users.role = 'member' THEN TRUE ELSE FALSE END) AS member, 

330 TRUE AS has_password 

331 FROM public.users LEFT JOIN public.basic_authentication ON users.user_id=basic_authentication.user_id 

332 WHERE users.email = %(email)s  

333 AND users.deleted_at IS NULL 

334 LIMIT 1;""", 

335 {"email": email}) 

336 ) 

337 r = cur.fetchone() 

338 return helper.dict_to_camel_case(r) 

339 

340 

341def get_member(tenant_id, user_id): 

342 with pg_client.PostgresClient() as cur: 

343 cur.execute( 

344 cur.mogrify( 

345 f"""SELECT  

346 users.user_id, 

347 users.email,  

348 users.role,  

349 users.name,  

350 users.created_at, 

351 (CASE WHEN users.role = 'owner' THEN TRUE ELSE FALSE END) AS super_admin, 

352 (CASE WHEN users.role = 'admin' THEN TRUE ELSE FALSE END) AS admin, 

353 (CASE WHEN users.role = 'member' THEN TRUE ELSE FALSE END) AS member, 

354 DATE_PART('day',timezone('utc'::text, now()) \ 

355 - COALESCE(basic_authentication.invited_at,'2000-01-01'::timestamp ))>=1 AS expired_invitation, 

356 basic_authentication.password IS NOT NULL AS joined, 

357 invitation_token 

358 FROM public.users LEFT JOIN public.basic_authentication ON users.user_id=basic_authentication.user_id  

359 WHERE users.deleted_at IS NULL AND users.user_id=%(user_id)s 

360 ORDER BY name, user_id""", 

361 {"user_id": user_id}) 

362 ) 

363 u = helper.dict_to_camel_case(cur.fetchone()) 

364 if u: 

365 u["createdAt"] = TimeUTC.datetime_to_timestamp(u["createdAt"]) 

366 if u["invitationToken"]: 366 ↛ 369line 366 didn't jump to line 369 because the condition on line 366 was always true

367 u["invitationLink"] = __get_invitation_link(u.pop("invitationToken")) 

368 else: 

369 u["invitationLink"] = None 

370 

371 return u 

372 

373 

374def get_members(tenant_id): 

375 with pg_client.PostgresClient() as cur: 

376 cur.execute( 

377 f"""SELECT  

378 users.user_id, 

379 users.email,  

380 users.role,  

381 users.name,  

382 users.created_at, 

383 (CASE WHEN users.role = 'owner' THEN TRUE ELSE FALSE END) AS super_admin, 

384 (CASE WHEN users.role = 'admin' THEN TRUE ELSE FALSE END) AS admin, 

385 (CASE WHEN users.role = 'member' THEN TRUE ELSE FALSE END) AS member, 

386 DATE_PART('day',timezone('utc'::text, now()) \ 

387 - COALESCE(basic_authentication.invited_at,'2000-01-01'::timestamp ))>=1 AS expired_invitation, 

388 basic_authentication.password IS NOT NULL AS joined, 

389 invitation_token 

390 FROM public.users LEFT JOIN public.basic_authentication ON users.user_id=basic_authentication.user_id  

391 WHERE users.deleted_at IS NULL 

392 ORDER BY name, user_id""" 

393 ) 

394 r = cur.fetchall() 

395 if len(r): 395 ↛ 405line 395 didn't jump to line 405

396 r = helper.list_to_camel_case(r) 

397 for u in r: 

398 u["createdAt"] = TimeUTC.datetime_to_timestamp(u["createdAt"]) 

399 if u["invitationToken"]: 

400 u["invitationLink"] = __get_invitation_link(u.pop("invitationToken")) 

401 else: 

402 u["invitationLink"] = None 

403 return r 

404 

405 return [] 

406 

407 

408def transfer_ownership(tenant_id, user_id, new_owner_id): 

409 if user_id == new_owner_id: 

410 return {"errors": ["cannot transfer ownership to yourself"]} 

411 

412 current_owner = get_user_role(tenant_id=tenant_id, user_id=user_id) 

413 if current_owner is None or not current_owner["superAdmin"]: 413 ↛ 414line 413 didn't jump to line 414 because the condition on line 413 was never true

414 return {"errors": ["only the current owner can transfer ownership"]} 

415 

416 new_owner = get_member(tenant_id=tenant_id, user_id=new_owner_id) 

417 if new_owner is None: 

418 return {"errors": ["target user not found"]} 

419 

420 if not new_owner["joined"]: 420 ↛ 423line 420 didn't jump to line 423 because the condition on line 420 was always true

421 return {"errors": ["target user has not yet joined, they must accept their invitation first"]} 

422 

423 with pg_client.PostgresClient() as cur: 

424 cur.execute( 

425 cur.mogrify( 

426 """UPDATE public.users 

427 SET role = 'admin' 

428 WHERE user_id = %(current_owner_id)s 

429 AND role = 'owner' 

430 AND deleted_at IS NULL;""", 

431 {"current_owner_id": user_id})) 

432 if cur.rowcount == 0: 

433 return {"errors": ["ownership transfer failed, owner role may have already been transferred"]} 

434 cur.execute( 

435 cur.mogrify( 

436 """UPDATE public.users 

437 SET role = 'owner' 

438 WHERE user_id = %(new_owner_id)s 

439 AND role != 'owner' 

440 AND deleted_at IS NULL;""", 

441 {"new_owner_id": new_owner_id})) 

442 if cur.rowcount == 0: 

443 cur.execute( 

444 cur.mogrify( 

445 """UPDATE public.users 

446 SET role = 'owner' 

447 WHERE user_id = %(current_owner_id)s 

448 AND deleted_at IS NULL;""", 

449 {"current_owner_id": user_id})) 

450 return {"errors": ["ownership transfer failed, target user could not be promoted"]} 

451 

452 cache.pop((user_id, tenant_id), None) 

453 cache.pop((new_owner_id, tenant_id), None) 

454 

455 return {"data": get_member(tenant_id=tenant_id, user_id=new_owner_id)} 

456 

457 

458def delete_member(user_id, tenant_id, id_to_delete): 

459 if user_id == id_to_delete: 

460 return {"errors": ["unauthorized, cannot delete self"]} 

461 

462 admin = get_user(user_id=user_id, tenant_id=tenant_id) 

463 if admin["member"]: 463 ↛ 464line 463 didn't jump to line 464 because the condition on line 463 was never true

464 return {"errors": ["unauthorized"]} 

465 

466 to_delete = get_user(user_id=id_to_delete, tenant_id=tenant_id) 

467 if to_delete is None: 

468 return {"errors": ["not found"]} 

469 

470 if to_delete["superAdmin"]: 470 ↛ 471line 470 didn't jump to line 471 because the condition on line 470 was never true

471 return {"errors": ["cannot delete super admin"]} 

472 

473 with pg_client.PostgresClient() as cur: 

474 cur.execute( 

475 cur.mogrify(f"""UPDATE public.users 

476 SET deleted_at = timezone('utc'::text, now()), 

477 jwt_iat= NULL, jwt_refresh_jti= NULL,  

478 jwt_refresh_iat= NULL  

479 WHERE user_id=%(user_id)s;""", 

480 {"user_id": id_to_delete})) 

481 cur.execute( 

482 cur.mogrify(f"""UPDATE public.basic_authentication 

483 SET password= NULL, invitation_token= NULL, 

484 invited_at= NULL, changed_at= NULL, 

485 change_pwd_expire_at= NULL, change_pwd_token= NULL 

486 WHERE user_id=%(user_id)s;""", 

487 {"user_id": id_to_delete})) 

488 return {"data": get_members(tenant_id=tenant_id)} 

489 

490 

491def change_password(tenant_id, user_id, email, old_password, new_password): 

492 item = get_user(tenant_id=tenant_id, user_id=user_id) 

493 if item is None: 

494 return {"errors": ["access denied"]} 

495 if old_password == new_password: 

496 return {"errors": ["old and new password are the same"]} 

497 auth = authenticate(email, old_password, for_change_password=True) 

498 if auth is None: 

499 return {"errors": ["wrong password"]} 

500 changes = {"password": new_password} 

501 user = update(tenant_id=tenant_id, user_id=user_id, changes=changes) 

502 r = authenticate(user['email'], new_password) 

503 

504 return { 

505 "jwt": r.pop("jwt"), 

506 "refreshToken": r.pop("refreshToken"), 

507 "refreshTokenMaxAge": r.pop("refreshTokenMaxAge"), 

508 "spotJwt": r.pop("spotJwt"), 

509 "spotRefreshToken": r.pop("spotRefreshToken"), 

510 "spotRefreshTokenMaxAge": r.pop("spotRefreshTokenMaxAge") 

511 } 

512 

513 

514def set_password_invitation(user_id, new_password): 

515 changes = {"password": new_password} 

516 user = update(tenant_id=-1, user_id=user_id, changes=changes) 

517 r = authenticate(user['email'], new_password) 

518 

519 return { 

520 "jwt": r.pop("jwt"), 

521 "refreshToken": r.pop("refreshToken"), 

522 "refreshTokenMaxAge": r.pop("refreshTokenMaxAge"), 

523 "spotJwt": r.pop("spotJwt"), 

524 "spotRefreshToken": r.pop("spotRefreshToken"), 

525 "spotRefreshTokenMaxAge": r.pop("spotRefreshTokenMaxAge"), 

526 **r 

527 } 

528 

529 

530def email_exists(email): 

531 with pg_client.PostgresClient() as cur: 

532 cur.execute( 

533 cur.mogrify( 

534 f"""SELECT  

535 count(user_id)  

536 FROM public.users 

537 WHERE 

538 email = %(email)s 

539 AND deleted_at IS NULL 

540 LIMIT 1;""", 

541 {"email": email}) 

542 ) 

543 r = cur.fetchone() 

544 return r["count"] > 0 

545 

546 

547def get_deleted_user_by_email(email): 

548 with pg_client.PostgresClient() as cur: 

549 cur.execute( 

550 cur.mogrify( 

551 f"""SELECT  

552 *  

553 FROM public.users 

554 WHERE 

555 email = %(email)s 

556 AND deleted_at NOTNULL 

557 LIMIT 1;""", 

558 {"email": email}) 

559 ) 

560 r = cur.fetchone() 

561 return helper.dict_to_camel_case(r) 

562 

563 

564def get_by_invitation_token(token, pass_token=None): 

565 with pg_client.PostgresClient() as cur: 

566 cur.execute( 

567 cur.mogrify( 

568 f"""SELECT  

569 *, 

570 DATE_PART('day',timezone('utc'::text, now()) \ 

571 - COALESCE(basic_authentication.invited_at,'2000-01-01'::timestamp ))>=1 AS expired_invitation, 

572 change_pwd_expire_at <= timezone('utc'::text, now()) AS expired_change, 

573 (EXTRACT(EPOCH FROM current_timestamp-basic_authentication.change_pwd_expire_at))::BIGINT AS change_pwd_age 

574 FROM public.users INNER JOIN public.basic_authentication USING(user_id) 

575 WHERE invitation_token = %(token)s {"AND change_pwd_token = %(pass_token)s" if pass_token else ""} 

576 LIMIT 1;""", 

577 {"token": token, "pass_token": pass_token}) 

578 ) 

579 r = cur.fetchone() 

580 return helper.dict_to_camel_case(r) 

581 

582 

583def auth_exists(user_id, jwt_iat) -> bool: 

584 with pg_client.PostgresClient() as cur: 

585 cur.execute( 

586 cur.mogrify(f"""SELECT user_id, EXTRACT(epoch FROM jwt_iat)::BIGINT AS jwt_iat  

587 FROM public.users  

588 WHERE user_id = %(userId)s  

589 AND deleted_at IS NULL 

590 LIMIT 1;""", 

591 {"userId": user_id}) 

592 ) 

593 r = cur.fetchone() 

594 return r is not None \ 

595 and r.get("jwt_iat") is not None \ 

596 and abs(jwt_iat - r["jwt_iat"]) <= 1 

597 

598 

599def refresh_auth_exists(user_id, jwt_jti=None): 

600 with pg_client.PostgresClient() as cur: 

601 cur.execute( 

602 cur.mogrify(f"""SELECT user_id  

603 FROM public.users  

604 WHERE user_id = %(userId)s  

605 AND deleted_at IS NULL 

606 AND jwt_refresh_jti = %(jwt_jti)s 

607 LIMIT 1;""", 

608 {"userId": user_id, "jwt_jti": jwt_jti}) 

609 ) 

610 r = cur.fetchone() 

611 return r is not None 

612 

613 

614class FullLoginJWTs(BaseModel): 

615 jwt_iat: int 

616 jwt_refresh_jti: str 

617 jwt_refresh_iat: int 

618 spot_jwt_iat: int 

619 spot_jwt_refresh_jti: str 

620 spot_jwt_refresh_iat: int 

621 

622 @model_validator(mode="before") 

623 @classmethod 

624 def _transform_data(cls, values): 

625 if values.get("jwt_refresh_jti") is not None: 625 ↛ 627line 625 didn't jump to line 627 because the condition on line 625 was always true

626 values["jwt_refresh_jti"] = str(values["jwt_refresh_jti"]) 

627 if values.get("spot_jwt_refresh_jti") is not None: 627 ↛ 629line 627 didn't jump to line 629 because the condition on line 627 was always true

628 values["spot_jwt_refresh_jti"] = str(values["spot_jwt_refresh_jti"]) 

629 return values 

630 

631 

632class RefreshLoginJWTs(FullLoginJWTs): 

633 spot_jwt_iat: Optional[int] = None 

634 spot_jwt_refresh_jti: Optional[str] = None 

635 spot_jwt_refresh_iat: Optional[int] = None 

636 

637 

638class RefreshSpotJWTs(FullLoginJWTs): 

639 jwt_iat: Optional[int] = None 

640 jwt_refresh_jti: Optional[str] = None 

641 jwt_refresh_iat: Optional[int] = None 

642 

643 

644def change_jwt_iat_jti(user_id): 

645 with pg_client.PostgresClient() as cur: 

646 query = cur.mogrify(f"""UPDATE public.users 

647 SET jwt_iat = timezone('utc'::text, now()-INTERVAL '10s'), 

648 jwt_refresh_jti = 0,  

649 jwt_refresh_iat = timezone('utc'::text, now()-INTERVAL '10s'), 

650 spot_jwt_iat = timezone('utc'::text, now()-INTERVAL '10s'), 

651 spot_jwt_refresh_jti = 0,  

652 spot_jwt_refresh_iat = timezone('utc'::text, now()-INTERVAL '10s') 

653 WHERE user_id = %(user_id)s  

654 RETURNING EXTRACT (epoch FROM jwt_iat)::BIGINT AS jwt_iat,  

655 jwt_refresh_jti,  

656 EXTRACT (epoch FROM jwt_refresh_iat)::BIGINT AS jwt_refresh_iat, 

657 EXTRACT (epoch FROM spot_jwt_iat)::BIGINT AS spot_jwt_iat,  

658 spot_jwt_refresh_jti,  

659 EXTRACT (epoch FROM spot_jwt_refresh_iat)::BIGINT AS spot_jwt_refresh_iat;""", 

660 {"user_id": user_id}) 

661 cur.execute(query) 

662 row = cur.fetchone() 

663 return FullLoginJWTs(**row) 

664 

665 

666def refresh_jwt_iat_jti(user_id): 

667 with pg_client.PostgresClient() as cur: 

668 query = cur.mogrify(f"""UPDATE public.users 

669 SET jwt_iat = timezone('utc'::text, now()-INTERVAL '10s'), 

670 jwt_refresh_jti = jwt_refresh_jti + 1  

671 WHERE user_id = %(user_id)s  

672 RETURNING EXTRACT (epoch FROM jwt_iat)::BIGINT AS jwt_iat,  

673 jwt_refresh_jti,  

674 EXTRACT (epoch FROM jwt_refresh_iat)::BIGINT AS jwt_refresh_iat;""", 

675 {"user_id": user_id}) 

676 cur.execute(query) 

677 row = cur.fetchone() 

678 return RefreshLoginJWTs(**row) 

679 

680 

681def authenticate(email, password, for_change_password=False) -> dict | bool | None: 

682 with pg_client.PostgresClient() as cur: 

683 query = cur.mogrify( 

684 f"""SELECT  

685 users.user_id, 

686 1 AS tenant_id, 

687 users.role, 

688 users.name, 

689 (CASE WHEN users.role = 'owner' THEN TRUE ELSE FALSE END) AS super_admin, 

690 (CASE WHEN users.role = 'admin' THEN TRUE ELSE FALSE END) AS admin, 

691 (CASE WHEN users.role = 'member' THEN TRUE ELSE FALSE END) AS member 

692 FROM public.users INNER JOIN public.basic_authentication USING(user_id) 

693 WHERE users.email = %(email)s  

694 AND basic_authentication.password = crypt(%(password)s, basic_authentication.password) 

695 AND basic_authentication.user_id = (SELECT su.user_id FROM public.users AS su WHERE su.email=%(email)s AND su.deleted_at IS NULL LIMIT 1) 

696 LIMIT 1;""", 

697 {"email": email, "password": password}) 

698 

699 cur.execute(query) 

700 r = cur.fetchone() 

701 

702 if r is not None: 

703 if for_change_password: 703 ↛ 704line 703 didn't jump to line 704 because the condition on line 703 was never true

704 return True 

705 r = helper.dict_to_camel_case(r) 

706 j_r = change_jwt_iat_jti(user_id=r['userId']) 

707 response = { 

708 "jwt": authorizers.generate_jwt(user_id=r['userId'], tenant_id=r['tenantId'], iat=j_r.jwt_iat, 

709 aud=AUDIENCE), 

710 "refreshToken": authorizers.generate_jwt_refresh(user_id=r['userId'], 

711 tenant_id=r['tenantId'], 

712 iat=j_r.jwt_refresh_iat, 

713 aud=AUDIENCE, 

714 jwt_jti=j_r.jwt_refresh_jti, 

715 for_spot=False), 

716 "refreshTokenMaxAge": config("JWT_REFRESH_EXPIRATION", cast=int), 

717 "email": email, 

718 "spotJwt": authorizers.generate_jwt(user_id=r['userId'], tenant_id=r['tenantId'], 

719 iat=j_r.spot_jwt_iat, aud=spot.AUDIENCE, for_spot=True), 

720 "spotRefreshToken": authorizers.generate_jwt_refresh(user_id=r['userId'], 

721 tenant_id=r['tenantId'], 

722 iat=j_r.spot_jwt_refresh_iat, 

723 aud=spot.AUDIENCE, 

724 jwt_jti=j_r.spot_jwt_refresh_jti, 

725 for_spot=True), 

726 "spotRefreshTokenMaxAge": config("JWT_SPOT_REFRESH_EXPIRATION", cast=int), 

727 **r 

728 } 

729 return response 

730 

731 return None 

732 

733 

734def logout(user_id: int): 

735 with pg_client.PostgresClient() as cur: 

736 query = cur.mogrify( 

737 """UPDATE public.users 

738 SET jwt_iat = NULL, 

739 jwt_refresh_jti = NULL, 

740 jwt_refresh_iat = NULL, 

741 spot_jwt_iat = NULL, 

742 spot_jwt_refresh_jti = NULL, 

743 spot_jwt_refresh_iat = NULL 

744 WHERE user_id = %(user_id)s;""", 

745 {"user_id": user_id}) 

746 cur.execute(query) 

747 

748 

749def refresh(user_id: int, tenant_id: int = -1) -> dict: 

750 j = refresh_jwt_iat_jti(user_id=user_id) 

751 return { 

752 "jwt": authorizers.generate_jwt(user_id=user_id, tenant_id=tenant_id, iat=j.jwt_iat, 

753 aud=AUDIENCE), 

754 "refreshToken": authorizers.generate_jwt_refresh(user_id=user_id, tenant_id=tenant_id, iat=j.jwt_refresh_iat, 

755 aud=AUDIENCE, jwt_jti=j.jwt_refresh_jti), 

756 "refreshTokenMaxAge": config("JWT_REFRESH_EXPIRATION", cast=int) - (j.jwt_iat - j.jwt_refresh_iat), 

757 } 

758 

759 

760def get_user_role(tenant_id, user_id): 

761 with pg_client.PostgresClient() as cur: 

762 cur.execute( 

763 cur.mogrify( 

764 f"""SELECT  

765 users.user_id, 

766 users.email,  

767 users.role,  

768 users.name,  

769 users.created_at, 

770 (CASE WHEN users.role = 'owner' THEN TRUE ELSE FALSE END) AS super_admin, 

771 (CASE WHEN users.role = 'admin' THEN TRUE ELSE FALSE END) AS admin, 

772 (CASE WHEN users.role = 'member' THEN TRUE ELSE FALSE END) AS member 

773 FROM public.users  

774 WHERE users.deleted_at IS NULL  

775 AND users.user_id=%(user_id)s 

776 LIMIT 1""", 

777 {"user_id": user_id}) 

778 ) 

779 return helper.dict_to_camel_case(cur.fetchone()) 

780 

781 

782def get_user_settings(user_id): 

783 # read user settings from users.settings:jsonb column 

784 with pg_client.PostgresClient() as cur: 

785 cur.execute( 

786 cur.mogrify( 

787 f"""SELECT  

788 settings 

789 FROM public.users  

790 WHERE users.deleted_at IS NULL  

791 AND users.user_id=%(user_id)s 

792 LIMIT 1""", 

793 {"user_id": user_id}) 

794 ) 

795 return helper.dict_to_camel_case(cur.fetchone()) 

796 

797 

798def update_user_module(user_id, data: schemas.ModuleStatus): 

799 # example data = {"settings": {"modules": ['ASSIST', 'METADATA']} 

800 # update user settings from users.settings:jsonb column only update settings.modules 

801 # if module property is not exists, it will be created 

802 # if module property exists, it will be updated, modify here and call update_user_settings 

803 # module is a single element to be added or removed 

804 user_settings = get_user_settings(user_id) 

805 if user_settings is None: 805 ↛ 806line 805 didn't jump to line 806 because the condition on line 805 was never true

806 settings = {} 

807 else: 

808 settings = user_settings.get("settings") 

809 if settings is None or not isinstance(settings, dict): 809 ↛ 810line 809 didn't jump to line 810 because the condition on line 809 was never true

810 settings = {} 

811 

812 if settings.get("modules") is None: 812 ↛ 813line 812 didn't jump to line 813 because the condition on line 812 was never true

813 settings["modules"] = [] 

814