Frank Schwenk
88bdf6f34b
- Split monolithic app.py into server/api/ routers - Add Pydantic schemas for input validation - Add /agent/* API with separate AGENT_TOKEN auth - Replace static schema.sql with versioned migrations (datenerhaltend) - Add tools/backup_db.py for SQLite backups - Add Dockerfile HEALTHCHECK and compose.yml AGENT_TOKEN env - Add pytest tests for user and agent API - Update README with API docs and deployment safety notes - Bump version to 1.2.0
58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
import hmac
|
|
import secrets
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
import jwt
|
|
from fastapi import Depends, HTTPException, status
|
|
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(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)
|
|
|
|
|
|
async def require_agent(
|
|
creds: HTTPAuthorizationCredentials | None = Depends(_bearer),
|
|
) -> dict[str, Any]:
|
|
"""Authenticate external agents via AGENT_TOKEN (long-lived, not JWT)."""
|
|
if creds is None or creds.scheme.lower() != "bearer":
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing agent token")
|
|
token = creds.credentials
|
|
if not settings.agent_token:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="Agent API not configured",
|
|
)
|
|
if not secrets.compare_digest(token.strip(), settings.agent_token.strip()):
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid agent token")
|
|
return {"sub": "agent"}
|