scense/app/_archive_disabled_20261028_151505/auth_gate_strict.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""yuanjing 应用级「最后一层」会话鉴权闸门（Bug vRZrx0Ff9tz / R18 复测，P1 匿名读越权）。

为什么还需要这一层（根因链，实测于部署态 2026-10-25）
------------------------------------------------------
`app/route_guard_middleware.py::auth_required_middleware` 的判定逻辑本身是对的
（非公开 `.dspy` + 无会话 → 401），但部署态实测：

    NOCOOKIE GET /scense/api/list_scense.dspy -> 200 {"ok": true, "total": 0, "rows": []}
    NOCOOKIE GET /scense/api/session_create.dspy -> 401      # 同层对照
    NOCOOKIE GET /no/such/path.dspy            -> 401        # 框架层回 401

即「门禁按端点覆盖不全」而非整链缺失。语义中间件链的装配点是
`app/notfound_middleware.py::compose_guard_chain()`，它**按模块名**导入
`route_guard_middleware` 等中间件；部署态 `python app/scense.py` 的 sys.path 不含应用根时
导入失败 → `compose_guard_chain` 抛 RuntimeError → 外层降级为「只做 404 异常映射」的
handler-only 链 → **鉴权层整层缺席**，业务处理器直接执行 → 200。
（同一降级路径也是 2026-10-25 上一轮 business_status 层缺席的同源故障。）

本模块是**自持、零 app 包依赖**的兜底闸门：
* 只依赖 stdlib + aiohttp + cryptography（Fernet），不 import `app.*`；
* 默认 deny：除显式公开路径（登录/登出/健康探针）与静态资源外，**任何 `.dspy` 都必须
  带可验证的会话密文**，否则 401 并以统一错误信封返回（无业务字段）；
* 自检/解密异常一律 fail-closed（401），绝不 fail-open；
* 由 `app/scense.py::_load_semantic_middlewares()` **按绝对文件路径**加载并注册，
  与 `compose_guard_chain` 的装配成败解耦 —— 组合链修好后本层成为幂等冗余保险丝。

