fix(security): 4端点补跨租户鉴权 — 实体1会话读实体2数据(200)→403 (cma-cross-entity-endpoints-auth-20260914)
背景(项目Bot 补充复验 §四#2):以下 4 个只读端点未做会话账套鉴权,
用 entity=1(酣客) 会话 token + 显式 entity 参数可读到 entity=2(博海) 数据(HTTP 200):
- GET /data-quality/governance-check?entity_id=2 → 200 score=88(对照 entity1=72)
- GET /products/matrix?entity_id=2 → 200
- GET /alerts/risk-matrix?entity=bohai → 200 博海风险矩阵
- GET /action-plans/coso-checklist?entity=bohai → 200 博海 COSO total_score=55
另:risk-matrix / coso-checklist 在非法实体值下静默回落酣客数据,UI 标签与数据不一致。
改动:
1. deps.py 新增统一防护入口 assert_entity_scope / assert_entity_slug_scope /
get_session_entity / resolve_session_slug(文案对齐 get_entity_id:无权访问企业 entity_id=X(当前账套: Y))
2. governance-check / products/matrix:entity_id 须等于会话账套,否则 403;
缺省按会话账套(governance-check 原缺省 0=全部 → 登录态下按账套收敛)
3. risk-matrix / coso-checklist:entity 须等于会话账套,否则 403;
非法值 → 400 显式报错(不再静默回落酣客)
4. P3:POST /auth/switch-entity 加 DEPRECATED 注释(前端已无调用,保留不删)
5. 测试:tests/test_security_multitenant.py 新增 TestCrossEntityEndpointAuth 15 例
(跨实体 403 / 同实体 200 / 缺省按账套 / 非法显式报错 / 参照端点行为不变)
验证:pytest tests/ -q → 951 passed(10 项失败为改动前既有基线,已 stash 对照确认);
真实后端冒烟 19/19 PASS(entity=1 会话 4 端点跨实体全 403)。
This commit is contained in:
@@ -230,3 +230,113 @@ class TestCheckGovernanceSQLite:
|
||||
resp = client.get("/api/cma/data-quality/governance-check", headers=auth_header(token))
|
||||
assert resp.status_code == 200, resp.text[:300]
|
||||
assert resp.json()["total_rules"] == 7
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 4 端点跨租户鉴权补单(cma-cross-entity-endpoints-auth-20260914)
|
||||
# 对齐参照:growth-quality / predict.accuracy(已 403)
|
||||
# ============================================================
|
||||
class TestCrossEntityEndpointAuth:
|
||||
"""会话 entity=1 显式请求 entity=2 → 403;同实体 → 200;非法实体 → 显式报错"""
|
||||
|
||||
def _login(self, client, db):
|
||||
_ensure_entity(db, 2)
|
||||
create_test_user(db)
|
||||
return get_token_for_user(client, "testadmin", "admin123") # 登录账套 entity_id=1
|
||||
|
||||
# ── 1. data-quality/governance-check ──────────────────
|
||||
def test_governance_check_cross_entity_403(self, client, db):
|
||||
token = self._login(client, db)
|
||||
r = client.get("/api/cma/data-quality/governance-check?entity_id=2", headers=auth_header(token))
|
||||
assert r.status_code == 403, f"跨实体应 403,实际 {r.status_code}: {r.text[:200]}"
|
||||
assert "无权访问企业 entity_id=2(当前账套: 1)" in r.text
|
||||
|
||||
def test_governance_check_same_entity_200(self, client, db):
|
||||
token = self._login(client, db)
|
||||
r = client.get("/api/cma/data-quality/governance-check?entity_id=1", headers=auth_header(token))
|
||||
assert r.status_code == 200, r.text[:200]
|
||||
assert r.json()["entity_id"] == 1
|
||||
|
||||
def test_governance_check_no_param_scoped_to_session(self, client, db):
|
||||
"""未传 entity_id:按会话账套(不再默认 0=全部)"""
|
||||
token = self._login(client, db)
|
||||
r = client.get("/api/cma/data-quality/governance-check", headers=auth_header(token))
|
||||
assert r.status_code == 200, r.text[:200]
|
||||
assert r.json()["entity_id"] == 1
|
||||
|
||||
def test_governance_check_illegal_entity_403(self, client, db):
|
||||
token = self._login(client, db)
|
||||
r = client.get("/api/cma/data-quality/governance-check?entity_id=999", headers=auth_header(token))
|
||||
assert r.status_code == 403, r.text[:200]
|
||||
|
||||
# ── 2. products/matrix ────────────────────────────────
|
||||
def test_products_matrix_cross_entity_403(self, client, db):
|
||||
token = self._login(client, db)
|
||||
r = client.get("/api/cma/products/matrix?entity_id=2&months=6", headers=auth_header(token))
|
||||
assert r.status_code == 403, f"跨实体应 403,实际 {r.status_code}: {r.text[:200]}"
|
||||
|
||||
def test_products_matrix_same_entity_200(self, client, db):
|
||||
token = self._login(client, db)
|
||||
r = client.get("/api/cma/products/matrix?entity_id=1", headers=auth_header(token))
|
||||
assert r.status_code == 200, r.text[:200]
|
||||
assert r.json()["entity_id"] == 1
|
||||
|
||||
def test_products_matrix_no_param_scoped_to_session(self, client, db):
|
||||
token = self._login(client, db)
|
||||
r = client.get("/api/cma/products/matrix", headers=auth_header(token))
|
||||
assert r.status_code == 200, r.text[:200]
|
||||
assert r.json()["entity_id"] == 1
|
||||
|
||||
# ── 3. alerts/risk-matrix(字符串实体)────────────────
|
||||
def test_risk_matrix_cross_entity_403(self, client, db):
|
||||
token = self._login(client, db)
|
||||
r = client.get("/api/cma/alerts/risk-matrix?entity=bohai", headers=auth_header(token))
|
||||
assert r.status_code == 403, f"跨实体应 403,实际 {r.status_code}: {r.text[:200]}"
|
||||
assert "无权访问企业 entity_id=2(当前账套: 1)" in r.text
|
||||
|
||||
def test_risk_matrix_same_entity_200(self, client, db):
|
||||
token = self._login(client, db)
|
||||
r = client.get("/api/cma/alerts/risk-matrix?entity=hanke", headers=auth_header(token))
|
||||
assert r.status_code == 200, r.text[:200]
|
||||
assert "酣客" in r.json()["entity_name"]
|
||||
|
||||
def test_risk_matrix_no_param_uses_session_entity(self, client, db):
|
||||
token = self._login(client, db)
|
||||
r = client.get("/api/cma/alerts/risk-matrix", headers=auth_header(token))
|
||||
assert r.status_code == 200, r.text[:200]
|
||||
assert "酣客" in r.json()["entity_name"]
|
||||
|
||||
def test_risk_matrix_illegal_entity_explicit_error(self, client, db):
|
||||
"""非法实体 → 400 显式报错(不再静默回落酣客数据)"""
|
||||
token = self._login(client, db)
|
||||
r = client.get("/api/cma/alerts/risk-matrix?entity=2", headers=auth_header(token))
|
||||
assert r.status_code == 400, f"非法实体应 400,实际 {r.status_code}: {r.text[:200]}"
|
||||
assert "非法实体" in r.text
|
||||
assert "默认返回酣客数据" not in r.text
|
||||
|
||||
# ── 4. action-plans/coso-checklist ────────────────────
|
||||
def test_coso_checklist_cross_entity_403(self, client, db):
|
||||
token = self._login(client, db)
|
||||
r = client.get("/api/cma/action-plans/coso-checklist?entity=bohai", headers=auth_header(token))
|
||||
assert r.status_code == 403, f"跨实体应 403,实际 {r.status_code}: {r.text[:200]}"
|
||||
|
||||
def test_coso_checklist_same_entity_200(self, client, db):
|
||||
token = self._login(client, db)
|
||||
r = client.get("/api/cma/action-plans/coso-checklist?entity=hanke", headers=auth_header(token))
|
||||
assert r.status_code == 200, r.text[:200]
|
||||
assert "酣客" in r.json()["entity_name"]
|
||||
|
||||
def test_coso_checklist_illegal_entity_explicit_error(self, client, db):
|
||||
token = self._login(client, db)
|
||||
r = client.get("/api/cma/action-plans/coso-checklist?entity=999", headers=auth_header(token))
|
||||
assert r.status_code == 400, r.text[:200]
|
||||
assert "非法实体" in r.text
|
||||
|
||||
# ── 对齐参照端点行为不变 ──────────────────────────────
|
||||
def test_reference_403_endpoints_unchanged(self, client, db):
|
||||
"""growth-quality / predict.accuracy 既有 403 行为不变"""
|
||||
token = self._login(client, db)
|
||||
r1 = client.get("/api/cma/growth-quality/periods?entity_id=2", headers=auth_header(token))
|
||||
assert r1.status_code == 403, r1.text[:200]
|
||||
r2 = client.get("/api/cma/predict/accuracy?entity_id=2", headers=auth_header(token))
|
||||
assert r2.status_code == 403, r2.text[:200]
|
||||
|
||||
Reference in New Issue
Block a user