initial commit

api/app/auth.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import os
+import secrets
+import time
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from fastapi import Depends, HTTPException, Request, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+import bcrypt
+from jose import JWTError, jwt
+
+ADMIN_USERNAME = os.getenv("ADMIN_USERNAME", "admin")
+ADMIN_PASSWORD_HASH = os.getenv("ADMIN_PASSWORD_HASH", "")
+ADMIN_JWT_SECRET = os.getenv("ADMIN_JWT_SECRET", "")
+ADMIN_JWT_EXPIRE_HOURS = int(os.getenv("ADMIN_JWT_EXPIRE_HOURS", "8"))
+ADMIN_LOGIN_MAX_ATTEMPTS = int(os.getenv("ADMIN_LOGIN_MAX_ATTEMPTS", "5"))
+ADMIN_LOGIN_WINDOW_SECONDS = int(os.getenv("ADMIN_LOGIN_WINDOW_SECONDS", "300"))
+
+bearer_scheme = HTTPBearer(auto_error=False)
+
+_login_attempts: dict[str, list[float]] = {}
+
+
+def admin_configured() -> bool:
+    return bool(ADMIN_PASSWORD_HASH and ADMIN_JWT_SECRET and len(ADMIN_JWT_SECRET) >= 32)
+
+
+def verify_password(plain_password: str, password_hash: str) -> bool:
+    try:
+        return bcrypt.checkpw(plain_password.encode("utf-8"), password_hash.encode("utf-8"))
+    except (ValueError, TypeError):
+        return False
+
+
+def hash_password(plain_password: str) -> str:
+    return bcrypt.hashpw(plain_password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
+
+
+def _client_key(request: Request) -> str:
+    return request.client.host if request.client else "unknown"
+
+
+def check_login_rate_limit(request: Request) -> None:
+    key = _client_key(request)
+    now = time.time()
+    attempts = [ts for ts in _login_attempts.get(key, []) if now - ts < ADMIN_LOGIN_WINDOW_SECONDS]
+    if len(attempts) >= ADMIN_LOGIN_MAX_ATTEMPTS:
+        raise HTTPException(status_code=429, detail="Too many login attempts. Try again later.")
+    _login_attempts[key] = attempts
+
+
+def record_failed_login(request: Request) -> None:
+    key = _client_key(request)
+    _login_attempts.setdefault(key, []).append(time.time())
+
+
+def create_access_token(username: str) -> str:
+    expire = datetime.now(timezone.utc) + timedelta(hours=ADMIN_JWT_EXPIRE_HOURS)
+    payload = {"sub": username, "exp": expire, "scope": "admin"}
+    return jwt.encode(payload, ADMIN_JWT_SECRET, algorithm="HS256")
+
+
+def decode_access_token(token: str) -> dict[str, Any]:
+    return jwt.decode(token, ADMIN_JWT_SECRET, algorithms=["HS256"])
+
+
+async def require_admin(
+    request: Request,
+    credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),
+) -> dict[str, Any]:
+    if not admin_configured():
+        raise HTTPException(status_code=503, detail="Admin access is not configured")
+    if credentials is None or credentials.scheme.lower() != "bearer":
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing bearer token")
+    try:
+        payload = decode_access_token(credentials.credentials)
+    except JWTError as exc:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token") from exc
+    if payload.get("sub") != ADMIN_USERNAME or payload.get("scope") != "admin":
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token scope")
+    return payload
+
+
+def generate_secret(length: int = 48) -> str:
+    return secrets.token_urlsafe(length)