Sage100-vps/security/auth.py

from passlib.context import CryptContext
from datetime import datetime, timedelta, timezone
from typing import Optional, Dict, Any, Tuple
import jwt
import secrets
import hashlib
import hmac
import logging

from config.config import settings

logger = logging.getLogger(__name__)

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto", bcrypt__rounds=12)


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


def verify_password(plain_password: str, hashed_password: str) -> bool:
    try:
        return pwd_context.verify(plain_password, hashed_password)
    except Exception as e:
        logger.warning(f"Erreur verification mot de passe: {e}")
        return False


def generate_secure_token(length: int = 32) -> str:
    return secrets.token_urlsafe(length)


def generate_verification_token() -> str:
    return generate_secure_token(32)


def generate_reset_token() -> str:
    return generate_secure_token(32)


def generate_csrf_token() -> str:
    return generate_secure_token(32)


def generate_refresh_token_id() -> str:
    return generate_secure_token(16)


def hash_token(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()


def constant_time_compare(val1: str, val2: str) -> bool:
    return hmac.compare_digest(val1.encode(), val2.encode())


def create_access_token(
    data: Dict[str, Any],
    expires_delta: Optional[timedelta] = None,
    fingerprint_hash: Optional[str] = None,
) -> str:
    to_encode = data.copy()
    now = datetime.now(timezone.utc)

    if expires_delta:
        expire = now + expires_delta
    else:
        expire = now + timedelta(minutes=settings.access_token_expire_minutes)

    to_encode.update(
        {
            "exp": expire,
            "iat": now,
            "nbf": now,
            "type": "access",
            "jti": generate_secure_token(8),
        }
    )

    if fingerprint_hash:
        to_encode["fph"] = fingerprint_hash

    return jwt.encode(to_encode, settings.jwt_secret, algorithm=settings.jwt_algorithm)


def create_refresh_token(
    user_id: str,
    token_id: Optional[str] = None,
    fingerprint_hash: Optional[str] = None,
    expires_delta: Optional[timedelta] = None,
) -> Tuple[str, str]:
    now = datetime.now(timezone.utc)

    if expires_delta:
        expire = now + expires_delta
    else:
        expire = now + timedelta(days=settings.refresh_token_expire_days)

    if not token_id:
        token_id = generate_refresh_token_id()

    to_encode = {
        "sub": user_id,
        "exp": expire,
        "iat": now,
        "nbf": now,
        "type": "refresh",
        "jti": token_id,
    }

    if fingerprint_hash:
        to_encode["fph"] = fingerprint_hash

    token = jwt.encode(to_encode, settings.jwt_secret, algorithm=settings.jwt_algorithm)

    return token, token_id


def create_csrf_token(session_id: str) -> str:
    now = datetime.now(timezone.utc)
    expire = now + timedelta(minutes=settings.csrf_token_expire_minutes)

    to_encode = {
        "sid": session_id,
        "exp": expire,
        "iat": now,
        "type": "csrf",
        "jti": generate_secure_token(8),
    }

    return jwt.encode(to_encode, settings.jwt_secret, algorithm=settings.jwt_algorithm)


def decode_token(
    token: str, expected_type: Optional[str] = None
) -> Optional[Dict[str, Any]]:
    try:
        payload = jwt.decode(
            token,
            settings.jwt_secret,
            algorithms=[settings.jwt_algorithm],
            options={
                "require": ["exp", "iat", "type"],
                "verify_exp": True,
                "verify_iat": True,
                "verify_nbf": True,
            },
        )

        if expected_type and payload.get("type") != expected_type:
            logger.warning(
                f"Type de token incorrect: attendu={expected_type}, recu={payload.get('type')}"
            )
            return None

        return payload

    except jwt.ExpiredSignatureError:
        logger.debug("Token expire")
        return None
    except jwt.InvalidTokenError as e:
        logger.warning(f"Token invalide: {e}")
        return None
    except Exception as e:
        logger.error(f"Erreur decodage token: {e}")
        return None


def validate_password_strength(password: str) -> Tuple[bool, str]:
    if len(password) < settings.password_min_length:
        return (
            False,
            f"Le mot de passe doit contenir au moins {settings.password_min_length} caracteres",
        )

    if settings.password_require_uppercase and not any(c.isupper() for c in password):
        return False, "Le mot de passe doit contenir au moins une majuscule"

    if settings.password_require_lowercase and not any(c.islower() for c in password):
        return False, "Le mot de passe doit contenir au moins une minuscule"

    if settings.password_require_digit and not any(c.isdigit() for c in password):
        return False, "Le mot de passe doit contenir au moins un chiffre"

    if settings.password_require_special:
        special_chars = "!@#$%^&*()_+-=[]{}|;:,.<>?/~`"
        if not any(c in special_chars for c in password):
            return False, "Le mot de passe doit contenir au moins un caractere special"

    common_passwords = [
        "password",
        "123456",
        "qwerty",
        "admin",
        "letmein",
        "welcome",
        "monkey",
        "dragon",
        "master",
        "login",
    ]
    if password.lower() in common_passwords:
        return False, "Ce mot de passe est trop courant"

    return True, ""


def generate_session_id() -> str:
    """Genere un identifiant de session unique."""
    return generate_secure_token(24)