ymq 702f27407b refactor(auth): 移除全部自造鉴权/会话/语义中间件,回归标准 ahserver+rbac
根因:元景迭代中为通过 QC 门禁反复叠加了 24 个自定义中间件
(auth_gate_strict / session_gate / login_param_source / route_guard /
 entry_semantics / http_semantics / notfound / get_only_405 / ...),
导致标准登录链路被破坏:
  - auth_gate_strict 手写 Fernet 解密 + 用 aiohttp_session 的会话创建时间
    created 按 1 天判过期 → 所有受保护端点带 cookie 仍 401
  - route_guard 的 _has_business_params 用 await request.read() 消费 POST body
    → up_login 收到 params_kw={} → password_encode(None) 崩 500

改动:
  - app/scense.py 回归 57 行标准入口:ServerEnv + load_appbase/load_rbac/
    load_pybricks + 业务模块 load_*,零自定义中间件
  - 24 个自造中间件归档到 app/_archive_disabled_20261028_151505/(可回滚)
  - /healthz 改用 ahserver 标准 FunctionProcessor(RegisterFunction 注册 +
    conf.website.startswiths 映射),不再用中间件拦截

实测闭环(127.0.0.1:9380 + https://gbak.opencomputing.cn):
  healthz 200 | 匿名 401 | admin 登录 200+cookie | 带cookie 200 |
  伪造cookie 401 | 注册 200+自动登录 | 新用户登录 200+cookie |
  新用户访问 403(已认证无权限,RBAC 正确行为)| 登出 200 |
  错误密码 200 业务信封(不崩 500)

register/login/logout/session 全部由 rbac 模块 up_login.dspy /
register.dspy / logout.dspy + ahserver session 中间件提供。
2026-10-28 15:19:43 +08:00

145 lines
5.2 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# -*- coding: utf-8 -*-
"""静态扩展兜底守卫:确保 `.html` 在 website.static 中。
## 缺陷(Bug qgn-8gwY1UEqGhjlQU_-c / 用例 C15)
`conf/config.json.example` 的 `website.processors` 只登记 `.tmpl/.ui/.dspy`,
`website.static` 扩展列表里**没有 `.html`**。于是 GET `/runtime/play.html`:
- 不命中任何 processor(无 `.html` 规则)
- 不命中 static 白名单(无 `.html`)
⇒ ahserver 走"未知资源类型"异常路径 → **HTTP 500**。
期望语义:静态页面入口在鉴权闸门之后返回 401(未登录)或 302(跳登录),
绝不返回 5xx。
## 修复策略(三层,幂等)
1. `conf/config.json.example` 的 static 列表补 `.html`(长期正确的模板);
2. `scripts/ensure_static_html.py` 幂等修补部署态运行 config(build/部署期);
3. **本模块**:进程启动期在内存配置里兜底补齐 `.html`——
即使部署机上的运行 config 未更新(历史包),启动后 `.html` 也按静态资源处理,
500 立即消失。这是本次修复"部署即生效、不依赖重新投递 config"的关键层。
## 兼容形态
appPublic 的 `getConfig()` 在不同版本下可能返回 `dict` 或属性式
`DictObject`。本模块对两种形态都做处理,且**任何异常都不得影响应用启动**
(try/except 包裹,失败只记 WARN)。
"""
STATIC_EXT = ".html"
def _get_container(cfg):
"""取 website 容器,兼容 dict / DictObject / 属性式对象。"""
if cfg is None:
return None
# 属性式
web = getattr(cfg, "website", None)
if web is None:
web = getattr(cfg, "WEB_SITE", None)
# 字典式
if web is None and hasattr(cfg, "get"):
try:
web = cfg.get("website")
except Exception:
web = None
if web is None and hasattr(cfg, "get"):
try:
web = cfg.get("WEB_SITE")
except Exception:
web = None
return web
def _get_static_list(web):
if web is None:
return None
if isinstance(web, dict):
return web.get("static")
lst = getattr(web, "static", None)
if lst is not None:
return lst
if hasattr(web, "get"):
try:
return web.get("static")
except Exception:
return None
return None
def _set_static_list(web, lst):
"""就地替换 static 列表(保持容器类型不变)。返回是否写入成功。"""
if web is None:
return False
if isinstance(web, dict):
web["static"] = lst
return True
try:
setattr(web, "static", lst)
return True
except Exception:
pass
if hasattr(web, "__setitem__"):
try:
web["static"] = lst
return True
except Exception:
return False
return False
def ensure_static_html(cfg, logger=None):
"""确保 cfg.website.static 含 `.html`。
Returns:
dict: {"changed": bool, "reason": str, "static": list|None}
语义:幂等。已含 `.html` → changed=False;容器不可达 → changed=False
并给出 reason(调用方据此决定是否抬高日志级别,但绝不因此中断启动)。
"""
try:
web = _get_container(cfg)
if web is None:
return {"changed": False, "reason": "website config not reachable",
"static": None}
lst = _get_static_list(web)
if lst is None:
return {"changed": False, "reason": "website.static not present",
"static": None}
# 统一成 list 副本(可能是 tuple)
try:
exts = [str(x) for x in lst]
except Exception:
return {"changed": False, "reason": "website.static not iterable",
"static": None}
norm = [e.lower() for e in exts]
if STATIC_EXT in norm or STATIC_EXT.lstrip(".") in [
n.lstrip(".") for n in norm]:
return {"changed": False, "reason": "already present", "static": exts}
new = [STATIC_EXT] + exts
if not _set_static_list(web, new):
return {"changed": False, "reason": "website.static not writable",
"static": exts}
if logger is not None:
try:
logger.warning(
"[static_ext_guard] website.static 缺少 %s,已在内存配置补齐"
"(未登录 GET /*.html 由 500 收敛为 401/302)。"
"请同步部署 conf/config.json(scripts/ensure_static_html.py)。",
STATIC_EXT)
except Exception:
pass
return {"changed": True, "reason": "appended", "static": new}
except Exception as e: # 兜底:绝不因本守卫影响启动
return {"changed": False, "reason": "guard error: %s" % (e,),
"static": None}
def assert_static_html(cfg):
"""断言式版本(供 selftest 使用):返回 True/False,不抛异常。"""
web = _get_container(cfg)
lst = _get_static_list(web)
if lst is None:
return False
try:
return STATIC_EXT in [str(x).lower() for x in lst]
except Exception:
return False