scense/app/_archive_disabled_20261028_151505/entry_guard_r19.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""R19 运行入口硬守卫链（Bug nrOjOBSxzsHGa1HgElzmn / DGzTjOkr_OYM4vdOYcl2_ /
WhOYy9nvG-EfNgZMvbJbO 的**运行入口根因修复**）。

## 为什么还要这个文件（根因结论，不是又写一遍语义实现）

三条语义的实现早就存在于仓库里：

    /healthz 非 GET → 405 + Allow     app/entry_semantics.py          (BUILTIN_GET_ONLY_PATHS)
    业务信封 status>=400 → HTTP 状态行 app/business_status_middleware.py (middleware + 信号)
    非公开 .dspy 无会话 → 401          app/auth_gate_strict.py / route_guard_middleware.py

但目标机真实运行入口是 `app/scense.py`（`ps -ef` 唯一进程 `venv/bin/python app/scense.py`），
R10 复测 5/5 全部在部署态失效（HTTP=404 / 200 / 503 / 200）。三处装配缺口：

1. **404 先于 405**：`entry_semantics` 的判定顺序是「资源是否存在」→ 再判方法。
   `/healthz` 是 `app.router.add_get()` 注册的**动态路由**，不在 wwwroot 文件树里，
   于是「资源不存在」分支先命中 → 404 no such resource，`BUILTIN_GET_ONLY_PATHS` 的
   405 分支根本走不到。`/healthz` 自检里 `http_semantics.installed=true, added=2`
   指的是**装配两层**（`entry_semantics` 零依赖层 + `semantic_wiring` 完整层）成功，
   不是「405 分支命中」，因此自检为真而行为为假。
2. **业务 400 层缺席**：`business_status_middleware` 从未进入
   `app/scense.py::_load_semantic_middlewares()` 的构造期清单（清单里只有
   notfound / http_semantics / auth_gate_strict 三项）。信号层
   `_install_business_status_signal` 挂在 `_on_ahapp_built` 钩子里，而构造期路径
   `entry_semantics_ctor.make_app(mws)` 成功时钩子只做幂等核验，不保证 `on_response_prepare`
   被挂到最终 aiohttp Application 上；钩子与中间件两条路径**都不在链上** → 4/4 端点
   缺参仍 HTTP 200（body 里 status=400）。
3. **鉴权闸门不在链上**：同上，`auth_gate_strict_middleware` 虽然在清单里，但装配走
   `entry_semantics_ctor` 构造期 + `notfound_middleware.compose_guard_chain()` 包导入
   `__import__("app." + 模块名)`，部署态 `python app/scense.py` 的 sys.path 不含应用根
   → 包导入静默失败 → 组合链降级「仅 404」→ 匿名 `GET /scense/api/list_scense.dspy`
   返回 200（同轮 `session_create.dspy` 返 401，说明闸门代码本身没坏，是没上链）。

## 本文件的设计要点（为什么这样修才在运行入口生效）

* **零应用包依赖**：只 import 标准库 + aiohttp + 同目录文件按**绝对路径**加载
  （`auth_gate_strict.py` 复用已验证的会话校验），彻底绕开 sys.path 陷阱；
* **顺序即语义**：作为 `app.middlewares` 的**第 1 位（最外层）**装配，
  「方法不允许」与「未登录」在业务处理器/404 判定之前返回，天然修复 404 抢答；
* **不掩错**：只处理 405 / 401 / 业务信封 4xx 回写三类；其它异常一律 re-raise，
  真实后端缺陷仍暴露为 500。
