scense/app/_archive_disabled_20261028_151505/auth_route_gate.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""统一鉴权门禁（存在性 → 鉴权 两段式；404 / 401 / 403 三态互不覆盖）。

覆盖缺陷族（T2 鉴权门禁族）
---------------------------
* LonNYvRxXk9EttSexkwkV：未登录 GET /scense/api/list_scense.dspy 返 200（应 401）
* LIyicannYDhJiXXkjTQ0N：不存在端点返 401 而非 404（鉴权先于路由）
* bj-n9DhoDKGz_lMHiUmbq：匿名 GET /scene/api/get_search_status.dspy 返 200+rows
* N4teiYPv2VUo2OWkoR8Fw：匿名 POST /scene/api/create_scene.dspy 进入业务分支返 200
* QuOjNlynfq3i8P7zdU2U4：匿名读 3 端点均 200 返业务数据

设计要点（为什么另起一个「统一门禁」而不是继续在旧 gate 上打补丁）
------------------------------------------------------------------
旧实现（route_guard_middleware.auth_required_middleware）依赖
`PROTECTED_DSPY_PATHS` 逐条点名 + 行为分支判定，结果是「点名的进闸、没点名的漏出去」，
于是 list_scense / get_search_status / create_scene 各自以 200 漏过，同一站内
"不存在路径 401、真实业务端点 200" 的语义自相矛盾。

本模块改为**默认拒绝（default-deny）**：

    *.dspy  ⇒  除非在 PUBLIC_DSPY_PATHS（逐条精确、无通配）中，
               否则一律「必须有有效会话」。
             新增端点无需登记即自动纳管（get_search_* / list_* / create_* 全部覆盖）。

判定顺序（硬约束，两者不得互相覆盖）：

    ① normalize 路径
    ② 后缀不是 .dspy            → 放行（静态资源 / .ui 页面 / /healthz 交给既有组件）
    ③ 命中公开白名单（auth_exempt.is_exempt_path，精确/段边界，无通配）
                                → 放行（**先于**存在性与鉴权：登录链路不得自锁）
    ④ 资源不存在                → 404   （先于鉴权：路径写错不应伪装成"未登录"）
    ⑤ 无有效会话 / 会话无效     → 401 + WWW-Authenticate
    ⑥ 有有效会话                → 放行（后续 RBAC 权限不足由 ahserver 判 403）

403 的职责边界：本门禁只判「有没有身份」，不判「这个身份有没有该端点权限」——
后者是 RBAC（permission / rolepermission）的事，返回 403 由框架层给出。
两者分工在矩阵表里体现为 `登录态=有, 无权限 → 403（框架）`。

安装位置
--------
`install(app)` 把中间件插到 `app.middlewares` **首位（最外层）**：
aiohttp 的中间件列表首位 = 最外层，必须先于 ahserver 的鉴权层执行，
否则框架自身的 401（对不存在路径也 401）会抢在我们的 404 之前返回 -
这正是 LIyicannYDhJiXXkjTQ0N 的根因（「鉴权前置于路由判定」）。

零硬编码：白名单路径、后缀、cookie 名全部是本文件常量或从 conf/config.json 读取；
不写任何 host/port/域名。
"""

import sys
import json
import os
import importlib.util
import logging

# R52 QC#4：本文件多处 `log.error(...)` 落在 _load_contract_module / _new_style 的
# **兜底路径**上（设计意图是「宁可抛错也不静默放行」）。此前未定义 `log`、也未
# import logging，一旦真的走到兜底分支会先抛 `NameError: name 'log' is not defined`，
# 把原始异常覆盖成模块级 import 失败 → 应用启动即崩。与 get_only_405.py 对称定义。
log = logging.getLogger("scense.auth_route_gate")

# --------------------------------------------------------------------------- #
# 常量（全部逐条精确，**禁止通配符**：QC 硬门禁 #6）
# --------------------------------------------------------------------------- #

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

#: 关闸开关（off/0/false/no 关闭；默认开启）
GATE_ENV = "YUANJING_AUTH_ROUTE_GATE"

#: 只对 .dspy 端点做门禁（.ui 页面由会话层处理，避免把登录页自身锁死）
PROTECTED_SUFFIXES = (".dspy",)

# ── 鉴权豁免清单：**单一真源** app/auth_exempt.py（T4 / Bug uifdBIMdmixZcmJEdV33O）──
# 修复前本文件自带一份 PUBLIC_DSPY_PATHS 副本，与 route_guard_middleware /
# auth_gate_strict / notfound_middleware 内联兜底闸门的另外几份副本**各自演化**，
# 结果「主线豁免、兜底路径不豁免」→ 登录端点在某些链路上仍 401。现在统一委托。
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", os.path.join(APP_DIR, "auth_exempt.py"))
        mod = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(mod)
        return mod


EXEMPT = _load_auth_exempt()


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

    [fix rvtGi1mlUsTwQrXnybZj6 / T4-A] 本文件的 PUBLIC_DSPY_PATHS /
    PUBLIC_PATH_PREFIXES 只是**派生镜像**，用于自测交叉断言与诊断输出；
    判定一律走 `EXEMPT.is_exempt_path()`。真源不可用/异常 → 空镜像 + fail-closed。
    """
    if mod is None:
        return {"revision": "unavailable", "dspy_paths": frozenset(),
                "prefixes": (), "wildcard_free": False}
    try:
        mf = mod.manifest()
    except Exception:                                          # noqa: BLE001
        # manifest 不可用时退回真源的属性（仍是同一份数据，不是本地字面量）
        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": bool(getattr(mod, "EXEMPT_REVISION", "")),
        }
    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"]

