import os import bcrypt from datetime import datetime, timedelta, timezone from jose import JWTError, jwt from fastapi import Depends, HTTPException, status, Request from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from .database import get_db from .models import User SECRET_KEY = os.getenv("SECRET_KEY", "skift-mig-til-en-sikker-noegle") ALGORITHM = "HS256" TOKEN_EXPIRE_DAYS = 30 bearer_scheme = HTTPBearer(auto_error=False) def hash_password(password: str) -> str: return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") def verify_password(plain: str, hashed: str) -> bool: return bcrypt.checkpw(plain.encode("utf-8"), hashed.encode("utf-8")) def create_token(username: str, role: str) -> str: expire = datetime.now(timezone.utc) + timedelta(days=TOKEN_EXPIRE_DAYS) payload = {"sub": username, "role": role, "exp": expire} return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM) def decode_token(token: str) -> dict | None: try: return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) except JWTError: return None async def get_current_user( request: Request, credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme), db: AsyncSession = Depends(get_db), ) -> User | None: """Return the current user, or None if not authenticated. Checks both Authorization header and cookie.""" token = None if credentials: token = credentials.credentials if not token: token = request.cookies.get("token") if not token: return None payload = decode_token(token) if not payload: return None result = await db.execute(select(User).where(User.username == payload["sub"])) return result.scalar_one_or_none() async def require_admin(user: User | None = Depends(get_current_user)) -> User: if not user or user.role != "admin": raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Admin login required", ) return user