theo-agent-dashboard

Unnamed repository; edit this file 'description' to name the repository.
Log | Files | Refs | README

auth.py (2527B)


      1 """Authentication: password hashing and JWT management."""
      2 
      3 import os
      4 import secrets
      5 import time
      6 
      7 import jwt
      8 from argon2 import PasswordHasher
      9 from argon2.exceptions import VerifyMismatchError
     10 
     11 _HASHING_INTERVAL = 86400  # 24 hours in seconds
     12 _ALGORITHM = "HS256"
     13 
     14 _ph = PasswordHasher()
     15 
     16 
     17 def _get_secret_key() -> str:  # pragma: no mutate: block
     18     """Return a stable secret key (from env or generated)."""
     19     return os.environ.get("THEO_JWT_SECRET", _load_or_create_secret())
     20 
     21 
     22 def _load_or_create_secret() -> str:  # pragma: no mutate: block
     23     """Load persisted secret or create a new one."""
     24     secret_path = os.path.join(
     25         os.environ.get("THEO_DATA_DIR", "data"), ".jwt_secret"
     26     )
     27     try:
     28         return open(secret_path).read().strip()
     29     except FileNotFoundError:
     30         os.makedirs(os.path.dirname(secret_path), exist_ok=True)
     31         secret = secrets.token_hex(32)
     32         open(secret_path, "w").write(secret)
     33         return secret
     34 
     35 
     36 def hash_password(password: str) -> str:  # pragma: no mutate: block
     37     """Hash a password with argon2id."""
     38     return _ph.hash(password)
     39 
     40 
     41 def verify_password(password: str, hashed: str) -> bool:  # pragma: no mutate: block
     42     """Verify a password against its hash."""
     43     try:
     44         return _ph.verify(hashed, password)
     45     except VerifyMismatchError:
     46         return False
     47 
     48 
     49 def create_jwt(  # pragma: no mutate: block
     50     payload: dict,
     51     expires_in_seconds: int = _HASHING_INTERVAL,
     52 ) -> str:  # pragma: no mutate: block
     53     """Create a signed JWT token."""
     54     to_encode = payload.copy()
     55     to_encode["exp"] = int(time.time()) + expires_in_seconds
     56     return jwt.encode(to_encode, _get_secret_key(), algorithm=_ALGORITHM)
     57 
     58 
     59 def verify_jwt(token: str) -> dict:  # pragma: no mutate: block
     60     """Decode and verify a JWT, raising on expiry or invalidity."""
     61     return jwt.decode(token, _get_secret_key(), algorithms=[_ALGORITHM])
     62 
     63 
     64 def set_jwt_cookie(response, token: str) -> None:  # pragma: no mutate: block
     65     """Set JWT as an HttpOnly cookie on a Response object."""
     66     response.set_cookie(
     67         key="theo_token",
     68         value=token,
     69         httponly=True,
     70         samesite="lax",
     71         max_age=_HASHING_INTERVAL,
     72     )
     73 
     74 
     75 def get_jwt_from_cookie(request) -> str | None:  # pragma: no mutate: block
     76     """Extract the JWT from the request cookie."""
     77     return request.cookies.get("theo_token")
     78 
     79 
     80 def clear_jwt_cookie(response) -> None:  # pragma: no mutate: block
     81     """Delete the JWT cookie."""
     82     response.delete_cookie(key="theo_token")