houhuan
dedc3b4183
- Full FastAPI backend with JWT auth, file management, processing pipeline, memory CRUD, barcode mappings, config management, cloud sync - Vue 3 + Element Plus frontend with dashboard, task history, HTTP logs, memory editor, barcode editor, config editor, sync page - HTTP request logging middleware with SQLite persistence - Task history tracking with progress and retry support - File metadata recording for upload/download operations - WebAuth section in config.ini for bcrypt password storage - Bug fix: logs.py count query returns tuple not dict Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
59 lines
2.0 KiB
Python
59 lines
2.0 KiB
Python
"""FastAPI auth dependencies"""
|
|
|
|
from fastapi import Depends, HTTPException, status, Query, Request
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
|
|
from .jwt_handler import decode_token
|
|
|
|
security = HTTPBearer()
|
|
|
|
|
|
async def get_current_user(
|
|
credentials: HTTPAuthorizationCredentials = Depends(security),
|
|
) -> dict:
|
|
try:
|
|
payload = decode_token(credentials.credentials)
|
|
username = payload.get("sub")
|
|
if username is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
|
|
return {"username": username}
|
|
except Exception:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的认证凭据")
|
|
|
|
|
|
async def get_current_user_ws(token: str = Query(...)) -> dict:
|
|
"""WebSocket auth via query parameter"""
|
|
try:
|
|
payload = decode_token(token)
|
|
username = payload.get("sub")
|
|
if username is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
|
|
return {"username": username}
|
|
except Exception:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的认证凭据")
|
|
|
|
|
|
async def get_current_user_flexible(
|
|
request: Request,
|
|
credentials: HTTPAuthorizationCredentials = Depends(HTTPBearer(auto_error=False)),
|
|
token: str = Query(None),
|
|
) -> dict:
|
|
"""Auth from header OR query param (for file downloads in browser)."""
|
|
token_str = None
|
|
if credentials:
|
|
token_str = credentials.credentials
|
|
elif token:
|
|
token_str = token
|
|
|
|
if not token_str:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="未提供认证凭据")
|
|
|
|
try:
|
|
payload = decode_token(token_str)
|
|
username = payload.get("sub")
|
|
if username is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
|
|
return {"username": username}
|
|
except Exception:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="无效的认证凭据")
|