YG_FT/backend/app/core/auth.py

"""鉴权依赖：从 Authorization header 解析当前用户，提供权限校验。"""
from __future__ import annotations

from typing import Any
import json

from fastapi import Depends, HTTPException, Query, Request, status

from app.db.platform_store import get_platform_store
from app.core.logging import get_client_ip

# 无需鉴权的路径前缀（健康检查、登录等）
PUBLIC_PATHS = ("/health", "/login", "/system-info")
OWNER_TABLES = {
    "dataset": ("datasets", "created_by"), "model": ("models", "created_by"),
    "trained_model": ("trained_models", "created_by"), "eval": ("eval_tasks", "created_by"),
    "fine-tune": ("fine_tune_tasks", "created_by"), "fine_tune_task": ("fine_tune_tasks", "created_by"),
    "compare": ("compare_tasks", "payload"), "inference": ("compare_tasks", "payload"),
    "project": ("projects", "create_by"), "data_process": ("data_process_tasks", "created_by"),
    "data_convert": ("data_convert_tasks", "created_by"),
}
RESOURCE_TABLES = {
    "dataset": "datasets",
    "model": "models",
    "trained_model": "trained_models",
    "eval": "eval_tasks",
    "fine-tune": "fine_tune_tasks",
    "fine_tune_task": "fine_tune_tasks",
    "compare": "compare_tasks",
    "inference": "compare_tasks",
    "project": "projects",
    "data_process": "data_process_tasks",
    "data_convert": "data_convert_tasks",
}
MODULE_PERMISSIONS = {
    "dashboard": "dashboard",
    "fine-tune": "fine-tune",
    "model-eval": "model-eval",
    "model-compare": "model-inference",
    "model-inference": "model-inference",
    "model-chat": "model-inference",
    "model-manage": "model-manage",
    "dataset-manage": "dataset",
    "data-process": "data-process",
    "data-convert": "data-convert",
    "compute": "compute",
    "hardware": "hardware",
    "users": "user-settings",
}

# These endpoints are the user-facing compute view used by training,
# inference, and evaluation forms. They only return the current user's
# assigned nodes/GPUs in the endpoint implementation, so they must remain
# available after an approval without granting access to the admin compute
# management page.
SELF_SERVICE_COMPUTE_PATHS = {
    "/compute/nodes",
    "/compute/gpus",
    "/compute/my-gpus",
}

# Resource actions are deliberately kept separate from module permissions.
# A user may be allowed to open a module while still lacking the action on a
# specific resource (for example, download or delete).
RESOURCE_ACTIONS = frozenset({
    "read",
    "write",
    "execute",
    "download",
    "export",
    "delete",
    "admin",
})
RESOURCE_ACTION_ALIASES = {"export": "download"}


def _extract_token(request: Request) -> str | None:
    """从 Authorization header 提取 token（格式: Bearer platform-token-{user_id}）。"""
    auth = request.headers.get("Authorization", "")
    token = auth.replace("Bearer ", "").strip()
    if token.startswith("platform-token-"):
        return token[len("platform-token-"):]
    return None

def _session_token(user_id: str, session_id: str) -> str:
    return f"platform-token-{user_id}.{session_id}"


def _record_auth_event(actor_id: str | None, action: str, reason: str, request: Request) -> None:
    """Best-effort security audit for authentication and permission denials."""
    try:
        get_platform_store().record_audit(
            action=action,
            actor_id=actor_id,
            target_type="auth",
            target_id=request.url.path,
            detail=reason,
            result="denied",
            reason=reason,
            request_id=request.headers.get("X-Request-ID"),
            ip=get_client_ip(request) or None,
        )
    except Exception:
        # An audit failure must never turn an authentication decision into an
        # accidental allow or an unrelated 500 response.
        pass


