DataOps-platform/deployment/app/core/system/auth.py

"""PostgreSQL-backed authentication using Argon2id and short-lived bearer tokens."""

from __future__ import annotations

import hashlib
import json
import logging
import threading
import time
from collections import defaultdict, deque
from typing import Any

from argon2 import PasswordHasher
from argon2.exceptions import InvalidHashError, VerificationError, VerifyMismatchError
from flask import current_app
from sqlalchemy import text
from sqlalchemy.exc import ProgrammingError, SQLAlchemyError

from app import db
from app.core.common.identifiers import new_governance_uid
from app.core.system.tokens import TokenError, decode_access_token, issue_access_token

logger = logging.getLogger(__name__)
_password_hasher = PasswordHasher()
_failure_lock = threading.Lock()
_login_failures: dict[str, deque[float]] = defaultdict(deque)
_FAILURE_WINDOW_SECONDS = 300
_FAILURE_LIMIT = 5


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


def verify_password(encoded: str, password: str) -> bool:
    try:
        return bool(_password_hasher.verify(encoded, password))
    except (VerifyMismatchError, VerificationError, InvalidHashError):
        return False


def _rate_key(username: str, ip_address: str | None) -> str:
    return f"{username.strip().lower()}|{ip_address or '-'}"


def _check_rate_limit(key: str, *, now: float | None = None) -> None:
    now = now or time.time()
    with _failure_lock:
        failures = _login_failures[key]
        while failures and failures[0] <= now - _FAILURE_WINDOW_SECONDS:
            failures.popleft()
        if len(failures) >= _FAILURE_LIMIT:
            raise PermissionError("too many login attempts")


def _record_failure(key: str, *, now: float | None = None) -> None:
    with _failure_lock:
        _login_failures[key].append(now or time.time())


def _clear_failures(key: str) -> None:
    with _failure_lock:
        _login_failures.pop(key, None)


def _audit(
    *,
    user_id: str | None,
    username: str,
    event_type: str,
    success: bool,
    ip_address: str | None,
    user_agent: str | None,
    detail: str | None = None,
) -> None:
    db.session.execute(
        text(
            "INSERT INTO public.auth_audit_events "
            "(user_id, username, event_type, success, ip_address, user_agent, detail) "
            "VALUES (CAST(:user_id AS uuid), :username, :event_type, :success, "
            ":ip_address, :user_agent, :detail)"
        ),
        {
            "user_id": user_id,
            "username": username[:64],
            "event_type": event_type,
            "success": success,
            "ip_address": (ip_address or "")[:64] or None,
            "user_agent": (user_agent or "")[:300] or None,
            "detail": (detail or "")[:500] or None,
        },
    )
    try:
        with db.session.begin_nested():
            db.session.execute(
                text(
                    "INSERT INTO public.identity_audit_events "
                    "(uid,event_type,outcome,actor_uid,enterprise_subject_digest,resource_type,resource_uid,safe_detail) "
                    "VALUES (CAST(:uid AS uuid),:event_type,:outcome,CAST(:actor AS uuid),:subject_digest,"
                    "'authentication',:resource_uid,CAST(:detail AS jsonb))"
                ),
                {
                    "uid": new_governance_uid(),
                    "event_type": event_type,
                    "outcome": "success" if success else "failure",
                    "actor": user_id,
                    "subject_digest": hashlib.sha256(username.strip().lower().encode()).hexdigest(),
                    "resource_uid": user_id,
                    "detail": json.dumps({"ip_address": (ip_address or "")[:64] or None,
                                          "user_agent": (user_agent or "")[:300] or None,
                                          "reason": (detail or "")[:160] or None}),
                },
            )
    except ProgrammingError:
        # Rolling deployment before migration 470 retains the legacy audit row.
        pass


