67 lines
1.8 KiB
Python
67 lines
1.8 KiB
Python
|
"""Password hashing + JWT tokens for the single admin account."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import os
|
||
|
|
from datetime import datetime, timedelta, timezone
|
||
|
|
|
||
|
|
import jwt
|
||
|
|
from fastapi import Depends, HTTPException, status
|
||
|
|
from fastapi.security import OAuth2PasswordBearer
|
||
|
|
from passlib.context import CryptContext
|
||
|
|
from sqlalchemy.orm import Session
|
||
|
|
|
||
|
|
from app.db import get_db
|
||
|
|
from app.models import AdminUser
|
||
|
|
|
||
|
|
_pwd = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||
|
|
|
||
|
|
SECRET_KEY = os.getenv("JWT_SECRET", "swatiyadav-dev-secret-change-me")
|
||
|
|
ALGORITHM = "HS256"
|
||
|
|
TOKEN_TTL_HOURS = 24 * 7
|
||
|
|
|
||
|
|
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/auth/login", auto_error=False)
|
||
|
|
|
||
|
|
|
||
|
|
def hash_password(password: str) -> str:
|
||
|
|
return _pwd.hash(password)
|
||
|
|
|
||
|
|
|
||
|
|
def verify_password(password: str, hashed: str) -> bool:
|
||
|
|
try:
|
||
|
|
return _pwd.verify(password, hashed)
|
||
|
|
except Exception: # noqa: BLE001
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def create_token(username: str) -> str:
|
||
|
|
payload = {
|
||
|
|
"sub": username,
|
||
|
|
"exp": datetime.now(timezone.utc) + timedelta(hours=TOKEN_TTL_HOURS),
|
||
|
|
}
|
||
|
|
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
|
||
|
|
|
||
|
|
|
||
|
|
def get_current_admin(
|
||
|
|
token: str | None = Depends(oauth2_scheme),
|
||
|
|
db: Session = Depends(get_db),
|
||
|
|
) -> AdminUser:
|
||
|
|
creds_error = HTTPException(
|
||
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
|
|
detail="Not authenticated",
|
||
|
|
headers={"WWW-Authenticate": "Bearer"},
|
||
|
|
)
|
||
|
|
if not token:
|
||
|
|
raise creds_error
|
||
|
|
try:
|
||
|
|
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||
|
|
username = payload.get("sub")
|
||
|
|
except jwt.PyJWTError:
|
||
|
|
raise creds_error from None
|
||
|
|
if not username:
|
||
|
|
raise creds_error
|
||
|
|
user = db.query(AdminUser).filter(AdminUser.username == username).first()
|
||
|
|
if not user:
|
||
|
|
raise creds_error
|
||
|
|
return user
|