Coverage for opt/mealie/lib/python3.12/site-packages/mealie/core/security/providers/auth_provider.py: 87%

1import abc 

2from datetime import UTC, datetime, timedelta 

3 

4import jwt 

5from sqlalchemy.orm.session import Session 

6 

7from mealie.core.config import get_app_settings 

8from mealie.repos.all_repositories import get_repositories 

9from mealie.schema.user.user import PrivateUser 

10 

11ALGORITHM = "HS256" 

12ISS = "mealie" 

13remember_me_duration = timedelta(days=14) 

14 

15 

16class AuthProvider[T](metaclass=abc.ABCMeta): 

17 """Base Authentication Provider interface""" 

18 

19 def __init__(self, session: Session, data: T) -> None: 

20 self.session = session 

21 self.data = data 

22 self.user: PrivateUser | None = None 

23 self.__has_tried_user = False 

24 

25 @classmethod 

26 def __subclasshook__(cls, __subclass: type) -> bool: 

27 return hasattr(__subclass, "authenticate") and callable(__subclass.authenticate) 

28 

29 def get_access_token(self, user: PrivateUser, remember_me=False) -> tuple[str, timedelta]: 

30 settings = get_app_settings() 

31 

32 duration = timedelta(hours=settings.TOKEN_TIME) 

33 if remember_me: 33 ↛ 34line 33 didn't jump to line 34 because the condition on line 33 was never true

34 duration = max(remember_me_duration, duration) 

35 

36 return AuthProvider.create_access_token({"sub": str(user.id)}, duration) 

37 

38 @staticmethod 

39 def create_access_token(data: dict, expires_delta: timedelta | None = None) -> tuple[str, timedelta]: 

40 settings = get_app_settings() 

41 

42 to_encode = data.copy() 

43 expires_delta = expires_delta or timedelta(hours=settings.TOKEN_TIME) 

44 

45 expire = datetime.now(UTC) + expires_delta 

46 

47 to_encode["exp"] = expire 

48 to_encode["iss"] = ISS 

49 return ( 

50 jwt.encode(to_encode, settings.SECRET, algorithm=ALGORITHM), 

51 expires_delta, 

52 ) 

53 

54 def try_get_user(self, username: str) -> PrivateUser | None: 

55 """Try to get a user from the database, first trying username, then trying email""" 

56 if self.__has_tried_user: 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true

57 return self.user 

58 

59 db = get_repositories(self.session, group_id=None, household_id=None) 

60 

61 user = user = db.users.get_one(username, "username", any_case=True) 

62 if not user: 62 ↛ 65line 62 didn't jump to line 65 because the condition on line 62 was always true

63 user = db.users.get_one(username, "email", any_case=True) 

64 

65 self.user = user 

66 return user 

67 

68 @abc.abstractmethod 

69 def authenticate(self) -> tuple[str, timedelta] | None: 

70 """Attempt to authenticate a user""" 

71 raise NotImplementedError