与 `route_guard_middleware` 的关系
----------------------------------
职责重叠是有意的（纵深防御）。责任重叠是有意的（纵深防御）。**公开路径集合不再由本文件持有**：
鉴权豁免清单的单一真源是 `app/auth_exempt.py`，本文件 `is_public_path()` 委托其
`is_exempt_path()`；本地 `PUBLIC_DSPY_PATHS` / `PUBLIC_PATH_PREFIXES` 已降级为只读镜像，
仅供 `scripts/selftest_auth_gate_list_scense.py` 交叉断言相等（防多清单漂移）。
只读镜像（仅供 `scripts/selftest_auth_gate_list_scense.py` 交叉断言相等）。
其余事实常量（密钥派生尾串、cookie 精确名、会话有效期）仍由该自测交叉断言，
改一处不改另一处会直接让自测 FAIL —— 以此防止实现漂移。
"""

import base64
import importlib.util
import json
import logging
import os
import time
import urllib.parse

from aiohttp import web

log = logging.getLogger("yuanjing.auth_gate_strict")

# ── 鉴权豁免清单：**单一真源** app/auth_exempt.py（Bug uifdBIMdmixZcmJEdV33O / T4）──
# 修复前本文件自持一份 PUBLIC_DSPY_PATHS / PUBLIC_PATH_PREFIXES 副本，与
# auth_route_gate / route_guard_middleware / notfound_middleware 的另外几份副本各自演化
# （R57b / R59：主线豁免、兜底路径不豁免 → 登录端点仍 401 auth_unavailable）。
# [fix rvtGi1mlUsTwQrXnybZj6 / R61 · QC#1] 现在本文件**不再持有第二份清单**：
# 判定一律委托 `auth_exempt.is_exempt_path()`，下面两个名字只是它的**只读镜像**，
# 仅用于自测交叉断言与诊断输出（自测 `scripts/selftest_auth_gate_list_scense.py`
# 断言三者相等，防止再次各自演化）。
def _load_auth_exempt():
    """加载豁免清单单一真源（部署态 sys.path 可能不含应用根，禁裸包导入）。"""
    try:
        from app import auth_exempt as mod                    # type: ignore
        return mod
    except Exception:                                          # noqa: BLE001
        spec = importlib.util.spec_from_file_location(
            "yj_auth_exempt_strict",
            os.path.join(os.path.dirname(os.path.abspath(__file__)), "auth_exempt.py"))
        mod = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(mod)
        return mod


try:
    EXEMPT = _load_auth_exempt()
except Exception as _load_exc:                                 # noqa: BLE001
    # 清单都加载不了 → 不做任何豁免（fail-closed），但必须显式告警，不静默。
    log.error("[auth_strict] 豁免清单单一真源加载失败（所有 .dspy 均按需会话处理）: %s",
              _load_exc)
    EXEMPT = None

def _derive_exempt_manifest(mod):
    """由单一真源 `auth_exempt.manifest()` **自动派生**镜像（禁止手工复制字面量）。

    [fix rvtGi1mlUsTwQrXnybZj6 / T4-A] 下面两个 PUBLIC_* 名字仅是**派生镜像**
    （自测交叉断言 + 诊断输出），判定一律走 `EXEMPT.is_exempt_path()`。
    """
    if mod is None:
        return {"revision": "unavailable", "dspy_paths": frozenset(),
                "prefixes": (), "wildcard_free": False}
    try:
        mf = mod.manifest()
    except Exception:                                          # noqa: BLE001
        return {
            "revision": getattr(mod, "EXEMPT_REVISION", "unavailable"),
            "dspy_paths": frozenset(getattr(mod, "AUTH_EXEMPT_DSPY_PATHS", ()) or ()),
            "prefixes": tuple(getattr(mod, "AUTH_EXEMPT_PREFIXES", ()) or ()),
            "wildcard_free": False,
        }
    return {
        "revision": str(mf.get("revision") or "unavailable"),
        "dspy_paths": frozenset(mf.get("dspy_paths") or ()),
        "prefixes": tuple(mf.get("prefixes") or ()),
        "wildcard_free": bool(mf.get("wildcard_free", False)),
    }


EXEMPT_MANIFEST = _derive_exempt_manifest(EXEMPT)

#: **交叉断言镜像**（非清单本体，由 manifest() 派生）：仅为自测比对与诊断输出保留。
PUBLIC_DSPY_PATHS = EXEMPT_MANIFEST["dspy_paths"]

#: **交叉断言镜像**（非清单本体，由 manifest() 派生）：同上。
PUBLIC_PATH_PREFIXES = EXEMPT_MANIFEST["prefixes"]

#: 受闸门约束的后缀。`.dspy` 是唯一需要会话的动态端点类型；
#: `.ui/.js/.css/.html/.json` 属页面与静态资产，由框架会话层处理（避免锁死登录页）。
GATED_SUFFIX = ".dspy"

#: 会话 cookie 精确名（与 ahserver AuthAPI.setupAuth / route_guard 同源）。
SESSION_COOKIE_NAME_DEFAULT = "AIOHTTP_SESSION"
SESSION_COOKIE_MAX_AGE = 24 * 60 * 60

#: 复刻 ahserver AuthAPI.setupAuth 的 Fernet 密钥派生（port 前缀 + 固定尾串，共 32 字节）。
SESSION_SECRET_TAIL = b"iqwertyuiopasdfghjklzxcvbnm12345"

#: 闸门总开关（仅排障用；生产禁开）。与 route_guard 同名环境变量。
GATE_ENV = "YJ_AUTH_GATE"

#: 应用根（本文件 → app/ → 应用根）。
APP_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def gate_enabled():
    """总开关：默认开；`YJ_AUTH_GATE=off|0|false|no` 才关（排障专用）。"""
    return (os.environ.get(GATE_ENV) or "on").strip().lower() not in (
        "off", "0", "false", "no")


def normalize_path(path):
    """归一化请求路径，封堵写法绕过。

    `/scense/api/list_scense.dspy/`、`/SCENSE/API/LIST_SCENSE.DSPY`、
    `//scense/api/list_scense.dspy`、`/scense/api/%6cist_scense.dspy` 都必须
    与 `/scense/api/list_scense.dspy` 得到同一判定，不能因为写法不同而绕过闸门。
    """
    p = str(path or "")
    p = urllib.parse.unquote(p)
    p = p.split("?", 1)[0].split("#", 1)[0]
    p = "/" + "/".join([seg for seg in p.split("/") if seg])
    if p != "/":
        p = p.rstrip("/")
    return p.lower() if p.lower().endswith(GATED_SUFFIX) or p.startswith("/rbac/") else p


def is_public_path(path):
    """显式公开/豁免路径判定 —— **委托单一真源** `auth_exempt.is_exempt_path()`。

    [fix rvtGi1mlUsTwQrXnybZj6 / R61 · QC#1] 本函数**不再**用本地常量判定：
    历史实现在这里持有第二份清单，与 route_guard / auth_route_gate /
    notfound 兜底闸门的副本各自演化（R57b / R59 多清单漂移同源缺陷）。
    现在只有一份清单（auth_exempt），本层只是调用方。

    :param path: 请求路径（可含 query / 尾斜杠 / 编码，内部归一化）
    :return: bool —— 豁免 True；清单不可用或判定异常一律 False（fail-closed）
    """
    if EXEMPT is None:
        return False
    try:
        return bool(EXEMPT.is_exempt_path(normalize_path(path)))
    except Exception as exc:                                   # noqa: BLE001
        log.error("[auth_strict] 豁免判定异常（按不豁免处理）: %s", exc)
        return False


def public_reason(path):
    """豁免依据（未豁免返回空串）—— 委托单一真源，供自测/诊断输出。"""
    if EXEMPT is None:
        return ""
    try:
        return EXEMPT.exempt_reason(normalize_path(path)) or ""
    except Exception:                                          # noqa: BLE001
        return ""


def is_gated_path(path):
    """该路径是否需要会话凭据。

    [R62 / 匿名读全站 P0，需求 1+2] 覆盖面从「仅 `.dspy` 后缀」扩为
    **后缀 ∪ 业务 API 前缀**，判定顺序（顺序即语义）：

      1. 显式公开路径（登录/登出/健康探针，单一真源 `auth_exempt`）→ 不需会话；
      2. `.dspy` 动态端点 → 需要会话（历史行为，保持）；
      3. 命中 `auth_exempt.PROTECTED_API_PREFIXES`（/scense /runtime /scene /world
         /entity）且路径含 `/api/` 段或以 `.dspy` 结尾 → **同样需要会话**。

    第 3 条是本次修复的关键：旧实现**只看后缀**（`if not endswith('.dspy'): return
    False`），任何「无 `.dspy` 后缀的业务 API」（反向代理/重写规则下的
    `/runtime/api/build_entity_tree`、`/scense/api/...` 变体写法）都会被判成
    「非 `.dspy` → 放行」，于是匿名 200 返回业务数据。按需求 2，覆盖面必须按
    **前缀**枚举而非按后缀，且不得只补被测的 3 条。
    """
    p = normalize_path(path)
    if is_public_path(p):
        return False
    if p.lower().endswith(GATED_SUFFIX):
        return True
    # 前缀覆盖面：判定委托单一真源，禁在本文件复制清单
    try:
        if EXEMPT is not None and EXEMPT.is_protected_api_path(p):
            return True
    except Exception as exc:                                   # noqa: BLE001
        # 判不了 → 按受保护处理（fail-closed，绝不放行业务数据）
        log.error("[auth_strict] 业务 API 前缀判定异常（按受保护处理）: %s", exc)
        return True
    return False


def _app_conf():
    """读 conf/config.json（读不到返回 {}，绝不因配置问题抛到中间件链）。"""
    try:
        with open(os.path.join(APP_ROOT, "conf", "config.json"), encoding="utf-8") as fp:
            return json.load(fp) or {}
    except Exception:                                          # noqa: BLE001
        return {}


def session_cookie_names():
    """允许承载会话的 cookie **精确名**（默认名 + conf 的 cookie_name）。"""
    names = [SESSION_COOKIE_NAME_DEFAULT]
    try:
        cfg = _app_conf()
        name = ((((cfg.get("website") or {}).get("session") or {})
                 .get("cookie_name")) or "").strip()
        if name and name not in names:
            names.append(name)
    except Exception:                                          # noqa: BLE001
        pass
    return tuple(names)


def session_key():
    """Fernet 密钥：port 前 32 字节 + 固定尾串补齐（复刻 ahserver 派生）。"""
    port = str(((_app_conf().get("website") or {}).get("port")) or "")
    b = port.encode("utf-8")[:32]
    cnt = max(0, 32 - len(b))
    return base64.urlsafe_b64encode(b + SESSION_SECRET_TAIL[:cnt])


def verify_cookie(raw_cookie):
    """校验会话密文 → `(ok, reason)`。

    reason ∈ ``no_session`` / ``forged_session`` / ``expired_session`` /
    ``session_decode_unavailable``（cryptography 不可用 → fail-closed）。
    """
    if not raw_cookie or not str(raw_cookie).strip():
        return False, "no_session"
    try:
        from cryptography.fernet import Fernet, InvalidToken
    except Exception:                                          # noqa: BLE001
        return False, "session_decode_unavailable"
    try:
        fernet = Fernet(session_key())
        token = str(raw_cookie).strip().encode("utf-8")
        try:
            data = json.loads(fernet.decrypt(token, ttl=SESSION_COOKIE_MAX_AGE).decode("utf-8"))
            return (True, "ok") if isinstance(data, dict) else (False, "forged_session")
        except InvalidToken:
            data = json.loads(fernet.decrypt(token).decode("utf-8"))
            if not isinstance(data, dict):
                return False, "forged_session"
            created = int(data.get("created") or 0)
            if created and (time.time() - created) > SESSION_COOKIE_MAX_AGE:
                return False, "expired_session"
            return False, "forged_session"
    except Exception:                                          # noqa: BLE001
        return False, "forged_session"


def _has_session(request):
    """请求是否携带可验证的会话 → `(ok, reason)`。"""
    cookies = getattr(request, "cookies", None) or {}
    for name in session_cookie_names():
        val = cookies.get(name)
        if val:
            return verify_cookie(val)
    return False, "no_session"


def _deny(request, reason):
    """统一 401 错误信封（**不含任何业务字段**）。

    hint 自指修正（R53 / QC#4）：
      * 登录端点自身（`/rbac/user/up_login.dspy` 等公开路径）被拒 → **不是未登录**，
        而是服务端鉴权闸门/会话组件异常（fail-closed 掩盖了真实故障）。此时 hint
        必须指向「闸门异常」而非「请先 POST 登录」，否则形成"先登录才能登录"的自指死循环。
      * 其余受保护端点 → 保持"请先登录"的指引。
    """
    path = normalize_path(getattr(request, "path", "") or "")
    log.warning("[auth_strict] 拒绝匿名/无效会话请求: %s %s reason=%s",
                getattr(request, "method", "-"), path, reason)
    if is_public_path(path):
        hint = ("该路径是公开登录端点，401 表示服务端鉴权闸门异常（fail-closed），"
                "不是未登录。请查看服务端日志 [auth_strict] 与 /healthz 的 gate/chain 诊断字段")
    else:
        hint = "未登录/会话无效：请先 POST /rbac/user/up_login.dspy 取得会话 cookie"
    resp = web.json_response(
        {
            "ok": False,
            "status": 401,
            "error": reason,
            "path": path,
            "hint": hint,
        },
        status=401,
    )
    resp.headers["WWW-Authenticate"] = 'Session realm="yuanjing"'
    return resp


def _is_response_contract_error(exc):
    """异常是否属于「中间件把非 Response（handler 函数本体）交回上游」这一类契约错误。"""
    txt = "%s: %s" % (type(exc).__name__, exc)
    for key in ("Expect response", "should return a response", "not <class 'function'>",
                "not a Response", "ResponseContractError", "__mw_signature_adapter__"):
        if key in txt:
            return True
    return False


def _internal_error(request, exc):
    """闸门内部异常 → **500**（不是 401）。带 traceback 与层名，供运维直接定位。

    :param request: aiohttp Request
    :param exc: 捕获到的异常
    :return: aiohttp web.Response(status=500, error=auth_gate_internal_error)
    """
    path = "-"
    try:
        path = normalize_path(getattr(request, "path", "") or "") or "-"
    except Exception:                                          # noqa: BLE001
        pass
    contract = _is_response_contract_error(exc)
    log.error("[auth_strict] 闸门内部异常（非会话缺失）→ 500 %s %s contract_error=%s: %s: %s",
              getattr(request, "method", "-"), path, contract,
              type(exc).__name__, exc, exc_info=True)
    detail = "%s: %s" % (type(exc).__name__, exc)
    if contract:
        hint = ("中间件链响应契约错误：某层把未 await 的 handler 当作 Response 返回。"
                "查日志 [mw-signature]/[auth_strict] 里的 layer= 字段定位该层")
    else:
        hint = ("服务端鉴权闸门内部异常（既不是未登录，也不是会话缺失）。"
                "查服务端日志 [auth_strict] 的 traceback")
    resp = web.json_response(
        {
            "ok": False,
            "status": 500,
            "error": "auth_gate_internal_error",
            "path": path,
            "detail": detail,
            "contract_error": contract,
            "hint": hint,
        },
        status=500,
    )
    try:
        resp.headers["X-Auth-Gate-Error"] = "internal"
    except Exception:                                          # noqa: BLE001
        pass
    return resp


@web.middleware
async def auth_gate_strict_middleware(request, handler):
    """最后一层会话闸门：非公开 `.dspy` 无有效会话 → 401（fail-closed）。

    判定顺序（顺序即语义）：
      1. 总开关关闭（排障）→ 放行；
      2. 非 `.dspy`（页面/静态）→ 放行；
      3. 显式公开路径（登录/登出/健康探针）→ 放行；
      4. 其余 `.dspy` → 校验会话密文，失败 401，成功放行给内层链/业务处理器。
    """
    # [fix Zhpd9MXyeGtznDMqini6X / R49] 与 route_guard 同源修法：
    # 只把「闸门自身判定/会话校验」的异常 fail-closed 成 401，
    # `await handler(request)`（下游业务处理器）必须在 try 之外，
    # 否则下游 500 / aiohttp `Expect response, not <class 'function'>`
    # 会被改写成 401 auth_unavailable，把真实故障整层遮蔽（R49 矩阵
    # 229×401、42 条 500 无法判定的机械原因）。
    try:
        if not gate_enabled():
            return await handler(request)
        path = normalize_path(getattr(request, "path", "") or "")
        if not is_gated_path(path):
            return await handler(request)
        ok, reason = _has_session(request)
    except Exception as exc:                                   # noqa: BLE001
        # fail-closed 仍成立（绝不放行业务数据），但**错误码不再冒充鉴权失败**。
        # 历史（已废除）：R50 曾把公开路径的闸门异常 `return await handler(request)`
        # 放行，其余一律 401 auth_unavailable —— 两者都在掩盖真实故障，R60 起废除。
        # [fix rvtGi1mlUsTwQrXnybZj6 / R60] 语义分离：**闸门内部异常 ≠ 会话缺失**。
        # 旧实现把任何异常（典型：链装配错误 `Expect response, not <class 'function'>`）
        # 一律降级成 401 auth_unavailable，于是「登录端点自己 401」被固化，
        # 真实故障（某层把未 await 的 handler 当 Response 返回）被 401 完全遮蔽 ——
        # R58 实测 logs/scense.log:1735-1737 就是这条遮蔽路径。
        # 现在：会话缺失/无效 → 401（见下方 ok 分支）；闸门自身或内层链异常 →
        # 500 + error=auth_gate_internal_error + 完整 traceback，绝不改写成鉴权失败。
        return _internal_error(request, exc)
    if ok:
        return await handler(request)
    return _deny(request, reason)