def authenticate_user(
    username: str,
    password: str,
    *,
    ip_address: str | None = None,
    user_agent: str | None = None,
) -> dict[str, Any] | None:
    key = _rate_key(username, ip_address)
    _check_rate_limit(key)
    row = db.session.execute(
        text(
            "SELECT u.id::text, u.username, u.display_name, u.password_hash, "
            "u.status, COALESCE(array_agg(r.name ORDER BY r.name) "
            "FILTER (WHERE r.name IS NOT NULL), ARRAY[]::varchar[]) AS roles "
            "FROM public.users u "
            "LEFT JOIN public.user_roles ur ON ur.user_id = u.id "
            "LEFT JOIN public.roles r ON r.id = ur.role_id "
            "WHERE lower(u.username) = lower(:username) "
            "GROUP BY u.id"
        ),
        {"username": username.strip()},
    ).one_or_none()

    valid = bool(row and row[4] == "active" and verify_password(row[3], password))
    if not valid:
        _record_failure(key)
        _audit(
            user_id=row[0] if row else None,
            username=username,
            event_type="login",
            success=False,
            ip_address=ip_address,
            user_agent=user_agent,
            detail="invalid credentials or disabled account",
        )
        db.session.commit()
        return None

    # Once an enterprise IdP is active, local authentication is a controlled
    # break-glass path only. Missing identity tables mean pre-migration/no-IdP
    # compatibility; an active IdP always fails closed.
    try:
        active_provider = db.session.execute(
            text("SELECT EXISTS (SELECT 1 FROM public.identity_idp_config_versions WHERE status='active')")
        ).scalar()
    except SQLAlchemyError:
        db.session.rollback()
        logger.exception("enterprise identity state unavailable; local login denied")
        return None
    if active_provider:
        emergency_request = db.session.execute(
            text(
                "SELECT uid::text,expires_at FROM public.identity_emergency_requests "
                "WHERE emergency_account_uid=CAST(:id AS uuid) AND status='active' "
                "AND starts_at<=CURRENT_TIMESTAMP AND expires_at>CURRENT_TIMESTAMP "
                "AND EXISTS (SELECT 1 FROM public.user_roles ur JOIN public.roles r ON r.id=ur.role_id "
                "WHERE ur.user_id=CAST(:id AS uuid) AND r.name='admin') "
                "AND NOT EXISTS (SELECT 1 FROM public.enterprise_identity_links l WHERE l.user_uid=CAST(:id AS uuid)) "
                "ORDER BY expires_at ASC LIMIT 1"
            ),
            {"id": row[0]},
        ).one_or_none()
        if not emergency_request:
            _audit(user_id=row[0], username=row[1], event_type="local_login_blocked_by_idp", success=False,
                   ip_address=ip_address, user_agent=user_agent, detail="active enterprise IdP requires approved emergency access")
            db.session.commit()
            return None

    db.session.execute(
        text("UPDATE public.users SET last_login_at = CURRENT_TIMESTAMP WHERE id = CAST(:id AS uuid)"),
        {"id": row[0]},
    )
    _audit(
        user_id=row[0],
        username=row[1],
        event_type="login",
        success=True,
        ip_address=ip_address,
        user_agent=user_agent,
    )
    db.session.commit()
    _clear_failures(key)
    roles = list(row[5])
    token = issue_access_token(
        user_id=row[0], roles=roles, secret=current_app.config["SECRET_KEY"]
    )
    if active_provider:
        from app.core.system.identity_repository import PostgresIdentityRepository
        from app.core.system.identity_sessions import SessionManager

        credentials = SessionManager(
            PostgresIdentityRepository(db.session), secret=current_app.config["SECRET_KEY"]
        ).create(
            subject=row[0],
            user_uid=row[0],
            roles=roles,
            identity_source="emergency",
            emergency_request_uid=emergency_request[0],
            emergency_expires_at=emergency_request[1],
        )
        token = credentials.access_token
        PostgresIdentityRepository(db.session).audit(
            {
                "event_type": "emergency_local_login",
                "outcome": "warning",
                "actor_uid": row[0],
                "resource_type": "emergency_access",
                "resource_uid": row[0],
                "safe_detail": {"alert_required": True},
            }
        )
    return {
        "id": row[0],
        "username": row[1],
        "display_name": row[2],
        "roles": roles,
        "token": token,
    }


