Stend_mont/app/auth.py

import secrets
from typing import Optional

from fastapi import Depends, HTTPException, Request, status
from fastapi.responses import RedirectResponse
from itsdangerous import BadSignature, URLSafeTimedSerializer
from passlib.context import CryptContext
from sqlalchemy.orm import Session

from config import COOKIE_MAX_AGE, COOKIE_NAME, CSRF_COOKIE
from database import get_db
from models import User, UserServiceAccess
from utils import now_utc
from sqlalchemy import select

import os

_SIGNING_KEY = os.getenv("SIGNING_KEY", secrets.token_urlsafe(32))
serializer = URLSafeTimedSerializer(_SIGNING_KEY, salt="portal-auth")
pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")


def hash_password(password: str) -> str:
    return pwd_context.hash(password)


def verify_password(password: str, password_hash: str) -> bool:
    return pwd_context.verify(password, password_hash)


def user_is_valid(user: User) -> bool:
    return bool(user.active and user.expires_at > now_utc())


def issue_auth_cookie(response: RedirectResponse, user: User) -> None:
    token = serializer.dumps({"user_id": user.id})
    response.set_cookie(
        key=COOKIE_NAME,
        value=token,
        httponly=True,
        secure=True,
        samesite="strict",
        max_age=COOKIE_MAX_AGE,
        path="/",
    )


def issue_csrf_cookie(response: RedirectResponse) -> str:
    token = secrets.token_urlsafe(24)
    response.set_cookie(
        key=CSRF_COOKIE,
        value=token,
        httponly=False,
        secure=True,
        samesite="lax",
        max_age=COOKIE_MAX_AGE,
        path="/",
    )
    return token


def get_current_user(request: Request, db: Session = Depends(get_db)) -> Optional[User]:
    raw = request.cookies.get(COOKIE_NAME)
    if not raw:
        return None
    try:
        payload = serializer.loads(raw, max_age=COOKIE_MAX_AGE)
    except BadSignature:
        return None
    user = db.get(User, int(payload["user_id"]))
    if not user or not user_is_valid(user):
        return None
    return user


def require_user(user: Optional[User] = Depends(get_current_user)) -> User:
    if not user:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized")
    return user


def require_admin(user: User = Depends(require_user)) -> User:
    if not user.is_admin:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin only")
    return user


def validate_csrf(request: Request) -> None:
    cookie = request.cookies.get(CSRF_COOKIE)
    form_val = request.headers.get("X-CSRF-Token")
    if request.headers.get("content-type", "").startswith("application/x-www-form-urlencoded"):
        return
    if not cookie or not form_val or cookie != form_val:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="CSRF failed")


def has_access(db: Session, user_id: int, service_id: int) -> bool:
    access = db.scalar(
        select(UserServiceAccess).where(
            UserServiceAccess.user_id == user_id,
            UserServiceAccess.service_id == service_id,
        )
    )
    if access is None:
        return False
    # A per-grant expires_at (used by pilots, whose access window can be
    # shorter than the account's own expires_at) overrides the account
    # expiry for this one product. NULL means "follow the account".
    if access.expires_at is not None and access.expires_at <= now_utc():
        return False
    return True

import threading
import time

_login_attempts: dict = {}
_login_lock = threading.Lock()

_LOGIN_MAX_ATTEMPTS = 5
_LOGIN_WINDOW = 300   # сброс счётчика через 5 минут
_LOGIN_BLOCK  = 900   # блокировка на 15 минут


def check_login_rate_limit(ip: str) -> Optional[float]:
    """Возвращает blocked_until (time.monotonic) если IP заблокирован, иначе None."""
    now = time.monotonic()
    with _login_lock:
        entry = _login_attempts.get(ip)
        if not entry:
            return None
        if entry["blocked_until"] > now:
            return entry["blocked_until"]
        if now - entry["first"] > _LOGIN_WINDOW:
            del _login_attempts[ip]
    return None


def record_login_failure(ip: str) -> None:
    now = time.monotonic()
    with _login_lock:
        entry = _login_attempts.get(ip)
        if not entry or now - entry["first"] > _LOGIN_WINDOW:
            _login_attempts[ip] = {"count": 1, "first": now, "blocked_until": 0.0}
            entry = _login_attempts[ip]
        else:
            entry["count"] += 1
        if entry["count"] >= _LOGIN_MAX_ATTEMPTS:
            entry["blocked_until"] = now + _LOGIN_BLOCK


def record_login_success(ip: str) -> None:
    with _login_lock:
        _login_attempts.pop(ip, None)