scense/app/_archive_disabled_20261028_151505/session_gate_middleware.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""`session_gate_middleware` —— 宿主应用**统一会话鉴权入口**（拦截本体）。

职责边界（单一真源，不复制判定口径）
------------------------------------
判定逻辑全部来自 `modules/pbl_common/pbl_common/auth.py`，按以下优先级取契约：

  1. `ServerEnv()` 上由 `pbl_common/init.py::load_pbl_common()` 挂载的名字
     （`env.pbl_common_auth_verify_session` / `_is_public` / `_requires_session`
      / `_unauthorized_payload` / `_forbidden_payload` / `_public_paths`
      / `_required_prefixes`）；
  2. 直接 `import pbl_common.auth`；
  3. 按**绝对文件路径**加载 `auth.py`（部署态 sys.path 常不含模块根，R62 实测）。

豁免（免会话）清单的单一真源是 `app/auth_exempt.py`；本文件只做**接线 + 短路**，
不自持第二份清单（历史多清单漂移缺陷见 auth_gate_strict.py 顶部 R57b/R59 注）。

短路语义（任务范围 1/2）
------------------------
1. 不要求会话的路径（显式公开项：`/healthz`、`/rbac/user/up_login.dspy` 等，
   以及 `.ui/.js/.css` 静态资产）→ `return await handler(request)`，框架行为不变。
2. 要求会话的路径（所有业务 `.dspy`，以及命中 `PROTECTED_API_PREFIXES` 的业务 API）
   → **在调用 handler 之前** `await verify_session(...)`：
     * 会话不存在（`no_session`）/ 无登录票据（`no_auth_ticket`）→ **401**；
     * cookie 伪造 / 解密失败 / 过期（`forged_session` / `expired_session`）→ **401**；
     * 会话与声明身份不匹配（`identity_mismatch`）→ **401**；
     * 会话有效但 RBAC 角色不足 → **403** `permission_denied`；
     * 闸门契约不可用（三处都取不到）→ **fail-closed 401** `auth_unavailable`。
   以上分支**全部在业务 handler 之前 return**，响应体只有
   `{ok:false, error, status, path[, hint]}` 四~五键，**零业务字段**。
3. 校验通过 → `request['pbl_session'] = identity`，继续下沉业务 handler。