def get_current_user(request: Request) -> dict[str, Any]:
    """
    FastAPI 依赖：解析当前登录用户。
    - 公开路径（/health, /login 等）直接放行，返回匿名用户。
    - 无 token 或 token 无效时抛 401。
    - admin 用户标记为超级管理员，拥有全部权限。
    """
    path = request.url.path
    # 去掉路由前缀后判断
    for prefix in PUBLIC_PATHS:
        if path.endswith(prefix):
            return {"id": None, "username": "anonymous", "role": "viewer", "permissions": [], "protected": False}

    token_value = _extract_token(request)
    if not token_value:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="missing or invalid token")

    store = get_platform_store()
    user_id, _, session_id = token_value.partition(".")
    if session_id:
        with store.connect() as conn:
            session = conn.execute(
                "SELECT user_id, logout_at, expires_at FROM sessions WHERE id=?", (session_id,)
            ).fetchone()
        if not session or session["user_id"] != user_id or session["logout_at"]:
            _record_auth_event(user_id, "auth.session.denied", "session expired or logged out", request)
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="session expired")
        if session["expires_at"]:
            from datetime import datetime, timezone
            try:
                if datetime.fromisoformat(str(session["expires_at"]).replace("Z", "+00:00")) <= datetime.now(timezone.utc):
                    _record_auth_event(user_id, "auth.session.denied", "session expired", request)
                    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="session expired")
            except ValueError:
                pass
    with store.connect() as conn:
        user_row = conn.execute("SELECT * FROM users WHERE id=?", (user_id,)).fetchone()
    if user_row:
        user = store._user(user_row)
        if user.get("status") != "active":
            _record_auth_event(user_id, "auth.user.disabled", "user is not active", request)
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="user disabled")
        requested_tenant = request.headers.get("X-Tenant-ID", "").strip()
        if requested_tenant and (is_admin(user) or requested_tenant in user_tenant_ids(user)):
            user["tenant_id"] = requested_tenant
        # Keep the session identifier in request context so business audit
        # records can be traced back to the exact login session.
        if session_id:
            user["session_id"] = session_id
        path_parts = path.strip("/").split("/")
        # The API may be mounted directly at /modelTF or behind /api/v1/modelTF.
        # Locate the first known module segment instead of relying on a fixed index.
        segment = next((part for part in path_parts if part in MODULE_PERMISSIONS), "")
        required = MODULE_PERMISSIONS.get(segment)
        relative_path = "/" + "/".join(path_parts[path_parts.index(segment):]) if segment else path
        self_service_compute = relative_path.rstrip("/") in SELF_SERVICE_COMPUTE_PATHS
        if (
            required
            and not is_admin(user)
            and required not in (user.get("permissions") or [])
            and not self_service_compute
        ):
            _record_auth_event(user_id, "auth.permission.denied", f"missing permission: {required}", request)
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"missing permission: {required}")
        return user
    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="user not found")


def require_admin(current_user: dict[str, Any] = Depends(get_current_user)) -> dict[str, Any]:
    """FastAPI 依赖：要求当前用户是平台管理员。"""
    if is_admin(current_user):
        return current_user
    raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="admin permission required")


def require_tenant_admin(
    tenant_id: str,
    current_user: dict[str, Any] = Depends(get_current_user),
) -> dict[str, Any]:
    """Allow platform admins and active owner/admin tenant members."""
    if is_admin(current_user) or is_tenant_admin(current_user, tenant_id):
        return current_user
    raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="tenant admin permission required")


def is_admin(user: dict[str, Any]) -> bool:
    """判断用户是否为平台管理员；兼容历史 role=admin/protected 数据。"""
    return (
        user.get("platform_role") == "platform_admin"
        or user.get("role") == "admin"
        or user.get("protected", False)
    )


def user_tenant_ids(user: dict[str, Any]) -> set[str]:
    """Return the tenant scope of a user.

    Existing installations keep the primary tenant on ``users.tenant_id``.
    ``tenant_members`` is optional during the migration and adds memberships
    when the permission v2 schema is available.
    """
    if is_admin(user):
        return {"*"}
    primary_tenant = str(user.get("tenant_id") or "default")
    result: set[str] = set()
    user_id = user.get("id")
    if not user_id:
        return result
    try:
        store = get_platform_store()
        with store.connect() as conn:
            rows = conn.execute(
                "SELECT tenant_id FROM tenant_members "
                "WHERE user_id=? AND status='active' "
                "AND (expires_at IS NULL OR expires_at='' OR expires_at > NOW()::text)",
                (user_id,),
            ).fetchall()
            result.update(str(row["tenant_id"]) for row in rows if row.get("tenant_id"))
            if not result:
                active_tenant = conn.execute(
                    "SELECT id FROM tenants WHERE id=? "
                    "AND COALESCE(status, 'active')='active' "
                    "AND COALESCE(deleted_at, '')=''",
                    (primary_tenant,),
                ).fetchone()
                if active_tenant:
                    result.add(primary_tenant)
    except Exception:
        # Older databases are upgraded lazily. The primary users.tenant_id
        # remains a valid fallback until the additive table is available.
        result.add(primary_tenant)
    return result


