Coverage for chalicelib/core/spot.py: 40%

1from decouple import config 

2 

3from chalicelib.core import authorizers, users 

4from chalicelib.utils import pg_client 

5 

6AUDIENCE = "spot:OpenReplay" 

7 

8 

9def refresh_spot_jwt_iat_jti(user_id): 

10 with pg_client.PostgresClient() as cur: 

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

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

13 spot_jwt_refresh_jti = spot_jwt_refresh_jti + 1  

14 WHERE user_id = %(user_id)s  

15 RETURNING EXTRACT (epoch FROM spot_jwt_iat)::BIGINT AS spot_jwt_iat,  

16 spot_jwt_refresh_jti,  

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

18 {"user_id": user_id}) 

19 cur.execute(query) 

20 row = cur.fetchone() 

21 return users.RefreshSpotJWTs(**row) 

22 

23 

24def logout(user_id: int): 

25 users.logout(user_id=user_id) 

26 

27 

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

29 j = refresh_spot_jwt_iat_jti(user_id=user_id) 

30 return { 

31 "jwt": authorizers.generate_jwt(user_id=user_id, tenant_id=tenant_id, iat=j.spot_jwt_iat, 

32 aud=AUDIENCE, for_spot=True), 

33 "refreshToken": authorizers.generate_jwt_refresh(user_id=user_id, tenant_id=tenant_id, iat=j.spot_jwt_refresh_iat, 

34 aud=AUDIENCE, jwt_jti=j.spot_jwt_refresh_jti, for_spot=True), 

35 "refreshTokenMaxAge": config("JWT_SPOT_REFRESH_EXPIRATION", cast=int) - (j.spot_jwt_iat - j.spot_jwt_refresh_iat) 

36 } 

37 

38 

39def refresh_auth_exists(user_id, jwt_jti): 

40 with pg_client.PostgresClient() as cur: 

41 cur.execute( 

42 cur.mogrify(f"""SELECT user_id  

43 FROM public.users  

44 WHERE user_id = %(userId)s  

45 AND deleted_at IS NULL 

46 AND spot_jwt_refresh_jti = %(jwt_jti)s 

47 LIMIT 1;""", 

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

49 ) 

50 r = cur.fetchone() 

51 return r is not None