与既有层的关系
--------------
`entry_guard_r19` / `route_guard_middleware` / `auth_gate_strict` 与本层职责重叠是
**有意的纵深防御**：本层是「会话完整性 + RBAC 角色」的统一入口，被 `app/scense.py`
的 `_load_semantic_middlewares()` 登记在 `auth_gate_strict` **之前**（更外层），
使统一入口在任何启动方式下都在链上；内层保险丝保留，互不遮蔽。
"""

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

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

try:                                          # 部署态必有；离线自测缺省时走 fallback
    from aiohttp import web as _web
except Exception:                             # noqa: BLE001
    _web = None

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

#: 闸门总开关（仅排障；生产禁开），与 route_guard / auth_gate_strict 同名。
GATE_ENV = "YJ_AUTH_GATE"

#: 契约模块绝对路径兜底候选（部署态 = pkgs/，开发态 = 机构 modules/）。
#: 注意目录层级：APP_ROOT = {space_root}/apps/{应用}，机构 modules/ 在**两级**之上。
_AUTH_PY_CANDIDATES = (
    os.environ.get("PBL_COMMON_AUTH_PY", ""),
    os.path.join(APP_ROOT, "pkgs", "pbl_common", "pbl_common", "auth.py"),
    os.path.join(APP_ROOT, os.pardir, os.pardir, "modules",
                 "pbl_common", "pbl_common", "auth.py"),
    os.path.join(APP_ROOT, os.pardir, "modules", "pbl_common", "pbl_common", "auth.py"),
    os.path.join(APP_ROOT, os.pardir, "pkgs", "pbl_common", "pbl_common", "auth.py"),
)

_LOADPATH_ROOTS = (
    os.environ.get("YJ_RBAC_LOADPATH_ROOT", ""),
    os.path.join(APP_ROOT, "pkgs"),
    os.path.join(APP_ROOT, os.pardir, os.pardir, "modules"),
    os.path.join(APP_ROOT, os.pardir, "modules"),
    os.path.join(APP_ROOT, os.pardir, "pkgs"),
)

#: 必须走统一校验的业务前缀（需求 4：/scense /runtime /scene /world /entity）。
DEFAULT_REQUIRED_PREFIXES = ('/scense', '/runtime', '/scene', '/world', '/entity')

_CONTRACT = None      # 解析后的契约对象（模块或 env 适配器）
_CONTRACT_NAME = "unresolved"
_RBAC_ROLES = None    # path -> 所需角色（来自各模块 scripts/load_path.py 的 PATHS）
_RBAC_META = {}


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


# ── 契约解析（env → import → 绝对路径）───────────────────────────────────────
class _EnvContract(object):
    """ServerEnv 挂载形态的适配器（env.pbl_common_auth_*）。"""

    def __init__(self, env):
        self._env = env

    def __getattr__(self, name):
        attr = "pbl_common_auth_" + name
        val = getattr(self._env, attr, None)
        if val is None:
            raise AttributeError(attr)
        return val

    def raw(self, name):
        return getattr(self._env, "pbl_common_auth_" + name, None)


def _load_auth_by_abspath():
    for cand in _AUTH_PY_CANDIDATES:
        if not cand or not os.path.isfile(cand):
            continue
        try:
            spec = importlib.util.spec_from_file_location("pbl_common_auth_abspath", cand)
            mod = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(mod)
            return mod, cand
        except Exception as exc:              # noqa: BLE001
            log.error("[session_gate] 绝对路径加载 auth.py 失败 %s: %s", cand, exc)
    return None, ""


def contract():
    """统一鉴权契约（惰性解析一次，解析结果带来源名，供诊断输出）。"""
    global _CONTRACT, _CONTRACT_NAME
    if _CONTRACT is not None:
        return _CONTRACT
    # ① ServerEnv 挂载（load_pbl_common 已执行时）
    for name in ("ahserver.serverenv",):
        try:
            m = importlib.import_module(name)
            env = m.ServerEnv()
            if getattr(env, "pbl_common_auth_verify_session", None) is not None:
                _CONTRACT, _CONTRACT_NAME = _EnvContract(env), "ServerEnv(env)"
                log.info("[session_gate] 鉴权契约来源=ServerEnv")
                return _CONTRACT
        except Exception:                     # noqa: BLE001
            pass
    # ② 正常包导入（失败**不返回**，继续走 ③：pbl_common 的包 __init__ 会拉 ahserver，
    #    离线/精简部署态下必然 ImportError，此时绝对路径形态仍可拿到纯 stdlib 的 auth.py）
    try:
        mod = importlib.import_module("pbl_common.auth")
        _CONTRACT, _CONTRACT_NAME = mod, "pbl_common.auth"
        log.info("[session_gate] 鉴权契约来源=pbl_common.auth")
        return _CONTRACT
    except Exception as exc:                  # noqa: BLE001
        log.warning("[session_gate] import pbl_common.auth 不可用，继续绝对路径兜底: %s", exc)
    # ③ 绝对路径兜底
    mod, path = _load_auth_by_abspath()
    if mod is not None:
        _CONTRACT, _CONTRACT_NAME = mod, path
        log.info("[session_gate] 鉴权契约来源=abspath %s", path)
        return _CONTRACT
    _CONTRACT, _CONTRACT_NAME = None, "unavailable"
    log.error("[session_gate] 鉴权契约三处解析全部失败 —— 闸门将 fail-closed 返回 401 "
              "auth_unavailable（绝不放行业务数据）")
    return None


def contract_name():
    contract()
    return _CONTRACT_NAME


# ── RBAC 角色登记表（load_path.py 单一真源，禁通配）───────────────────────────
def _loadpath_files():
    out = {}
    for prefix in required_prefixes():
        mod = str(prefix).strip("/")
        for root in _LOADPATH_ROOTS:
            if not root:
                continue
            cand = os.path.join(root, mod, "scripts", "load_path.py")
            if os.path.isfile(cand):
                out[prefix] = cand
                break
    return out


def rbac_registry():
    """path -> 所需角色；同时做**通配符硬校验**（命中即整表不可用，fail-closed）。"""
    global _RBAC_ROLES, _RBAC_META
    if _RBAC_ROLES is not None:
        return _RBAC_ROLES
    table = {}
    meta = {"sources": {}, "entries": 0, "wildcard_free": True, "wildcards": []}
    for prefix, file_path in sorted(_loadpath_files().items()):
        try:
            spec = importlib.util.spec_from_file_location(
                "yj_load_path_" + prefix.strip("/").replace("-", "_"), file_path)
            mod = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(mod)
        except Exception as exc:              # noqa: BLE001
            log.error("[session_gate] RBAC 登记表加载失败 %s: %s", file_path, exc)
            continue
        pairs = getattr(mod, "PATHS", None) or []
        bad = [p for p, _r in pairs if ("*" in p or "%" in p or "{" in p)]
        if bad:
            meta["wildcard_free"] = False
            meta["wildcards"].extend(bad)
            log.error("[session_gate] RBAC 登记表含通配/占位项（硬门禁拒绝）: %s", bad)
            continue
        for p, role in pairs:
            table[str(p)] = str(role)
        meta["sources"][prefix] = len(pairs)
        meta["entries"] += len(pairs)
    _RBAC_ROLES, _RBAC_META = table, meta
    log.info("[session_gate] RBAC 登记表载入 entries=%d wildcard_free=%s sources=%s",
             meta["entries"], meta["wildcard_free"], meta["sources"])
    return table


def rbac_meta():
    rbac_registry()
    return _RBAC_META


def required_role(path):
    """该路径在 RBAC 登记表中声明的角色（未登记返回 ''）。"""
    table = rbac_registry()
    if not table:
        return ''
    p = _norm(path)
    if p in table:
        return table[p]
    if p + ".dspy" in table:
        return table[p + ".dspy"]
    # 最长前缀兜底（模块根 '/entity' 这类登记项覆盖其下未单独登记的路径）
    best, best_len = '', 0
    for k, v in table.items():
        if k != "/" and (p == k or p.startswith(k.rstrip("/") + "/")) and len(k) > best_len:
            best, best_len = v, len(k)
    return best


# ── 公开/必需判定（一律委托契约，本文件不持有清单）───────────────────────────
def _norm(path):
    p = str(path or "")
    p = p.split("?", 1)[0].split("#", 1)[0]
    p = "/" + "/".join([s for s in p.split("/") if s])
    return p if p == "/" else p.rstrip("/")


def required_prefixes():
    c = contract()
    if c is not None:
        try:
            val = c.raw("required_prefixes") if isinstance(c, _EnvContract) else \
                getattr(c, "AUTH_REQUIRED_PREFIXES", None)
            if val:
                return tuple(val)
        except Exception:                     # noqa: BLE001
            pass
    return DEFAULT_REQUIRED_PREFIXES


def is_public(path):
    """显式公开项（白名单精确命中，禁通配 / 禁子路径放行）。"""
    c = contract()
    p = _norm(path)
    if c is not None:
        try:
            fn = c.raw("is_public") if isinstance(c, _EnvContract) else \
                getattr(c, "auth_is_public", None)
            if fn is not None and bool(fn(p)):
                return True
        except Exception as exc:              # noqa: BLE001
            log.error("[session_gate] 公开项判定异常（按不公开处理）: %s", exc)
    return False


def requires_session(path):
    """是否必须过会话校验。

    覆盖面 = 契约判定（所有非公开 `.dspy`） **∪** 业务 API 前缀（`/scense /runtime
    /scene /world /entity` 下的 `/api/…`，含无 `.dspy` 后缀的重写变体）。
    第二条是关键：只按后缀判定会让 `/runtime/api/build_entity_tree` 这类无后缀
    业务 API 漏网 → 匿名 200（R62 同源缺陷）。判定异常一律按「需要会话」处理。
    """
    p = _norm(path)
    if is_public(p):
        return False
    c = contract()
    if c is not None:
        try:
            fn = c.raw("requires_session") if isinstance(c, _EnvContract) else \
                getattr(c, "auth_requires_session", None)
            if fn is not None and bool(fn(p)):
                return True
        except Exception as exc:              # noqa: BLE001
            log.error("[session_gate] requires_session 异常（按需会话处理）: %s", exc)
            return True
    if p.lower().endswith(".dspy"):
        return True
    low = p.lower()
    for prefix in required_prefixes():
        head = prefix.rstrip("/") + "/"
        if low.startswith(head.lower()):
            rest = low[len(head) - 1:]
            if "/api/" in rest or rest.endswith(".dspy"):
                return True
    return False


# ── 响应与身份 ────────────────────────────────────────────────────────────────
def _payload(status, reason, path, extra_hint=None):
    """统一错误结构（**不含任何业务字段**）。"""
    c = contract()
    body = None
    try:
        if c is not None:
            fn = c.raw("unauthorized_payload") if status == 401 else \
                c.raw("forbidden_payload")
            if not isinstance(c, _EnvContract) or fn is None:
                fn = getattr(c, "unauthorized_payload" if status == 401
                             else "forbidden_payload", None)
            if fn is not None:
                body = dict(fn(reason, path) or {})
    except Exception as exc:                  # noqa: BLE001
        log.error("[session_gate] 契约错误信封不可用（退回本地结构）: %s", exc)
    if not isinstance(body, dict) or not body:
        body = {"ok": False, "error": reason, "status": status, "path": path}
    body.setdefault("ok", False)
    body.setdefault("status", status)
    body.setdefault("path", path)
    if extra_hint:
        body["hint"] = extra_hint
    return body


def _respond(status, reason, path, hint=None):
    body = _payload(status, reason, path, hint)
    if _web is not None:
        resp = _web.json_response(body, status=status)
        if status == 401:
            resp.headers["WWW-Authenticate"] = 'Session realm="yuanjing"'
        return resp
    return _FallbackResponse(status, body)


class _FallbackResponse(object):
    """离线自测（无 aiohttp）时的等价响应体：只有 status / body 两属性。"""

    def __init__(self, status, body):
        self.status = int(status)
        self.body = body
        self.text = json.dumps(body, ensure_ascii=False)
        self.headers = {}


def _principal_roles(request, identity):
    """取当前会话的角色集合；取不到返回 None（= 未知，交由框架 RBAC 判定）。"""
    bags = []
    if isinstance(identity, dict):
        bags.append(identity)
    try:
        sess = getattr(request, "session", None)
        if isinstance(sess, dict):
            bags.append(sess)
    except Exception:                         # noqa: BLE001
        pass
    try:
        if isinstance(request, dict):
            bags.append(request)
    except Exception:                         # noqa: BLE001
        pass
    roles = set()
    found = False
    for b in bags:
        if not isinstance(b, dict):
            continue
        for key in ("roles", "pbl_roles", "role", "user_roles"):
            val = b.get(key)
            if val in (None, ""):
                continue
            found = True
            if isinstance(val, (list, tuple, set)):
                roles.update(str(x).strip() for x in val if str(x).strip())
            else:
                roles.update(str(x).strip() for x in str(val).split(",") if str(x).strip())
    return roles if found else None


def _ticket_userid(identity):
    if isinstance(identity, dict):
        return str(identity.get("userid") or identity.get("username") or "")
    return ""


def _claimed_userid(request):
    """调用方声明的身份（用于「会话与用户不匹配」判定）。"""
    try:
        headers = getattr(request, "headers", None) or {}
        for key in ("X-User-Id", "X-Userid", "x-user-id"):
            val = headers.get(key)
            if val:
                return str(val).strip()
    except Exception:                         # noqa: BLE001
        pass
    return ""


# ── 闸门本体 ──────────────────────────────────────────────────────────────────
async def session_gate_middleware(request, handler):
    """统一会话鉴权入口：会话校验 + RBAC 角色校验，全部在业务 handler 之前终止。

    判定顺序（顺序即语义）：
      1. 总开关关闭（排障）→ 放行；
      2. 不要求会话的路径（显式公开项 / 静态资产）→ 放行；
      3. 契约不可用 → **401 auth_unavailable**（fail-closed，绝不放行）；
      4. `await verify_session(request)`：不存在/伪造/过期/票据缺失/身份不符 → 401；
      5. 会话有效但 RBAC 角色不足 → 403 permission_denied；
      6. 通过 → 挂 `request['pbl_session']`，`return await handler(request)`。
    """
    path = _norm(getattr(request, "path", None) or
                 (request.get("path") if isinstance(request, dict) else "") or "")
    try:
        if not gate_enabled():
            return await handler(request)
        if not requires_session(path):
            return await handler(request)

        c = contract()
        if c is None:
            log.error("[session_gate] 契约缺失 → 401 auth_unavailable path=%s", path)
            return _respond(401, "auth_unavailable", path,
                            hint="统一鉴权契约不可达（ServerEnv/pbl_common/auth.py 三处均未取到）")

        vs = c.raw("verify_session") if isinstance(c, _EnvContract) else \
            getattr(c, "auth_verify_session", None)
        port = None
        try:
            cfg_path = os.path.join(APP_ROOT, "conf", "config.json")
            if os.path.isfile(cfg_path):
                with open(cfg_path, encoding="utf-8") as fp:
                    cfg = json.load(fp) or {}
                port = str(((cfg.get("website") or {}).get("port")) or "") or None
        except Exception:                     # noqa: BLE001
            port = None
        ok, status, reason, identity = await vs(request, port=port)
    except Exception as exc:                  # noqa: BLE001
        # 闸门自身/判定异常 → fail-closed 500（不冒充鉴权失败，避免遮蔽真实故障，
        # 与 auth_gate_strict R60 同一口径）。
        log.exception("[session_gate] 闸门判定异常 path=%s: %s", path, exc)
        return _respond(500, "session_gate_internal_error", path, hint=str(exc)[:300])

    if not ok:
        status = int(status or 401)
        log.warning("[session_gate] 拦截 %s reason=%s path=%s",
                    getattr(request, "method", "-"), reason, path)
        # 会话校验未通过一律 401（契约已给 403 时保留 403，二者互不覆盖）
        hint = None
        if status == 401:
            hint = "会话缺失/无效：请 POST /rbac/user/up_login.dspy 取得 AIOHTTP_SESSION"
        return _respond(status if status in (401, 403) else 401, reason or "no_session",
                        path, hint)

    # 会话有效 → RBAC 角色判定（角色不可知时交由框架 RBAC，本层不臆断 403）
    claimed = _claimed_userid(request)
    ticket_user = _ticket_userid(identity)
    if claimed and ticket_user and claimed != ticket_user:
        log.warning("[session_gate] 身份不匹配 claimed=%s ticket=%s path=%s",
                    claimed, ticket_user, path)
        return _respond(401, "identity_mismatch", path,
                        hint="声明用户与会话票据身份不一致")

    need = required_role(path)
    if need and need != "any":
        roles = _principal_roles(request, identity if isinstance(identity, dict) else {})
        if roles is not None:
            allowed = (need == "logined") or (need in roles) or ("admin" in roles) \
                or ("super" in roles) or ("superuser" in roles)
            if not allowed:
                log.warning("[session_gate] RBAC 拒绝 path=%s need=%s roles=%s",
                            path, need, sorted(roles))
                return _respond(403, "permission_denied", path,
                                hint="会话有效但缺少该端点要求的角色: " + need)

    try:
        if isinstance(request, dict):
            request["pbl_session"] = identity or {}
        else:
            request["pbl_session"] = identity or {}
    except Exception:                         # noqa: BLE001
        pass
    return await handler(request)


def diagnostics():
    """供 /healthz 与自测读取的闸门装配状态（可远程核验，不静默）。"""
    meta = rbac_meta()
    return {
        "gate_enabled": gate_enabled(),
        "contract_source": contract_name(),
        "required_prefixes": list(required_prefixes()),
        "rbac_entries": meta.get("entries", 0),
        "rbac_wildcard_free": meta.get("wildcard_free", False),
        "rbac_sources": meta.get("sources", {}),
    }


if _web is not None:
    session_gate_middleware = _web.middleware(session_gate_middleware)



def install_session_gate(app, index=0):
    """把统一会话闸门装到 aiohttp 中间件链**最外层**（业务分支之前）。

    为什么用 insert(0)（QC#4 必做1/验收③）
    ------------------------------------
    ahserver `configuredServer.py:44` 只 `insert(0, real_ip_middleware())`，
    `auth_api.py:162` 是 `append(checkAuth)`；应用自己的语义层（notfound /
    http_semantics）也走 append。本函数在 `ahapp_built` 回调里、这些 append
    **之前**执行 insert(0)，因此闸门成为应用可控制的最外层：未登录请求在进入
    任何 .dspy 业务 handler、也早于框架 RBAC 判定，就已按统一结构短路返回。

    幂等：按名字去重，重复调用只移动到 index 位置，不会叠加第二层闸门。

    返回 dict（供入口写证据 / 日志）：
      installed / name / index / total / middlewares / contract_source / skipped
    """
    res = {"installed": False, "name": "session_gate_middleware", "index": None,
           "total": 0, "middlewares": [], "contract_source": contract_name(),
           "skipped": None}
    mws = getattr(app, "middlewares", None)
    if mws is None:
        res["skipped"] = "app.middlewares 不存在（非 aiohttp Application？）"
        log.error("[session_gate] install 跳过：%s", res["skipped"])
        return res

    mw = session_gate_middleware          # 已按 _web.middleware 包装（若 aiohttp 可用）
    try:
        from middleware_chain_contract import ensure_middleware
        mw = ensure_middleware(mw, "session_gate_middleware")
    except Exception as exc:              # noqa: BLE001
        log.warning("[session_gate] middleware_chain_contract 不可用，沿用原对象: %s", exc)

    try:
        for i, m in enumerate(list(mws)):
            if _mw_name_of(m) == "session_gate_middleware":
                if i != index:
                    mws.pop(i)
                    mws.insert(index, m)
                res.update(installed=True, index=index, total=len(mws),
                           middlewares=[_mw_name_of(x) for x in mws])
                log.info("[session_gate] installed as first middleware (already present) "
                         "index=%d total=%d contract=%s",
                         index, res["total"], res["contract_source"])
                return res
        mws.insert(index, mw)
        res.update(installed=True, index=index, total=len(mws),
                   middlewares=[_mw_name_of(x) for x in mws])
        log.info("[session_gate] installed as first middleware index=%d total=%d "
                 "contract=%s gate_enabled=%s prefixes=%s",
                 index, res["total"], res["contract_source"], gate_enabled(),
                 required_prefixes())
    except Exception as exc:              # noqa: BLE001
        res["skipped"] = "%s: %s" % (type(exc).__name__, exc)
        log.exception("[session_gate] install FAILED（闸门未生效，业务端点将无统一 401）")
    return res


def _mw_name_of(mw):
    """尽力取中间件名（装饰后 __name__ 可能被包装层保留/丢失）。"""
    for attr in ("__name__", "_mw_name", "name"):
        v = getattr(mw, attr, None)
        if isinstance(v, str) and v:
            return v
    inner = getattr(mw, "__wrapped__", None)
    if inner is not None and inner is not mw:
        return _mw_name_of(inner)
    return type(mw).__name__