mirror of
https://gitlab.com/crafty-controller/crafty-4.git
synced 2024-08-30 18:23:09 +00:00
70 lines
2.2 KiB
Python
70 lines
2.2 KiB
Python
import logging
|
|
import time
|
|
from typing import Optional, Dict, Any, Tuple
|
|
import jwt
|
|
from jwt import PyJWTError
|
|
|
|
from app.classes.models.users import helper_users, 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 = helper_users.get_user_api_key(key_id)
|
|
if key is None:
|
|
return None
|
|
user_id: str = data["user_id"]
|
|
user = helper_users.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
|