#: 豁免口径版本（写进诊断与自测输出；同样来自 manifest()）
EXEMPT_REVISION = EXEMPT_MANIFEST["revision"]

#: 中间件函数名（供装配后置核验使用）
MIDDLEWARE_NAME = "auth_route_gate_middleware"

#: 会话凭据来源（与 ahserver/rbac 对齐；具体 cookie 名从 conf 读）
DEFAULT_SESSION_COOKIE = "AIOSESSION"


# --------------------------------------------------------------------------- #
# 路径工具
# --------------------------------------------------------------------------- #

def _app_root():
    """应用根目录（= app/ 的上级）。"""
    return os.path.dirname(APP_DIR)


def normalize_request_path(path):
    """归一路径：去 query/fragment、URL 解码、折叠重复斜杠、去尾斜杠。

    不做大小写折叠（文件系统可能大小写敏感），小写化只用于白名单匹配。
    """
    try:
        from urllib.parse import unquote
    except Exception:                                          # pragma: no cover
        def unquote(x):
            return x
    p = str(path or "")
    for sep in ("?", "#"):
        idx = p.find(sep)
        if idx >= 0:
            p = p[:idx]
    try:
        p = unquote(p)
    except Exception:                                          # noqa: BLE001
        pass
    p = p.replace("\\", "/")
    while "//" in p:
        p = p.replace("//", "/")
    if len(p) > 1 and p.endswith("/"):
        p = p.rstrip("/")
    if not p.startswith("/"):
        p = "/" + p
    return p


def _candidate_paths(url_path):
    """把 URL 映射为候选文件路径（应用 wwwroot + 外部扫描根，任一存在即视为存在）。"""
    p = normalize_request_path(url_path)
    cands = []
    for root, prefix in _scan_roots():
        if prefix:
            if not (p.lower() == prefix.lower() or p.lower().startswith(prefix.lower() + "/")):
                continue
            rel = p[len(prefix):].lstrip("/")
        else:
            rel = p.lstrip("/")
        if not rel:
            continue
        cands.append(os.path.join(root, rel))
        cands.append(os.path.join(root, rel, "index.dspy"))
        cands.append(os.path.join(root, rel, "index.ui"))
    return cands


def resource_exists(url_path):
    """资源在 wwwroot 下是否存在（软链已跟随即存在）。"""
    for c in _candidate_paths(url_path):
        try:
            if os.path.isfile(c):
                return True
        except OSError:
            continue
    return False


def is_public_endpoint(url_path):
    """是否免鉴权（委托单一真源 auth_exempt.is_exempt_path，精确比对无通配）。"""
    try:
        return bool(EXEMPT.is_exempt_path(url_path))
    except Exception:                                          # noqa: BLE001
        # 豁免判定自身故障 → fail-closed（不放行），但记日志，绝不静默。
        log.error("[auth_route_gate] 豁免判定异常，按不豁免处理: %r", url_path)
        return False


def is_protected(url_path):
    """该路径是否受本门禁保护（= 后缀在 PROTECTED_SUFFIXES 内且非公开）。"""
    p = normalize_request_path(url_path)
    if not p.lower().endswith(PROTECTED_SUFFIXES):
        return False
    return not is_public_endpoint(p)