"""
import importlib.util
import json
import logging
import os
import urllib.parse

from aiohttp import web

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

APP_DIR = os.path.dirname(os.path.abspath(__file__))

#: 内建「只允许读」的路径（精确匹配，禁通配）。新增内建 GET 端点必须逐条登记。
GET_ONLY_PATHS = frozenset((
    "/healthz",
    "/health",
))

#: 405 时下发的 Allow 头（清单顺序固定，便于断言）。
ALLOW_HEADER = "GET, HEAD, OPTIONS"

#: 视为「读」的安全方法——非此集合即方法不允许。
SAFE_METHODS = frozenset(("GET", "HEAD", "OPTIONS"))

#: 业务信封回写只作用于 .dspy（页面/静态资产语义不动）。
ENVELOPE_SUFFIX = ".dspy"


# ---------------------------------------------------------------- 纯逻辑（可离线桩测）
def normalize_path(path):
    """归一化路径：去掉重复斜杠 / 尾斜杠 / 查询串，并解 URL 编码。

    与 `auth_gate_strict.normalize_path` 同口径，保证 `/HEALTHZ/`、
    `//healthz`、`/healthz?x=1` 都能命中 GET_ONLY_PATHS，不能靠写法绕过 405。
    """
    p = urllib.parse.unquote(str(path or ""))
    p = p.split("?", 1)[0].split("#", 1)[0]
    segs = [s for s in p.split("/") if s]
    p = "/" + "/".join(segs)
    return p.lower() if p != "/" else p


def method_not_allowed(path, method):
    """(path, method) 是否需要 405。返回 Allow 头字符串；不需要则 None。

    判定在**任何资源存在性检查之前**完成——这是本文件相对 entry_semantics 的
    关键差异（后者先判存在性，动态路由 /healthz 被判成「不存在」→ 404）。
    """
    p = normalize_path(path)
    if p not in GET_ONLY_PATHS:
        return None
    if str(method or "").upper() in SAFE_METHODS:
        return None
    return ALLOW_HEADER


def envelope_status(raw_body):
    """从响应体解析业务信封 status（只认 JSON 对象的 4xx/5xx 整数 status）。"""
    if not raw_body:
        return None
    try:
        if isinstance(raw_body, bytes):
            raw_body = raw_body.decode("utf-8", "replace")
        obj = json.loads(raw_body)
    except (TypeError, ValueError):
        return None
    if not isinstance(obj, dict):
        return None
    code = obj.get("status")
    if isinstance(code, bool):
        return None
    if isinstance(code, int) and 400 <= code <= 599:
        return code
    return None


def should_rewrite_status(path, http_status, raw_body):
    """是否需要把 HTTP 状态行回写成信封 status。返回目标码或 None。"""
    p = normalize_path(path)
    if not p.endswith(ENVELOPE_SUFFIX):
        return None
    if int(http_status or 0) != 200:
        return None
    return envelope_status(raw_body)


# ---------------------------------------------------------------- 会话闸门（复用已验证实现）
_GATE = {"mod": None}


def _auth_gate():
    """按绝对路径加载 auth_gate_strict.py（部署态 sys.path 不含应用根，禁包导入）。"""
    if _GATE["mod"] is not None:
        return _GATE["mod"]
    path = os.path.join(APP_DIR, "auth_gate_strict.py")
    spec = importlib.util.spec_from_file_location("yj_auth_gate_strict_r19", path)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    _GATE["mod"] = mod
    return mod


def auth_verdict(path, request):
    """鉴权判定：返回 "public" / "allow" / "deny"。

    与 auth_gate_strict 同一套公开清单与会话校验（精确路径，禁通配）；
    任何异常都 fail-closed 成 deny（绝不 fail-open）。
    """
    try:
        gate = _auth_gate()
        if not gate.gate_enabled():
            return "public"
        if gate.is_public_path(path):
            return "public"
        if not gate.is_gated_path(path):
            return "public"
        # [R21 401 根因补修] auth_gate_strict._has_session() 返回的是
        # **(ok, reason) 元组**，不是布尔。原写法 `"allow" if gate._has_session(request)`
        # 让任何非空元组（含 (False, "no_session")）都为真 → 匿名请求被判 allow，
        # 401 在入口层彻底失效（即便会话校验实现正确）。此处显式取第 0 位布尔，
        # 对列表/字典/布尔三种返回形态都做归一，绝不依赖对象真值。
        verdict = gate._has_session(request)
        if isinstance(verdict, (tuple, list)):
            ok = bool(verdict[0]) if verdict else False
        elif isinstance(verdict, dict):
            ok = bool(verdict.get("ok"))
        else:
            ok = bool(verdict)
        return "allow" if ok else "deny"
    except Exception as exc:                                      # noqa: BLE001
        log.error("[guard19] 鉴权判定异常，fail-closed 401: %s", exc)
        return "deny"


# ---------------------------------------------------------------- 中间件本体
def _json_response(status, payload, extra_headers=None):
    headers = {"Cache-Control": "no-store"}
    if extra_headers:
        headers.update(extra_headers)
    return web.json_response(payload, status=status, headers=headers)


@web.middleware                       # ← aiohttp>=3.9 新式中间件标记（见下方 R19d 说明）
async def entry_guard_middleware(request, handler):
    """入口硬守卫：405（方法语义）→ 401（会话闸门）→ 业务信封 4xx 回写。

    [fix-iter17 r19d] 必须带 `@web.middleware`（= 设置 `__middleware_version__ = 1`）。
    aiohttp >= 3.9（requirements 锁 `aiohttp>=3.9,<4.0`）在
    `Application._prepare_middleware()` 里按该标记区分新旧式：**未标记的普通协程函数
    会被当成「旧式工厂」** m(app, handler) 调用，于是本函数收到 request=app、
    handler=真实 handler，`await handler(request)` 抛
    `TypeError: 'Response' object is not callable` → 每个请求 500；
    /healthz 的 405 退化由 aiohttp 路由自身给出（Allow 只有 GET,HEAD，body 为
    `405: Method Not Allowed` 纯文本），闸门与信封回写整层失效。
    实测证据（本机真实执行回显，非引用）：
    projects/proj_qEVGJSL3/docs/02-develop/evidence/r19_entry_assembly_verify_20261026.txt
    （由 apps/yuanjing/scripts/verify_entry_assembly_r19.py 生成，含 TOTAL=/PASS=/FAIL= 行）
    """
    path = getattr(request, "path", "") or ""

    # ① 方法语义（先于 404/鉴权；/healthz 等内建只读路径非读方法 → 405 + Allow）
    #
    # [QC#7] 原写法 `if allow == "GET, HEAD, OPTIONS"` 把返回值退化成布尔信号，
    # 与 ALLOW_HEADER 形成两处字面量（改 ALLOW_HEADER 后判定静默漂移），且后面
    # 再判一次 OPTIONS（method_not_allowed 已用 SAFE_METHODS 排除了 OPTIONS/GET/HEAD）。
    # 现在：返回值即单一真源——None = 不需要 405，非 None = 需要 405 且它就是 Allow 头。
    allow = method_not_allowed(path, getattr(request, "method", ""))
    if allow:
        return _json_response(
            405,
            {"status": 405, "success": False, "reason": "method not allowed",
             "path": normalize_path(path), "allow": allow},
            {"Allow": allow},
        )

    # ② 会话闸门（非公开 .dspy 无会话 → 401，fail-closed）
    verdict = auth_verdict(path, request)
    if verdict == "deny":
        return _json_response(
            401,
            {"status": 401, "success": False, "reason": "unauthorized",
             "path": normalize_path(path)},
        )

    # ③ 业务信封 status>=400 → HTTP 状态行（只改状态行，不改 body）
    resp = await handler(request)
    try:
        target = should_rewrite_status(path, getattr(resp, "status", 200),
                                      getattr(resp, "body", None))
        if target is not None and hasattr(resp, "set_status"):
            resp.set_status(target)
            log.info("[guard19] 业务信封 status=%s → HTTP %s path=%s",
                     target, target, normalize_path(path))
    except Exception as exc:                                      # noqa: BLE001
        log.warning("[guard19] 状态行回写失败（不影响 body）: %s", exc)
    return resp