def load_identity(user_id: str) -> dict[str, Any] | None:
    row = db.session.execute(
        text(
            "SELECT u.id::text, u.username, u.display_name, u.status, "
            "COALESCE(array_agg(r.name ORDER BY r.name) FILTER "
            "(WHERE r.name IS NOT NULL), ARRAY[]::varchar[]) "
            "FROM public.users u "
            "LEFT JOIN public.user_roles ur ON ur.user_id = u.id "
            "LEFT JOIN public.roles r ON r.id = ur.role_id "
            "WHERE u.id = CAST(:id AS uuid) GROUP BY u.id"
        ),
        {"id": user_id},
    ).one_or_none()
    if not row or row[3] != "active":
        return None
    return {"id": row[0], "username": row[1], "display_name": row[2], "roles": list(row[4])}


def load_identity_from_token(token: str, *, secret: str) -> dict[str, Any] | None:
    try:
        claims = decode_access_token(token, secret=secret)
    except TokenError:
        return None
    try:
        active_provider = bool(db.session.execute(text(
            "SELECT EXISTS (SELECT 1 FROM public.identity_idp_config_versions WHERE status='active')"
        )).scalar())
    except SQLAlchemyError:
        db.session.rollback()
        logger.exception("enterprise identity state unavailable; bearer token denied")
        return None
    if active_provider and not claims.get("sid"):
        return None
    if claims.get("sid"):
        try:
            session = db.session.execute(
                text("""
                SELECT s.status,s.token_version,s.user_uid::text,s.identity_source,
                  CASE WHEN s.identity_source='emergency' THEN
                    (e.status='active' AND e.starts_at<=CURRENT_TIMESTAMP AND e.expires_at>CURRENT_TIMESTAMP
                     AND u.status='active'
                     AND EXISTS (SELECT 1 FROM public.user_roles ur JOIN public.roles r ON r.id=ur.role_id
                                 WHERE ur.user_id=u.id AND r.name='admin')
                     AND NOT EXISTS (SELECT 1 FROM public.enterprise_identity_links l WHERE l.user_uid=u.id))
                  ELSE TRUE END emergency_valid
                FROM public.identity_sessions s LEFT JOIN public.identity_emergency_requests e
                  ON e.uid=s.emergency_request_uid LEFT JOIN public.users u ON u.id=s.user_uid
                WHERE s.uid=CAST(:sid AS uuid)
                """),
                {"sid": claims["sid"]},
            ).one_or_none()
        except SQLAlchemyError:
            db.session.rollback()
            return None
        if session and session[3] == "emergency" and not session[4]:
            db.session.execute(text("UPDATE public.identity_sessions SET status='revoked',revoke_reason='emergency_window_closed',revoked_at=CURRENT_TIMESTAMP WHERE uid=CAST(:sid AS uuid)"), {"sid": claims["sid"]})
            db.session.commit()
            return None
        if (not session or session[0] != "active" or session[2] != claims["sub"]
                or int(session[1]) != int(claims.get("token_version", -1))):
            return None
    identity = load_identity(claims["sub"])
    if identity and claims.get("sid"):
        identity.update(sid=claims["sid"], token_version=claims["token_version"],
                        identity_source=claims.get("identity_source", "oidc"))
    return identity


# Compatibility aliases retained only for imports during the route migration.
def login_user(username: str, password: str):
    result = authenticate_user(username, password)
    return (True, result) if result else (False, "用户名或密码错误")


def get_user_by_username(username: str):
    row = db.session.execute(
        text("SELECT id::text FROM public.users WHERE lower(username) = lower(:username)"),
        {"username": username},
    ).scalar_one_or_none()
    return load_identity(row) if row else None


def init_db() -> bool:
    logger.warning("init_db is deprecated; use Alembic migrations")
    return True


def require_auth(view):
    from app.core.system.permissions import READ_GOVERNANCE, require_permissions

    return require_permissions(READ_GOVERNANCE)(view)