def tenant_membership(user: dict[str, Any], tenant_id: str | None = None) -> dict[str, Any] | None:
    """Return the active membership for the selected tenant, if any."""
    if is_admin(user):
        return {"tenant_id": tenant_id or user.get("tenant_id") or "default", "role": "owner", "status": "active"}
    target = str(tenant_id or user.get("tenant_id") or "default")
    try:
        with get_platform_store().connect() as conn:
            row = conn.execute(
                "SELECT tenant_id, role, status, expires_at FROM tenant_members "
                "WHERE tenant_id=? AND user_id=? AND status='active'",
                (target, user.get("id")),
            ).fetchone()
        if not row:
            return None
        if row.get("expires_at"):
            from datetime import datetime, timezone
            try:
                if datetime.fromisoformat(str(row["expires_at"]).replace("Z", "+00:00")) <= datetime.now(timezone.utc):
                    return None
            except ValueError:
                return None
        return dict(row)
    except Exception:
        return None


def is_tenant_admin(user: dict[str, Any], tenant_id: str | None = None) -> bool:
    membership = tenant_membership(user, tenant_id)
    return bool(membership and membership.get("role") in {"owner", "admin"})


def is_tenant_admin_for_resource(resource_type: str, resource: dict[str, Any], user: dict[str, Any]) -> bool:
    if is_admin(user) or resource_type == "model":
        return False
    tenant_id = resource_tenant_id(resource_type, resource)
    return bool(tenant_id and is_tenant_admin(user, tenant_id))


def bind_active_tenant(payload: dict[str, Any], user: dict[str, Any]) -> dict[str, Any]:
    """Bind a new resource to the authenticated tenant context.

    Platform administrators may explicitly create a resource in another
    active tenant. Ordinary users can only use the tenant selected by the
    authenticated session/X-Tenant-ID header, never a client-supplied tenant
    id alone.
    """
    requested = str(payload.get("tenant_id") or user.get("tenant_id") or "default")
    if not is_admin(user) and requested not in user_tenant_ids(user):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="tenant access denied")
    payload["tenant_id"] = requested if is_admin(user) else str(user.get("tenant_id") or requested)
    return payload


def resource_record(resource_type: str, resource_id: str) -> dict[str, Any] | None:
    """Load a resource row for authorization without exposing storage details."""
    table = RESOURCE_TABLES.get(resource_type)
    if not table or not resource_id:
        return None
    store = get_platform_store()
    try:
        with store.connect() as conn:
            row = conn.execute(f"SELECT * FROM {table} WHERE id=?", (resource_id,)).fetchone()
    except Exception:
        return None
    if not row:
        return None
    result = dict(row)
    if result.get("deleted_at"):
        return None
    return result


def resource_tenant_id(resource_type: str, resource: dict[str, Any]) -> str | None:
    """Resolve tenant ownership, including legacy JSON-backed task rows."""
    # Administrator-created base models are platform shared. Online models
    # created by ordinary users retain tenant scope, so their visibility can
    # be limited to the owner and members of that tenant.
    if resource_type == "model" and str(resource.get("model_source") or "").lower() not in {"api", "online"}:
        return None
    tenant_id = resource.get("tenant_id")
    if tenant_id:
        return str(tenant_id)
    payload = resource.get("payload")
    if payload:
        try:
            data = json.loads(payload) if isinstance(payload, str) else payload
            if isinstance(data, dict) and data.get("tenant_id"):
                return str(data["tenant_id"])
        except (TypeError, ValueError, json.JSONDecodeError):
            pass
    return None


def resource_in_user_tenant(resource_type: str, resource: dict[str, Any], user: dict[str, Any]) -> bool:
    if is_admin(user):
        return True
    if resource_type == "model" and str(resource.get("model_source") or "").lower() not in {"api", "online"}:
        return True
    tenant_id = resource_tenant_id(resource_type, resource)
    if not tenant_id:
        # A missing tenant is not an implicit shared scope. The migration must
        # assign historical rows before ordinary users can access them.
        return False
    return "*" in user_tenant_ids(user) or tenant_id in user_tenant_ids(user)


