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

1import abc 1 ctx1p

2from datetime import UTC, datetime, timedelta 1 ctx1p

3 

4import jwt 1 ctx1p

5from sqlalchemy.orm.session import Session 1 ctx1p

6 

7from mealie.core.config import get_app_settings 1 ctx1p

8from mealie.repos.all_repositories import get_repositories 1 ctx1p

9from mealie.schema.user.user import PrivateUser 1 ctx1p

10 

11ALGORITHM = "HS256" 1 ctx1p

12ISS = "mealie" 1 ctx1p

13remember_me_duration = timedelta(days=14) 1 ctx1p

14 

15 

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

17 """Base Authentication Provider interface""" 

18 

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

20 self.session = session 46 ctx1efgqrshijklmntauvwxyzbABCDEFGcHIJKLMNoOPQRSTUd

21 self.data = data 46 ctx1efgqrshijklmntauvwxyzbABCDEFGcHIJKLMNoOPQRSTUd

22 self.user: PrivateUser | None = None 46 ctx1efgqrshijklmntauvwxyzbABCDEFGcHIJKLMNoOPQRSTUd

23 self.__has_tried_user = False 46 ctx1efgqrshijklmntauvwxyzbABCDEFGcHIJKLMNoOPQRSTUd

24 

25 @classmethod 1 ctx1p

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

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

28 

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

30 settings = get_app_settings() 15 ctx1efghijklmnabcod

31 

32 duration = timedelta(hours=settings.TOKEN_TIME) 15 ctx1efghijklmnabcod

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

34 duration = max(remember_me_duration, duration) 

35 

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

37 

38 @staticmethod 1 ctx1p

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

40 settings = get_app_settings() 15 ctx1efghijklmnabcod

41 

42 to_encode = data.copy() 15 ctx1efghijklmnabcod

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

44 

45 expire = datetime.now(UTC) + expires_delta 15 ctx1efghijklmnabcod

46 

47 to_encode["exp"] = expire 15 ctx1efghijklmnabcod

48 to_encode["iss"] = ISS 15 ctx1efghijklmnabcod

49 return ( 15 ctx1efghijklmnabcod

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

51 expires_delta, 

52 ) 

53 

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

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 true46 ctx1efgqrshijklmntauvwxyzbABCDEFGcHIJKLMNoOPQRSTUd

57 return self.user 

58 

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

60 

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

62 if not user: 46 ctx1efgqrshijklmntauvwxyzbABCDEFGcHIJKLMNoOPQRSTUd

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

64 

65 self.user = user 46 ctx1efgqrshijklmntauvwxyzbABCDEFGcHIJKLMNoOPQRSTUd

66 return user 46 ctx1efgqrshijklmntauvwxyzbABCDEFGcHIJKLMNoOPQRSTUd

67 

68 @abc.abstractmethod 1 ctx1p

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

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

71 raise NotImplementedError