Coverage for yaptide/routes/utils/tokens.py: 78%

1from datetime import datetime, timedelta 

2from secrets import token_hex 

3from typing import Union 

4 

5import jwt 

6 

7SECRET_KEY_TOKEN = token_hex(256) 

8SECRET_KEY_TOKEN_REFRESH = token_hex(256) 

9_Refresh_Token_Expiration_Time = 120 # minutes 

10_Access_Token_Expiration_Time = 10 # minutes 

11_Keycloak_Token_Expiration_Time = 30 # minutes 

12 

13 

14def encode_auth_token(user_id: int, 

15 is_refresh: bool = False, 

16 is_keycloak: bool = False) -> tuple[Union[str, Exception], datetime]: # skipcq: FLK-E101 

17 """Function encoding the token""" 

18 if is_refresh: 

19 secret = SECRET_KEY_TOKEN_REFRESH 

20 exp_time_minutes = _Refresh_Token_Expiration_Time 

21 else: 

22 secret = SECRET_KEY_TOKEN 

23 exp_time_minutes = _Keycloak_Token_Expiration_Time if is_keycloak else _Access_Token_Expiration_Time 

24 

25 exp = datetime.utcnow() + timedelta(minutes=exp_time_minutes) 

26 

27 try: 

28 # For a description of the payload fields, take look 

29 # at JSON Web Token RFC https://datatracker.ietf.org/doc/html/rfc7519 

30 payload = { 

31 'exp': exp, # Token Expiration Time 

32 'iat': datetime.utcnow(), # Issued At Time 

33 'sub': user_id # Subject 

34 } 

35 return jwt.encode(payload, secret, algorithm='HS256'), exp 

36 except Exception as e: # skipcq: PYL-W0703 

37 return e, exp 

38 

39 

40def decode_auth_token(token: str, 

41 is_refresh: bool = False) -> Union[int, str]: 

42 """Function decoding the token""" 

43 if is_refresh: 

44 secret = SECRET_KEY_TOKEN_REFRESH 

45 else: 

46 secret = SECRET_KEY_TOKEN 

47 

48 try: 

49 payload = jwt.decode(token, secret, algorithms=['HS256']) 

50 return payload['sub'] 

51 except jwt.ExpiredSignatureError: 

52 return 'Signature expired.' 

53 except jwt.InvalidTokenError: 

54 return 'Invalid token.'