45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
## Copyright © 2026 Olaf Kolkman
|
|
## SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
|
|
from backend.app.database import get_connection
|
|
from backend.app.services.token_service import validate_token
|
|
|
|
bearer_scheme = HTTPBearer(auto_error=False)
|
|
|
|
|
|
def get_current_user(
|
|
credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),
|
|
):
|
|
if credentials is None or credentials.scheme.lower() != 'bearer':
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail='Authentication required',
|
|
headers={'WWW-Authenticate': 'Bearer'},
|
|
)
|
|
|
|
token_data = validate_token(credentials.credentials)
|
|
if token_data is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail='Token expired or invalid',
|
|
headers={'WWW-Authenticate': 'Bearer'},
|
|
)
|
|
|
|
with get_connection() as conn:
|
|
user = conn.execute(
|
|
'SELECT * FROM users WHERE id = ?',
|
|
(token_data['user_id'],),
|
|
).fetchone()
|
|
if user is None:
|
|
raise HTTPException(status_code=404, detail='User not found')
|
|
return dict(user)
|
|
|
|
|
|
def require_admin(user: dict = Depends(get_current_user)):
|
|
if not user['is_admin']:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail='Administrator access required')
|
|
return user
|