根因:元景迭代中为通过 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 中间件提供。
135 lines
5.0 KiB
Python
135 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""向后兼容别名面:`from app import auth_gate` 的外部调用方不应 ImportError。
|
||
|
||
[fix rvtGi1mlUsTwQrXnybZj6 / T4-A] 本文件此前把豁免表达成
|
||
`is_exempt_path(p) = not is_protected(p)` —— 这是**第二份判定入口**:
|
||
`is_protected()` 是「按资源存在性 + 扩展名」推导的受保护集合,与豁免清单
|
||
(`app/auth_exempt.py`)的判据不同源。两份口径一旦分叉(例如某路径磁盘不可见
|
||
但语义上必须免鉴权),调用方拿到的豁免结论就与主线闸门不一致,
|
||
正是 Bug uifdBIMdmixZcmJEdV33O(登录端点自己 401)的原始形态。
|
||
|
||
现在的约定(唯一真源 = `app/auth_exempt.py`):
|
||
|
||
* `is_exempt_path()` / `is_anonymous_path()` 都**纯转发** `auth_exempt.is_exempt_path()`,
|
||
不再用 `not is_protected(...)` 二次表达豁免;
|
||
* `normalize_request_path()` / `resource_exists()` / `is_protected()` 仍委托
|
||
`app/auth_route_gate.py`(资源面语义,与豁免判据无关,保留原样);
|
||
* 本文件不维护任何名单字面量,也不新增策略。
|
||
|
||
pbls 与 yuanjing 顶层包名同为 app,跨应用自测脚本
|
||
apps/pbls/scripts/selftest_auth_gate_r39.py 引用 normalize_request_path /
|
||
is_anonymous_path / is_exempt_path / resource_exists 四个符号,故保留全部四个。
|
||
"""
|
||
|
||
import logging
|
||
|
||
log = logging.getLogger("scense.auth_gate")
|
||
|
||
|
||
def _route_gate():
|
||
"""取资源面实现模块 app/auth_route_gate.py(保持历史委托关系)。"""
|
||
try:
|
||
from . import auth_route_gate as _m
|
||
return _m
|
||
except Exception: # noqa: BLE001
|
||
try:
|
||
import auth_route_gate as _m
|
||
return _m
|
||
except Exception: # noqa: BLE001
|
||
return None
|
||
|
||
|
||
def _auth_exempt():
|
||
"""取鉴权豁免**单一真源** app/auth_exempt.py(部署态 sys.path 可能不含应用根)。"""
|
||
try:
|
||
from . import auth_exempt as _m
|
||
return _m
|
||
except Exception: # noqa: BLE001
|
||
try:
|
||
from app import auth_exempt as _m # type: ignore
|
||
return _m
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
try:
|
||
import auth_exempt as _m
|
||
return _m
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
try:
|
||
import importlib.util
|
||
import os
|
||
p = os.path.join(os.path.dirname(os.path.abspath(__file__)), "auth_exempt.py")
|
||
spec = importlib.util.spec_from_file_location("yj_auth_exempt_gate", p)
|
||
_m = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(_m)
|
||
return _m
|
||
except Exception as exc: # noqa: BLE001
|
||
log.error("[auth_gate] 豁免清单单一真源加载失败: %s", exc)
|
||
return None
|
||
|
||
|
||
def normalize_request_path(path):
|
||
m = _route_gate()
|
||
if m is None:
|
||
return path
|
||
return m.normalize_request_path(path)
|
||
|
||
|
||
def resource_exists(url_path):
|
||
m = _route_gate()
|
||
if m is None:
|
||
return False
|
||
return bool(m.resource_exists(url_path))
|
||
|
||
|
||
def is_protected(path):
|
||
"""资源面受保护判定(**不是**豁免判据;豁免一律走 is_exempt_path)。"""
|
||
m = _route_gate()
|
||
if m is None:
|
||
return True
|
||
try:
|
||
return bool(m.is_protected(path))
|
||
except Exception as exc: # noqa: BLE001
|
||
log.warning("[auth_gate] is_protected 失败,按受保护处理: %s", exc)
|
||
return True
|
||
|
||
|
||
def is_exempt_path(path):
|
||
"""是否免鉴权——**纯转发**单一真源 `auth_exempt.is_exempt_path()`。
|
||
|
||
禁止在此用 `not is_protected(...)` 表达豁免(那会形成第二份判定入口,
|
||
Bug uifdBIMdmixZcmJEdV33O 的原始形态)。单一真源不可用时 fail-closed(False)。
|
||
"""
|
||
m = _auth_exempt()
|
||
if m is None:
|
||
return False
|
||
try:
|
||
return bool(m.is_exempt_path(path))
|
||
except Exception as exc: # noqa: BLE001
|
||
log.error("[auth_gate] 豁免判定异常(按不豁免处理): %s", exc)
|
||
return False
|
||
|
||
|
||
def is_anonymous_path(path):
|
||
"""历史符号:匿名可到达 == 免鉴权。与 `is_exempt_path()` 同源同义(纯转发)。"""
|
||
return is_exempt_path(path)
|
||
|
||
|
||
def manifest():
|
||
"""豁免清单快照——同样纯转发单一真源,便于调用方做一致性断言。"""
|
||
m = _auth_exempt()
|
||
if m is None:
|
||
return {"revision": "unavailable", "dspy_paths": [], "prefixes": [],
|
||
"wildcard_free": False}
|
||
return m.manifest()
|
||
|
||
|
||
def main(): # pragma: no cover
|
||
import json as _json
|
||
print(_json.dumps(manifest(), ensure_ascii=False, indent=2))
|
||
|
||
|
||
if __name__ == "__main__": # pragma: no cover
|
||
main()
|