# --------------------------------------------------------------------------- #
# 会话校验（复用 route_guard_middleware 的会话解密，不重复实现密码学）
# --------------------------------------------------------------------------- #

_ROUTE_GUARD = None


def _route_guard_module():
    """按文件路径加载 route_guard_middleware（不依赖 app 包可导入）。"""
    global _ROUTE_GUARD
    if _ROUTE_GUARD is not None:
        return _ROUTE_GUARD
    try:
        from app import route_guard_middleware as rg            # noqa: F401
        _ROUTE_GUARD = rg
        return rg
    except Exception:                                          # noqa: BLE001
        pass
    path = os.path.join(APP_DIR, "route_guard_middleware.py")
    try:
        spec = importlib.util.spec_from_file_location("_yj_route_guard", path)
        mod = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(mod)
        _ROUTE_GUARD = mod
        return mod
    except Exception:                                          # noqa: BLE001
        _ROUTE_GUARD = False
        return False


async def _verify_session_impl(request):
    """返回 (ok: bool, reason: str)。

    优先委托 route_guard_middleware.verify_session（真实 cookie 解密 + rbac 会话）。
    取不到实现时 **fail-closed**（返回 False）——宁可 401 也不放行匿名。
    """
    rg = _route_guard_module()
    if rg and hasattr(rg, "verify_session"):
        try:
            res = await rg.verify_session(request)
        except Exception as exc:                               # noqa: BLE001
            return False, "session_verify_error:%s" % type(exc).__name__
        if isinstance(res, tuple):
            return bool(res[0]), str(res[1] if len(res) > 1 else "")
        return bool(res), ""
    return False, "no_session_verifier"


#: 测试替身挂载点（selftest 覆盖此名字）
_verify_session = _verify_session_impl


async def verify_session(request):
    return await _verify_session(request)


# --------------------------------------------------------------------------- #
# 纯函数判定（可脱框架单测）
# --------------------------------------------------------------------------- #

def gate_decision(url_path, exists=None, session_ok=None):
    """返回 (http_code, reason)。http_code 为 None = 放行。

    exists / session_ok 传入 None 时按真实实现推算（exists 用文件系统，
    session_ok 无法在纯函数里 await，必须显式传入）。
    """
    p = normalize_request_path(url_path)
    if not p.lower().endswith(PROTECTED_SUFFIXES):
        return None, "not_dspy"
    # 豁免先于存在性（与真实中间件同口径，T4）
    if is_public_endpoint(p):
        return None, "public_endpoint"
    if exists is None:
        exists = resource_exists(p)
    if not exists:
        return 404, "no such resource"
    if not session_ok:
        return 401, "authentication required"
    return None, "authorized"


# --------------------------------------------------------------------------- #
# aiohttp 中间件
# --------------------------------------------------------------------------- #

def _json_response(status, payload, headers=None):
    """构造 JSON 响应；aiohttp 不可用时退化为轻量对象（供离线自测）。"""
    body = json.dumps(payload, ensure_ascii=False)
    try:
        from aiohttp import web
        return web.Response(status=status, text=body, content_type="application/json",
                            headers=headers or {})
    except Exception:                                          # noqa: BLE001
        class _LiteResponse(object):
            def __init__(self, status, text, hdrs):
                self.status = status
                self.text = text
                self.headers = dict(hdrs or {})
                self.content_type = "application/json"

            def __repr__(self):
                return "<LiteResponse %s %s>" % (self.status, self.text)
        return _LiteResponse(status, body, headers)


def _gate_enabled():
    return (os.environ.get(GATE_ENV) or "on").strip().lower() not in ("off", "0", "false", "no")


