Frank Schwenk
40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
import hmac
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
import jwt
|
|
from fastapi import Depends, HTTPException, Request
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
|
|
from server.settings import settings
|
|
|
|
_bearer = HTTPBearer(auto_error=False)
|
|
ALGORITHM = "HS256"
|
|
|
|
|
|
def verify_pin(pin: str) -> bool:
|
|
return hmac.compare_digest(pin.strip(), settings.app_pin.strip())
|
|
|
|
|
|
def create_token() -> str:
|
|
expire = datetime.now(timezone.utc) + timedelta(days=settings.jwt_expire_days)
|
|
payload = {"sub": "user", "exp": expire}
|
|
return jwt.encode(payload, settings.jwt_secret, algorithm=ALGORITHM)
|
|
|
|
|
|
def decode_token(token: str) -> dict[str, Any]:
|
|
try:
|
|
return jwt.decode(token, settings.jwt_secret, algorithms=[ALGORITHM])
|
|
except jwt.PyJWTError as exc:
|
|
raise HTTPException(status_code=401, detail="Invalid token") from exc
|
|
|
|
|
|
async def require_auth(
|
|
creds: HTTPAuthorizationCredentials | None = Depends(_bearer),
|
|
) -> dict[str, Any]:
|
|
if creds is None or creds.scheme.lower() != "bearer":
|
|
raise HTTPException(status_code=401, detail="Missing token")
|
|
return decode_token(creds.credentials)
|