notes-qpid-pro/app/auth.py

import hashlib
import os
import secrets
import time
from datetime import datetime, timezone, timedelta
from typing import Optional
from fastapi import Request, HTTPException, status
from app import config
from app.database import get_db

# In-memory IP-based rate limiting for login attempts
_login_failures: dict[str, list[float]] = {}

def hash_password(password: str, salt: Optional[str] = None) -> tuple[str, str]:
    if not salt:
        salt = secrets.token_hex(16)
    salt_bytes = bytes.fromhex(salt)
    key = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt_bytes, 100_000)
    return key.hex(), salt

def verify_password(password: str, stored_hash: str, salt: str) -> bool:
    new_hash, _ = hash_password(password, salt)
    return secrets.compare_digest(new_hash, stored_hash)

def check_login_rate_limit(ip: str) -> bool:
    now = time.time()
    cutoff = now - config.RATE_LIMIT_LOGIN_WINDOW_SEC
    attempts = _login_failures.get(ip, [])
    # Filter attempts within the time window
    recent_attempts = [t for t in attempts if t > cutoff]
    _login_failures[ip] = recent_attempts
    return len(recent_attempts) < config.RATE_LIMIT_LOGIN_MAX

def record_login_failure(ip: str):
    now = time.time()
    if ip not in _login_failures:
        _login_failures[ip] = []
    _login_failures[ip].append(now)

def record_login_success(ip: str):
    if ip in _login_failures:
        _login_failures.pop(ip, None)

def create_user(username: str, password: str) -> int:
    pwd_hash, salt = hash_password(password)
    with get_db() as conn:
        cursor = conn.execute(
            "INSERT INTO users (username, password_hash, salt) VALUES (?, ?, ?)",
            (username.strip(), pwd_hash, salt)
        )
        return cursor.lastrowid

def authenticate_user(username: str, password: str) -> Optional[dict]:
    with get_db() as conn:
        row = conn.execute(
            "SELECT id, username, password_hash, salt FROM users WHERE username = ?",
            (username.strip(),)
        ).fetchone()
        if not row:
            return None
        if verify_password(password, row["password_hash"], row["salt"]):
            return {"id": row["id"], "username": row["username"]}
        return None

def create_session(user_id: int, user_agent: str = "", ip: str = "") -> str:
    token = secrets.token_urlsafe(32)
    expires_at = (datetime.now(timezone.utc) + timedelta(days=config.SESSION_EXPIRE_DAYS)).isoformat()
    with get_db() as conn:
        conn.execute(
            "INSERT INTO sessions (token, user_id, expires_at, user_agent, ip) VALUES (?, ?, ?, ?, ?)",
            (token, user_id, expires_at, user_agent, ip)
        )
    return token

def get_user_from_session(token: str) -> Optional[dict]:
    if not token:
        return None
    with get_db() as conn:
        row = conn.execute(
            """
            SELECT u.id, u.username, s.expires_at
            FROM sessions s
            JOIN users u ON s.user_id = u.id
            WHERE s.token = ?
            """,
            (token,)
        ).fetchone()
        if not row:
            return None
        
        expires_at_str = row["expires_at"]
        try:
            expires_at = datetime.fromisoformat(expires_at_str)
            if expires_at.tzinfo is None:
                expires_at = expires_at.replace(tzinfo=timezone.utc)
        except Exception:
            return None

        if expires_at < datetime.now(timezone.utc):
            # Session expired
            delete_session(token)
            return None

        return {"id": row["id"], "username": row["username"]}

def delete_session(token: str):
    if not token:
        return
    with get_db() as conn:
        conn.execute("DELETE FROM sessions WHERE token = ?", (token,))

def change_user_password(user_id: int, old_password: str, new_password: str) -> tuple[bool, str]:
    """Changes password for the specified user after verifying old password."""
    if not new_password or len(new_password) < 6:
        return False, "Новый пароль должен содержать минимум 6 символов"
    with get_db() as conn:
        row = conn.execute(
            "SELECT password_hash, salt FROM users WHERE id = ?",
            (user_id,)
        ).fetchone()
        if not row:
            return False, "Пользователь не найден"
        if not verify_password(old_password, row["password_hash"], row["salt"]):
            return False, "Неверный текущий пароль"
        pwd_hash, salt = hash_password(new_password)
        conn.execute(
            "UPDATE users SET password_hash = ?, salt = ? WHERE id = ?",
            (pwd_hash, salt, user_id)
        )
        return True, "Пароль успешно изменён"


def ensure_admin_user():
    """Ensures at least one user exists; creates default admin from config if table empty."""
    with get_db() as conn:
        count = conn.execute("SELECT COUNT(*) FROM users").fetchone()[0]
        if count == 0 and config.ADMIN_PASSWORD:
            create_user(config.ADMIN_USERNAME, config.ADMIN_PASSWORD)


def get_optional_user(request: Request) -> Optional[dict]:
    token = request.cookies.get(config.SESSION_COOKIE_NAME)
    if not token:
        auth_header = request.headers.get("Authorization", "")
        if auth_header.startswith("Bearer "):
            token = auth_header[7:].strip()
    if token:
        return get_user_from_session(token)
    return None


def require_user(request: Request) -> dict:
    user = get_optional_user(request)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Authentication required",
        )
    return user