async def auth_route_gate_middleware(request, handler):
    """存在性 → 鉴权（顺序不可颠倒；404 不得被 401 覆盖）。"""
    if not _gate_enabled():
        return await handler(request)

    path = normalize_request_path(getattr(request, "path", "") or "")
    if not path.lower().endswith(PROTECTED_SUFFIXES):
        return await handler(request)

    # ① 豁免（**必须先于存在性与会话校验**，T4 / Bug uifdBIMdmixZcmJEdV33O）：
    #    登录/登出/注册/会话自检端点在「还没有会话」的时刻被调用，任何形式的
    #    404/401 都会造成「先登录才能登录」的自指死循环。豁免判定放在最前，
    #    意味着 wwwroot 软链缺失（存在性为假）、会话组件不可用（校验抛错）
    #    都不会把登录端点变成本门禁的错误信封；真不存在时由框架自己给 404。
    if is_public_endpoint(path):
        return await handler(request)

    # ② 存在性：不存在 → 404（先于鉴权）
    if not resource_exists(path):
        return _json_response(404, {
            "status": 404,
            "success": False,
            "ok": False,
            "reason": "no such resource",
            "path": path,
        })

    # ③ 鉴权：无有效会话 → 401
    try:
        ok, reason = await verify_session(request)
    except Exception as exc:                                   # noqa: BLE001
        ok, reason = False, "session_verify_error:%s" % type(exc).__name__
    if not ok:
        return _json_response(401, {
            "status": 401,
            "success": False,
            "ok": False,
            "reason": "authentication required",
            "path": path,
            "detail": reason or "",
        }, headers={"WWW-Authenticate": 'Session realm="yuanjing"'})

    # ④ 有身份 → 放行（权限不足的 403 由框架 RBAC 判定）
    return await handler(request)


# --------------------------------------------------------------------------- #
# 装配
# --------------------------------------------------------------------------- #

def _middleware_names(app):
    return [getattr(m, "__name__", repr(m)) for m in (getattr(app, "middlewares", None) or ())]


def install(app):
    """把门禁插到 app.middlewares 首位（最外层）。幂等，永不抛出。"""
    out = {"layer": "auth_route_gate", "ok": False, "added": [], "error": None,
           "position": None}
    try:
        names = _middleware_names(app)
        if MIDDLEWARE_NAME in names:
            out["ok"] = True
            out["position"] = names.index(MIDDLEWARE_NAME)
            out["added"] = []
        else:
            # [R52 chain-contract] 必须插 new-style（带 __middleware_version__）
            # 中间件：裸 async def 进链会被 aiohttp 按 legacy 工厂 mw(app) 调用，
            # request 形参收到的是 AHApp → `request.remote` AttributeError（R48 E1）。
            app.middlewares.insert(0, _new_style(auth_route_gate_middleware))
            names = _middleware_names(app)
            out["added"] = [MIDDLEWARE_NAME]
            out["position"] = names.index(MIDDLEWARE_NAME) if MIDDLEWARE_NAME in names else None
            out["ok"] = out["position"] == 0
        out["middlewares"] = names
        if not out["ok"] and not out["error"]:
            out["error"] = "not at outermost position: names=%s" % (names,)
    except Exception as exc:                                   # noqa: BLE001
        out["error"] = "%s: %s" % (type(exc).__name__, exc)
    return out


# --------------------------------------------------------------------------- #
# 覆盖矩阵（全模块同类排查的机械口径）
# --------------------------------------------------------------------------- #

def _scan_roots():
    """返回 [(物理根目录, URL 前缀)]：应用 wwwroot + 可选的外部模块仓库根。

    外部根由环境变量 YUANJING_AUTH_GATE_SCAN_ROOTS 注入（逗号分隔的目录，
    每个目录下按 <模块>/wwwroot 扫描），**脚本内不写任何绝对路径**——
    部署机与开发工作区的目录层级不同，写死会误报「漏管」。
    """
    roots = [(os.path.join(_app_root(), "wwwroot"), "")]
    extra = (os.environ.get("YUANJING_AUTH_GATE_SCAN_ROOTS") or "").strip()
    for one in extra.split(","):
        one = one.strip()
        if not one or not os.path.isdir(one):
            continue
        try:
            names = sorted(os.listdir(one))
        except OSError:
            continue
        for name in names:
            ww = os.path.join(one, name, "wwwroot")
            if os.path.isdir(ww):
                roots.append((ww, "/" + name))
    return roots


def iter_wwwroot_dspy():
    """遍历 wwwroot 下全部 .dspy（含经软链挂载的各模块），产出 URL 路径。"""
    found = []
    for root, prefix in _scan_roots():
        for dirpath, _dirnames, filenames in os.walk(root, followlinks=True):
            for fn in filenames:
                if not fn.endswith(".dspy"):
                    continue
                full = os.path.join(dirpath, fn)
                rel = os.path.relpath(full, root).replace(os.sep, "/")
                found.append((prefix + "/" + rel) if prefix else ("/" + rel))
    return sorted(set(found))


def coverage_matrix(paths=None):
    """返回 [{path, decision, public}]，用于「纳管/漏管」机械核验。"""
    rows = []
    for p in (paths if paths is not None else iter_wwwroot_dspy()):
        public = is_public_endpoint(p)
        rows.append({
            "path": p,
            "public": public,
            "decision": "public" if public else "protected",
        })
    return rows


