import logging import time from typing import Optional, Dict, Any, Tuple import jwt from jwt import PyJWTError from app.classes.models.users import HelperUsers, ApiKeys logger = logging.getLogger(__name__) class Authentication: def __init__(self, helper): self.helper = helper self.secret = "my secret" self.secret = self.helper.get_setting("apikey_secret", None) if self.secret is None or self.secret == "random": self.secret = self.helper.random_string_generator(64) self.helper.set_setting("apikey_secret", self.secret) def generate(self, user_id, extra=None): if extra is None: extra = {} jwt_encoded = jwt.encode( {"user_id": user_id, "iat": int(time.time()), **extra}, self.secret, algorithm="HS256", ) return jwt_encoded def read(self, token): return jwt.decode(token, self.secret, algorithms=["HS256"]) def check_no_iat(self, token) -> Optional[Dict[str, Any]]: try: return jwt.decode(token, self.secret, algorithms=["HS256"]) except PyJWTError as error: logger.debug("Error while checking JWT token: ", exc_info=error) return None def check( self, token, ) -> Optional[Tuple[Optional[ApiKeys], Dict[str, Any], Dict[str, Any]]]: try: data = jwt.decode(token, self.secret, algorithms=["HS256"]) except PyJWTError as error: logger.debug("Error while checking JWT token: ", exc_info=error) return None iat: int = data["iat"] key: Optional[ApiKeys] = None if "token_id" in data: key_id = data["token_id"] key = HelperUsers.get_user_api_key(key_id) if key is None: return None user_id: str = data["user_id"] user = HelperUsers.get_user(user_id) # TODO: Have a cache or something so we don't constantly # have to query the database if int(user.get("valid_tokens_from").timestamp()) < iat: # Success! return key, data, user else: return None def check_bool(self, token) -> bool: return self.check(token) is not None