agatha
5fbbc1e67f
Protects image upload, delete, and tag-update endpoints behind Bearer token auth. Public read endpoints remain open. Angular SPA gains a login page, auth interceptor, and route guard for /upload. - JWTAuthProvider (HS256, sub/iat/exp, secrets.compare_digest) - POST /api/v1/auth/token login endpoint - require_auth FastAPI dependency on all write routes - AuthService, LoginComponent, authInterceptor, authGuard - Detail page hides write controls for unauthenticated visitors - 43 unit tests passing; integration tests require Docker stack Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
61 lines
1.7 KiB
Python
61 lines
1.7 KiB
Python
from collections.abc import AsyncGenerator
|
|
|
|
from fastapi import Depends, Header, HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.auth.jwt_provider import JWTAuthProvider
|
|
from app.auth.provider import AuthProvider, Identity
|
|
from app.database import get_session_factory
|
|
from app.storage.backend import StorageBackend
|
|
from app.storage.s3_backend import S3StorageBackend
|
|
|
|
_storage: StorageBackend | None = None
|
|
_auth: AuthProvider | None = None
|
|
|
|
|
|
def get_storage() -> StorageBackend:
|
|
global _storage
|
|
if _storage is None:
|
|
_storage = S3StorageBackend()
|
|
return _storage
|
|
|
|
|
|
def get_auth() -> AuthProvider:
|
|
global _auth
|
|
if _auth is None:
|
|
from app.config import get_settings
|
|
|
|
s = get_settings()
|
|
_auth = JWTAuthProvider(
|
|
secret_key=s.jwt_secret_key,
|
|
expiry_seconds=s.jwt_expiry_seconds,
|
|
owner_username=s.owner_username,
|
|
owner_password=s.owner_password,
|
|
)
|
|
return _auth
|
|
|
|
|
|
def get_jwt_auth() -> JWTAuthProvider:
|
|
auth = get_auth()
|
|
assert isinstance(auth, JWTAuthProvider)
|
|
return auth
|
|
|
|
|
|
async def require_auth(
|
|
authorization: str | None = Header(None, alias="Authorization"),
|
|
auth: AuthProvider = Depends(get_auth),
|
|
) -> Identity:
|
|
identity = await auth.get_identity(authorization)
|
|
if identity.anonymous:
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail={"detail": "Authentication required", "code": "unauthorized"},
|
|
)
|
|
return identity
|
|
|
|
|
|
async def get_db() -> AsyncGenerator[AsyncSession, None]:
|
|
factory = get_session_factory()
|
|
async with factory() as session, session.begin():
|
|
yield session
|