def assert_no_wildcard(items):
    """白名单禁通配符（QC 硬门禁 #6）。"""
    bad = [p for p in items if "*" in str(p) or "%" in str(p)]
    if bad:
        raise AssertionError("wildcard not allowed in allowlist: %r" % (bad,))
    return True


assert_no_wildcard(tuple(PUBLIC_DSPY_PATHS) + tuple(PUBLIC_PATH_PREFIXES))


# ---------------------------------------------------------------- R52 签名契约
# 根因（R48 E1）：本文件的 `auth_route_gate_middleware` 是**裸** `async def
# (request, handler)`，没有 `@web.middleware`。aiohttp 用 `__middleware_version__`
# 区分中间件风格，缺该标记的可调用对象会被当作 **legacy 工厂** 调用 `mw(app)` ——
# 应用对象被填进第一个形参（request），于是 `pkgs/ahserver/ahserver/real_ip.py:16`
# 的 `request['client_ip'] = request.remote` 抛
# `AttributeError: 'AHApp' object has no attribute 'remote'`，全站 500。
# 这里在**模块级**把它规范成 new-style，并在 install() 插入前再规范化一次（幂等）。
_MCC = None


def _load_contract_module():
    """加载同目录 middleware_chain_contract.py（不依赖 sys.path / 包上下文）。

    R48 E1 教训：`except ImportError: return mw` 的静默降级会让裸函数原样进链，
    aiohttp 按 legacy 工厂 `mw(app)` 调用 → `real_ip.py:16 request.remote`
    AttributeError，全站 500。这里改为 spec_from_file_location 按 `__file__`
    同级路径显式加载（同仓 app/scense.py 的既有写法），仍失败返回 None 并由
    调用方显式补标记 + 记 error（绝不静默放行）。
    """
    global _MCC
    if _MCC is not None:
        return _MCC
    try:
        import middleware_chain_contract as _m
        _MCC = _m
        return _MCC
    except ImportError:
        pass
    try:
        from app import middleware_chain_contract as _m
        _MCC = _m
        return _MCC
    except Exception:                                          # noqa: BLE001
        pass
    _path = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                         "middleware_chain_contract.py")
    try:
        _spec = importlib.util.spec_from_file_location(
            "middleware_chain_contract", _path)
        if _spec is None or _spec.loader is None:
            raise ImportError("spec_from_file_location 失败: %s" % _path)
        _mod = importlib.util.module_from_spec(_spec)
        _spec.loader.exec_module(_mod)
        sys.modules.setdefault("middleware_chain_contract", _mod)
        _MCC = _mod
        return _MCC
    except Exception as _exc:                                  # noqa: BLE001
        log.error("[chain-contract] 无法加载 %s: %s", _path, _exc)
        return None


def _new_style(mw):
    """把中间件规范成 new-style（带 __middleware_version__）。

    **禁止静默放行裸函数**（R48 E1 根因）：规范化不可用也要显式补标记并记
    error 日志；规范化后立即自校验，不成立直接抛错而不是继续装配。
    """
    _mcc = _load_contract_module()
    fixed = mw
    if _mcc is not None:
        try:
            fixed = _mcc.ensure_middleware(mw)
        except Exception as _exc:                              # noqa: BLE001
            log.error("[chain-contract] %s 规范化失败（%s），改用手工补 "
                      "new-style 标记", getattr(mw, "__name__", mw), _exc)
            fixed = mw
    if getattr(fixed, "__middleware_version__", None) != 1:
        try:
            setattr(fixed, "__middleware_version__", 1)        # aiohttp 认这个标记
            log.error("[chain-contract] %s 未经契约模块规范化，已手工补 "
                      "__middleware_version__=1（请检查中间件链装配）",
                      getattr(mw, "__name__", mw))
        except Exception as _exc:                              # noqa: BLE001
            raise RuntimeError(
                "中间件 %s 既无法规范化也无法补标记: %s"
                % (getattr(mw, "__name__", mw), _exc))
    if getattr(fixed, "__middleware_version__", None) != 1:     # 自校验，失败即抛
        raise RuntimeError("中间件 %s 规范化后仍缺 __middleware_version__"
                           % (getattr(mw, "__name__", mw),))
    return fixed


auth_route_gate_middleware = _new_style(auth_route_gate_middleware)