def has_resource_access(
    resource_type: str,
    resource_id: str,
    user: dict[str, Any],
    permission: str = "read",
) -> bool:
    """
    检查用户对某资源是否有指定权限。
    - admin/protected 用户直接放行（旁路）。
    - 其他用户检查 acls 表中是否有对应授权。
    """
    if permission not in RESOURCE_ACTIONS:
        return False
    permission = RESOURCE_ACTION_ALIASES.get(permission, permission)
    if is_admin(user):
        return True

    store = get_platform_store()
    resource = resource_record(resource_type, resource_id)
    if not resource:
        return False
    # Some historical datasets were created before tenant_id was introduced.
    # They are not shared by default, but an explicit user ACL granted by an
    # administrator is still a valid compatibility path for that legacy row.
    legacy_unscoped = resource_type != "model" and not resource_tenant_id(resource_type, resource)
    if not resource_in_user_tenant(resource_type, resource, user) and not legacy_unscoped:
        return False
    if resource_type == "model":
        model_source = str(resource.get("model_source") or "").lower()
        if permission in {"read", "execute"}:
            # Local/registered base models are platform resources. For online
            # models, only administrator-created records are global; a normal
            # user's online model is shared with the active tenant below.
            if model_source not in {"api", "online"}:
                return True
            creator_id = str(resource.get("created_by") or "")
            if creator_id:
                with store.connect() as conn:
                    creator = conn.execute(
                        "SELECT platform_role, role, protected FROM users WHERE id=?",
                        (creator_id,),
                    ).fetchone()
                if creator and (
                    creator.get("platform_role") == "platform_admin"
                    or creator.get("role") == "admin"
                    or bool(creator.get("protected"))
                ):
                    return True
            # The tenant check above has already rejected models outside the
            # user's active tenant. Same-tenant online models are usable.
            if resource_tenant_id(resource_type, resource) in user_tenant_ids(user):
                return True
    if is_tenant_admin_for_resource(resource_type, resource, user):
        return True
    acls = store.get_acl(resource_type, resource_id)
    user_id = user.get("id")
    # ACL role principals are tenant roles, not platform roles.  Falling back
    # to the platform role keeps compatibility with old ACL records.
    membership = tenant_membership(user, resource_tenant_id(resource_type, resource))
    user_role = (membership or {}).get("role") or user.get("role")

    table_info = OWNER_TABLES.get(resource_type)
    if table_info and user_id:
        table, column = table_info
        with store.connect() as conn:
            row = conn.execute(f"SELECT {column} FROM {table} WHERE id=?", (resource_id,)).fetchone()
        if row:
            owner = row[column]
            if column == "payload":
                try:
                    owner = json.loads(owner or "{}").get("created_by")
                except (TypeError, ValueError):
                    owner = None
            if owner == user_id:
                return True

    for entry in acls:
        if entry.get("revoked_at"):
            continue
        if entry.get("expires_at"):
            from datetime import datetime, timezone
            try:
                if datetime.fromisoformat(str(entry["expires_at"]).replace("Z", "+00:00")) <= datetime.now(timezone.utc):
                    continue
            except ValueError:
                pass
        # 按 user 授权
        if entry.get("principal_type") == "user" and entry.get("principal_id") == user_id:
            if _permission_covers(entry.get("permission"), permission):
                return True
        # 按 role 授权
        if entry.get("principal_type") == "role" and entry.get("principal_id") == user_role:
            if _permission_covers(entry.get("permission"), permission):
                return True
    return False


def _permission_covers(granted: str | None, required: str) -> bool:
    """权限覆盖判断：write/execute 覆盖 read；admin 覆盖一切。"""
    if not granted:
        return False
    if granted == "admin":
        return True
    if granted == required:
        return True
    # write 覆盖 read
    if required == "read" and granted in ("write", "execute"):
        return True
    return False


def filter_accessible_resource_ids(
    resource_type: str,
    all_ids: list[str],
    user: dict[str, Any],
) -> list[str]:
    """
    从全部资源 ID 中过滤出当前用户可访问的 ID 列表。
    - admin 直接返回全部。
    - 普通用户查 acls 表取交集。
    """
    if is_admin(user):
        return all_ids

    if not all_ids:
        return []

    accessible = filter_accessible_resource_ids_batch(resource_type, all_ids, user)
    return [rid for rid in all_ids if rid in accessible]


