Class HashService

src/core/service/hash_service.py:16–153

class HashService:
    def __init__(self, cfg: Settings) -> None:
        self.cfg = cfg
        self.jwt_public_key = base64.b64decode(cfg.jwt_public_key).decode("utf-8")
        self.jwt_private_key = base64.b64decode(cfg.jwt_private_key).decode("utf-8")
        self.jwt_algorithm = cfg.jwt_algorithm
        if self.jwt_algorithm is None or self.jwt_algorithm == "":
            self.jwt_algorithm = "RS256"
        self.hasher = PasswordHasher()

    def verify_x_api_key(self, key: str) -> bool:
        return self.verify_hash(key, self.cfg.x_api_key)

    @staticmethod
    def verify_hash(hash1: str, hash2: str) -> bool:
        return secrets.compare_digest(hash1, hash2)

    @staticmethod
    def uuid4() -> uuid.UUID:
        return uuid.uuid4()

    def hash_password(self, password: str) -> str:
        return self.hasher.hash(password)

    def verify_password(self, password: str, hashed_password: str) -> bool:
        try:
            self.hasher.verify(hashed_password, password)
            return True
        except VerifyMismatchError:
            return False

    def create_access_token(self, user: User) -> Token:
        exp = datetime.now() + timedelta(hours=self.cfg.jwt_access_token_exp_h)
        payload = {
            "sub": str(user.id),
            "type": TokenType.ACCESS,
            "session": user.session,
            "exp": int(exp.timestamp()),
        }

        token = jwt.encode(payload, self.jwt_private_key, algorithm=self.jwt_algorithm)

        return Token(
            subject=str(user.id),
            token=str(token),
            token_type=TokenType.ACCESS,
            expires_in=exp,
        )

    def create_refresh_token(self, user: User) -> Token:
        exp = datetime.now() + timedelta(hours=self.cfg.jwt_refresh_token_exp_h)
        payload = {
            "sub": str(user.id),
            "type": TokenType.REFRESH,
            "session": user.session,
            "exp": int(exp.timestamp()),
        }