security-alert-center/backend/app/auth/jwt_auth.py

from dataclasses import dataclass



from datetime import datetime, timedelta, timezone







from fastapi import Depends, HTTPException, status



from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer



from jose import JWTError, jwt



from sqlalchemy import func, select



from sqlalchemy.orm import Session







from app.config import get_settings



from app.database import get_db



from app.models.user import USER_ROLE_ADMIN, User







bearer_scheme = HTTPBearer(auto_error=False)











@dataclass(frozen=True)



class CurrentUser:
    username: str
    role: str
    device_id: int | None = None







    @property



    def is_admin(self) -> bool:



        return self.role == USER_ROLE_ADMIN











def create_access_token(subject: str, role: str, *, device_id: int | None = None) -> str:
    settings = get_settings()
    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_expire_minutes)
    payload: dict[str, object] = {"sub": subject, "role": role, "exp": expire}
    if device_id is not None:
        payload["device_id"] = device_id
    return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)











def _assert_mobile_device_active(db: Session, device_id: int, user_id: int) -> None:
    from app.models.mobile_device import MobileDevice

    device = db.get(MobileDevice, device_id)
    if device is None or device.user_id != user_id or device.revoked_at is not None:
        raise HTTPException(status_code=401, detail="Mobile device revoked or invalid")


def get_token_device_id(token: str) -> int | None:
    settings = get_settings()
    try:
        payload = jwt.decode(
            token,
            settings.jwt_secret,
            algorithms=[settings.jwt_algorithm],
        )
    except JWTError:
        return None
    raw = payload.get("device_id")
    if raw is None:
        return None
    try:
        return int(raw)
    except (TypeError, ValueError):
        return None


def _user_from_db(db: Session, username: str) -> User:



    normalized = username.strip()



    user = db.scalar(select(User).where(func.lower(User.username) == normalized.lower()))



    if user is None or not user.is_active:



        raise HTTPException(status_code=401, detail="Invalid token")



    return user











def _decode_current_user(token: str, db: Session | None = None) -> CurrentUser:



    settings = get_settings()



    try:



        payload = jwt.decode(



            token,



            settings.jwt_secret,



            algorithms=[settings.jwt_algorithm],



        )



        username = payload.get("sub")



        if not username:



            raise HTTPException(status_code=401, detail="Invalid token")



        parsed_device_id: int | None = None
        raw_device_id = payload.get("device_id")
        if raw_device_id is not None:
            try:
                parsed_device_id = int(raw_device_id)
            except (TypeError, ValueError) as exc:
                raise HTTPException(status_code=401, detail="Invalid token") from exc

        if db is not None:
            user = _user_from_db(db, str(username))
            if parsed_device_id is not None:
                _assert_mobile_device_active(db, parsed_device_id, user.id)
            return CurrentUser(username=user.username, role=user.role, device_id=parsed_device_id)

        role = str(payload.get("role") or USER_ROLE_ADMIN)
        return CurrentUser(username=str(username), role=role, device_id=parsed_device_id)



    except JWTError as exc:



        raise HTTPException(status_code=401, detail="Invalid token") from exc











def verify_access_token(token: str, db: Session | None = None) -> str:



    """Проверка JWT (в т.ч. query-параметр для SSE). Возвращает username."""



    return _decode_current_user(token, db=db).username











def get_current_user(



    credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),



    db: Session = Depends(get_db),



) -> CurrentUser:



    if credentials is None or credentials.scheme.lower() != "bearer":



        raise HTTPException(



            status_code=status.HTTP_401_UNAUTHORIZED,



            detail="Not authenticated",



        )



    return _decode_current_user(credentials.credentials, db=db)











def require_admin(user: CurrentUser = Depends(get_current_user)) -> CurrentUser:



    if not user.is_admin:



        raise HTTPException(



            status_code=status.HTTP_403_FORBIDDEN,



            detail="Admin access required",



        )



    return user