mirror of
https://github.com/EstrellaXD/Auto_Bangumi.git
synced 2026-04-14 02:20:53 +08:00
027222a24d
- Fix aaguid type (str not bytes) in registration verification - Fix missing credential_backup_eligible field (use credential_device_type) - Remove invalid credential_id param from verify_authentication_response - Fix origin detection to use browser Origin header for WebAuthn verification - Add async database engine support (aiosqlite) for passkey operations - Convert UserDatabase to async-compatible with sync/async session detection - Update Database class to support both sync and async context managers Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
57 lines
1.5 KiB
Python
57 lines
1.5 KiB
Python
from fastapi import Cookie, Depends, HTTPException, status
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
|
|
from module.database import Database
|
|
from module.models.user import User, UserUpdate
|
|
|
|
from .jwt import verify_token
|
|
|
|
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
|
|
|
|
active_user = []
|
|
|
|
|
|
async def get_current_user(token: str = Cookie(None)):
|
|
if not token:
|
|
raise UNAUTHORIZED
|
|
payload = verify_token(token)
|
|
if not payload:
|
|
raise UNAUTHORIZED
|
|
username = payload.get("sub")
|
|
if not username:
|
|
raise UNAUTHORIZED
|
|
if username not in active_user:
|
|
raise UNAUTHORIZED
|
|
return username
|
|
|
|
|
|
async def get_token_data(token: str = Depends(oauth2_scheme)):
|
|
payload = verify_token(token)
|
|
if not payload:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid token"
|
|
)
|
|
return payload
|
|
|
|
|
|
async def update_user_info(user_data: UserUpdate, current_user):
|
|
try:
|
|
async with Database() as db:
|
|
await db.user.update_user(current_user, user_data)
|
|
return True
|
|
except Exception as e:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
|
|
|
|
|
|
async def auth_user(user: User):
|
|
async with Database() as db:
|
|
resp = await db.user.auth_user(user)
|
|
if resp.status:
|
|
active_user.append(user.username)
|
|
return resp
|
|
|
|
|
|
UNAUTHORIZED = HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized"
|
|
)
|