根因:元景迭代中为通过 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 中间件提供。
145 lines
5.2 KiB
Python
145 lines
5.2 KiB
Python
# -*- 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
|