def filter_accessible_resource_ids_batch(
    resource_type: str,
    resource_ids: list[str],
    user: dict[str, Any],
) -> set[str]:
    """Filter a list endpoint with a bounded set of SQL queries.

    The previous implementation called ``has_resource_access`` once per
    resource. Each call loaded the resource, tenant membership and ACL again,
    which made ordinary-user list pages slow on a remote PostgreSQL server.
    """
    if is_admin(user):
        return set(resource_ids)
    ids = list(dict.fromkeys(str(item) for item in resource_ids if item))
    if not ids:
        return set()

    table = RESOURCE_TABLES.get(resource_type)
    if not table:
        return set()
    placeholders = ",".join("?" for _ in ids)
    primary_tenant = str(user.get("tenant_id") or "default")
    tenant_ids = {primary_tenant}
    tenant_roles: dict[str, str] = {primary_tenant: str(user.get("role") or "")}
    user_id = str(user.get("id") or "")
    store = get_platform_store()
    with store.connect() as conn:
        memberships = conn.execute(
            "SELECT tenant_id, role FROM tenant_members "
            "WHERE user_id=? AND status='active' "
            "AND (expires_at IS NULL OR expires_at='' OR expires_at > NOW()::text)",
            (user_id,),
        ).fetchall()
        for membership in memberships:
            tenant_id = str(membership["tenant_id"] or "")
            if tenant_id:
                tenant_ids.add(tenant_id)
                tenant_roles[tenant_id] = str(membership["role"] or "")

        rows = conn.execute(
            f"SELECT * FROM {table} WHERE id IN ({placeholders})", tuple(ids)
        ).fetchall()
        row_by_id = {str(row["id"]): dict(row) for row in rows}
        acl_rows = conn.execute(
            "SELECT resource_id, principal_type, principal_id, permission, expires_at "
            f"FROM acls WHERE resource_type=? AND resource_id IN ({placeholders}) "
            "AND (revoked_at IS NULL OR revoked_at='')",
            (resource_type, *ids),
        ).fetchall()

    acl_by_resource: dict[str, list[dict[str, Any]]] = {}
    for row in acl_rows:
        acl_by_resource.setdefault(str(row["resource_id"]), []).append(dict(row))

    allowed: set[str] = set()
    for resource_id in ids:
        resource = row_by_id.get(resource_id)
        if not resource or resource.get("deleted_at"):
            continue
        if resource_type == "model":
            model_source = str(resource.get("model_source") or "").lower()
            if model_source not in {"api", "online"}:
                allowed.add(resource_id)
                continue
            creator_id = str(resource.get("created_by") or "")
            if creator_id:
                with store.connect() as conn:
                    creator = conn.execute(
                        "SELECT platform_role, role, protected FROM users WHERE id=?",
                        (creator_id,),
                    ).fetchone()
                if creator and (
                    creator.get("platform_role") == "platform_admin"
                    or creator.get("role") == "admin"
                    or bool(creator.get("protected"))
                ):
                    allowed.add(resource_id)
                    continue
            tenant_id = resource_tenant_id(resource_type, resource)
            if tenant_id and tenant_id in tenant_ids:
                allowed.add(resource_id)
                continue
        tenant_id = resource_tenant_id(resource_type, resource)
        if not tenant_id:
            # Legacy rows without a tenant are never implicitly visible. Only
            # a direct user ACL can expose one to its explicitly named user;
            # role ACLs remain blocked until the row is tenant-migrated.
            for entry in acl_by_resource.get(resource_id, []):
                expires_at = entry.get("expires_at")
                if expires_at:
                    try:
                        from datetime import datetime, timezone
                        if datetime.fromisoformat(str(expires_at).replace("Z", "+00:00")) <= datetime.now(timezone.utc):
                            continue
                    except ValueError:
                        continue
                if (
                    entry.get("principal_type") == "user"
                    and entry.get("principal_id") == user_id
                    and _permission_covers(entry.get("permission"), "read")
                ):
                    allowed.add(resource_id)
                    break
            continue
        if tenant_id not in tenant_ids:
            continue
        if tenant_roles.get(tenant_id) in {"owner", "admin"}:
            allowed.add(resource_id)
            continue
        owner = resource.get("created_by")
        if resource_type in {"eval", "fine-tune", "fine_tune_task", "compare", "inference"} and resource.get("payload"):
            try:
                payload = json.loads(resource["payload"]) if isinstance(resource["payload"], str) else resource["payload"]
                if isinstance(payload, dict) and payload.get("created_by"):
                    owner = payload["created_by"]
            except (TypeError, ValueError, json.JSONDecodeError):
                pass
        if owner == user_id:
            allowed.add(resource_id)
            continue
        role = tenant_roles.get(tenant_id) or str(user.get("role") or "")
        for entry in acl_by_resource.get(resource_id, []):
            expires_at = entry.get("expires_at")
            if expires_at:
                try:
                    from datetime import datetime, timezone
                    if datetime.fromisoformat(str(expires_at).replace("Z", "+00:00")) <= datetime.now(timezone.utc):
                        continue
                except ValueError:
                    continue
            user_match = entry.get("principal_type") == "user" and entry.get("principal_id") == user_id
            role_match = entry.get("principal_type") == "role" and entry.get("principal_id") == role
            if (user_match or role_match) and _permission_covers(entry.get("permission"), "read"):
                allowed.add(resource_id)
                break
    return allowed