sxbh_admin/cma-management
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(security): retire default-credential surface (yanxue-20260914-cma-defaultcred-code-fix)

Browse Source 
- Login.vue: no prefilled default creds; no anonymous entity enumeration on mount

- deploy.sh: readiness polling (0.5s/30s cap) replaces fixed sleep 3; login smoke reads same-source 600 secret file, missing file = explicit abort

- 19 verify/check/gen_report scripts: shared _admin_creds() loader (CMA_ADMIN_ENV -> ops .secrets -> profile copy); no hardcoded creds

- tests: single TEST_USER_PASSWORD fixture constant; docs: pointer to 600 secret file; removed 4 legacy .bak + pycache residue
This commit is contained in:
Hermes CI Fix
2026-09-14 17:39:53 +08:00
parent 8adc783501
commit 40d522cd34
39 changed files with 602 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/conftest.py
+6 -2
View File
@@ -134,11 +134,15 @@ def client(db) -> Generator[TestClient, None, None]:
# ── 测试数据工厂 ──
# 测试夹具口令（仅内存 SQLite 测试库使用；不是生产默认口令——生产凭据只存 600 密钥文件
# .secrets/cma-admin.env）。可用 $CMA_TEST_USER_PASSWORD 覆盖；用例一律从这里导入，禁止写字面量。
TEST_USER_PASSWORD = os.environ.get("CMA_TEST_USER_PASSWORD", "test-fixture-pass")
def create_test_user(db: Session, **kwargs) -> User:
"""创建测试用户"""
defaults = {
"username": "testadmin",
"password_hash": hashlib.sha256("admin123".encode()).hexdigest(),
"password_hash": hashlib.sha256(TEST_USER_PASSWORD.encode()).hexdigest(),
"name": "测试管理员",
"role": "ceo",
}
@@ -150,7 +154,7 @@ def create_test_user(db: Session, **kwargs) -> User:
return user
def get_token_for_user(client: TestClient, username: str = "testadmin", password: str = "admin123") -> str:
def get_token_for_user(client: TestClient, username: str = "testadmin", password: str = TEST_USER_PASSWORD) -> str:
"""获取测试用户的token（账套模式：需entity_id）"""
resp = client.post("/api/cma/auth/login", json={
"username": username,
backend/tests/test_api_auth_surface_20260914.py
+2 -2
View File
@@ -17,7 +17,7 @@ from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.models import Entity, UserEntity
from tests.conftest import create_test_user, auth_header
from tests.conftest import create_test_user, auth_header, TEST_USER_PASSWORD
BOT_KEY = {"X-BOT-KEY": "cma-bot-finance-2026"}
@@ -93,7 +93,7 @@ def _ensure_entity(db: Session, eid: int) -> Entity:
def _login(client: TestClient, entity_id: int = 1) -> str:
resp = client.post("/api/cma/auth/login", json={
"username": "testadmin", "password": "admin123", "entity_id": entity_id,
"username": "testadmin", "password": TEST_USER_PASSWORD, "entity_id": entity_id,
})
assert resp.status_code == 200, f"登录 entity{entity_id} 失败: {resp.text[:300]}"
return resp.json().get("token")
backend/tests/test_auth.py
+4 -4
View File
@@ -4,7 +4,7 @@
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from tests.conftest import create_test_user, get_token_for_user, auth_header
from tests.conftest import create_test_user, get_token_for_user, auth_header, TEST_USER_PASSWORD
class TestAuth:
@@ -16,7 +16,7 @@ class TestAuth:
resp = client.post("/api/cma/auth/login", json={
"username": "testadmin",
"password": "admin123",
"password": TEST_USER_PASSWORD,
"entity_id": 1,
})
@@ -41,7 +41,7 @@ class TestAuth:
"""用户不存在"""
resp = client.post("/api/cma/auth/login", json={
"username": "nobody",
"password": "admin123",
"password": TEST_USER_PASSWORD,
})
assert resp.status_code == 401
@@ -89,7 +89,7 @@ class TestAuth:
resp = client.post("/api/cma/auth/register", json={
"username": "testadmin",
"password": "admin123",
"password": TEST_USER_PASSWORD,
"name": "重复用户",
})
backend/tests/test_dashboard.py
+4 -4
View File
@@ -5,7 +5,7 @@ import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from datetime import datetime
from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi
from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi, TEST_USER_PASSWORD
from app.models import KPIAlert, KPIValue
@@ -74,7 +74,7 @@ class TestDashboard:
def test_my_kpis(self, client: TestClient, db: Session):
user = create_test_user(db, username="biz_user", name="业务经理", role="business")
token = get_token_for_user(client, username="biz_user", password="admin123")
token = get_token_for_user(client, username="biz_user", password=TEST_USER_PASSWORD)
kpi = create_test_kpi(db, responsible_user="biz_user")
kpi2 = create_test_kpi(db, kpi_code="F_OTHER", kpi_name="无关KPI", responsible_user="其他人")
resp = client.get("/api/cma/dashboard/my-kpis", headers=auth_header(token))
@@ -82,7 +82,7 @@ class TestDashboard:
def test_my_kpis_ceo_sees_all(self, client: TestClient, db: Session):
user = create_test_user(db, username="ceo_user", name="CEO", role="ceo")
token = get_token_for_user(client, username="ceo_user", password="admin123")
token = get_token_for_user(client, username="ceo_user", password=TEST_USER_PASSWORD)
kpi = create_test_kpi(db, kpi_code="F_KPI_A", responsible_user="张三")
resp = client.get("/api/cma/dashboard/my-kpis", headers=auth_header(token))
assert resp.status_code == 200
@@ -103,7 +103,7 @@ class TestDashboard:
def test_my_dashboard(self, client: TestClient, db: Session):
user = create_test_user(db, username="ceo_user", name="CEO", role="ceo")
token = get_token_for_user(client, username="ceo_user", password="admin123")
token = get_token_for_user(client, username="ceo_user", password=TEST_USER_PASSWORD)
kpi = create_test_kpi(db)
resp = client.get("/api/cma/dashboard/my-dashboard", headers=auth_header(token))
assert resp.status_code == 200
backend/tests/test_data_entity_isolation_20260908.py
+5 -5
View File
@@ -13,7 +13,7 @@ from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.models import KPIDefinition, KPIValue, DataSourceConfig, Entity
from tests.conftest import create_test_user, get_token_for_user, auth_header
from tests.conftest import create_test_user, get_token_for_user, auth_header, TEST_USER_PASSWORD
DATA = "/api/cma/data"
@@ -65,7 +65,7 @@ class TestExcelImportEntityIsolation:
k1 = self._mk_kpi(db, "F_REVENUE", 1)
_ensure_entity(db, 2)
create_test_user(db)
token2 = _login(client, "testadmin", "admin123", entity_id=2)
token2 = _login(client, "testadmin", TEST_USER_PASSWORD, entity_id=2)
before = db.query(KPIValue).filter(KPIValue.kpi_id == k1.id).count()
df = pd.DataFrame({
@@ -91,7 +91,7 @@ class TestExcelImportEntityIsolation:
_ensure_entity(db, 2)
k1 = self._mk_kpi(db, "F_REVENUE", 1)
create_test_user(db)
token2 = _login(client, "testadmin", "admin123", entity_id=2)
token2 = _login(client, "testadmin", TEST_USER_PASSWORD, entity_id=2)
df = pd.DataFrame({
"kpi_code": ["F_REVENUE"],
@@ -156,7 +156,7 @@ class TestSourceCRUDEntityIsolation:
_ensure_entity(db, 2)
create_test_user(db)
token1 = get_token_for_user(client)
token2 = _login(client, "testadmin", "admin123", entity_id=2)
token2 = _login(client, "testadmin", TEST_USER_PASSWORD, entity_id=2)
r1 = client.post(f"{DATA}/sources", headers=auth_header(token1),
json={"name": "entity1数据源", "source_type": "manual"})
@@ -178,7 +178,7 @@ class TestSourceCRUDEntityIsolation:
_ensure_entity(db, 2)
create_test_user(db)
token1 = get_token_for_user(client)
token2 = _login(client, "testadmin", "admin123", entity_id=2)
token2 = _login(client, "testadmin", TEST_USER_PASSWORD, entity_id=2)
r1 = client.post(f"{DATA}/sources", headers=auth_header(token1),
json={"name": "entity1的源", "source_type": "manual"})
backend/tests/test_kpi_forecast.py
+2 -2
View File
@@ -13,7 +13,7 @@ from sqlalchemy.orm import Session
from app.models import Entity, KPIValue, User, UserEntity
from app.utils import kpi_forecast_engine as engine
from tests.conftest import create_test_kpi, create_test_user, get_token_for_user, auth_header
from tests.conftest import create_test_kpi, create_test_user, get_token_for_user, auth_header, TEST_USER_PASSWORD
BASE = "/api/cma/predict"
@@ -34,7 +34,7 @@ def _token_for_entity(client: TestClient, db: Session, eid: int) -> str:
db.add(UserEntity(user_id=user.id, entity_id=eid))
db.commit()
resp = client.post("/api/cma/auth/login", json={
"username": "testadmin", "password": "admin123", "entity_id": eid})
"username": "testadmin", "password": TEST_USER_PASSWORD, "entity_id": eid})
assert resp.status_code == 200, f"登录 entity{eid} 失败: {resp.text[:200]}"
return resp.json()["token"]
backend/tests/test_kr_full_fix.py
+2 -1
View File
@@ -9,6 +9,7 @@ from datetime import date
from tests.conftest import (
create_test_user, get_token_for_user, auth_header, create_test_kpi,
TEST_USER_PASSWORD,
)
from app.models import Objective, KR, KPIDefinition, Entity
@@ -253,7 +254,7 @@ class TestMultiTenant:
# 用 entity 2 登录
ent2_user = create_test_user(db, username="user2", role="ceo")
resp2 = client.post("/api/cma/auth/login", json={
"username": "user2", "password": "admin123", "entity_id": 2})
"username": "user2", "password": TEST_USER_PASSWORD, "entity_id": 2})
token2 = resp2.json().get("token")
assert token2
backend/tests/test_map_versions.py
+2 -2
View File
@@ -6,7 +6,7 @@ import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.models import Entity, User, UserEntity, StrategicMap, StrategicMapVersion
from tests.conftest import create_test_user, get_token_for_user, auth_header
from tests.conftest import create_test_user, get_token_for_user, auth_header, TEST_USER_PASSWORD
def _create_second_entity(db: Session, user: User) -> int:
@@ -18,7 +18,7 @@ def _create_second_entity(db: Session, user: User) -> int:
def _token_for_entity(client: TestClient, entity_id: int) -> str:
resp = client.post("/api/cma/auth/login", json={
"username": "testadmin", "password": "admin123", "entity_id": entity_id,
"username": "testadmin", "password": TEST_USER_PASSWORD, "entity_id": entity_id,
})
assert resp.status_code == 200
data = resp.json()
backend/tests/test_maps.py
+2 -2
View File
@@ -5,7 +5,7 @@ import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.models import Entity, User, UserEntity
from tests.conftest import create_test_user, get_token_for_user, auth_header
from tests.conftest import create_test_user, get_token_for_user, auth_header, TEST_USER_PASSWORD
def _create_second_entity(db: Session, user: User) -> int:
@@ -20,7 +20,7 @@ def _token_for_entity(client: TestClient, entity_id: int) -> str:
"""以指定账套登录获取token"""
resp = client.post("/api/cma/auth/login", json={
"username": "testadmin",
"password": "admin123",
"password": TEST_USER_PASSWORD,
"entity_id": entity_id,
})
assert resp.status_code == 200
backend/tests/test_multitenant_ai_20260907.py
+2 -2
View File
@@ -15,7 +15,7 @@ from sqlalchemy.orm import Session
from app.models import KPIDefinition, KPIValue, KPIAlert, ActionPlan, AISuggestion, Entity, UserEntity, ActualCost
from app.api import ai_analysis
from tests.conftest import create_test_kpi, create_test_user, get_token_for_user, auth_header
from tests.conftest import create_test_kpi, create_test_user, get_token_for_user, auth_header, TEST_USER_PASSWORD
def _ensure_entity(db: Session, eid: int) -> Entity:
@@ -44,7 +44,7 @@ def _seed_user(db: Session, entity_id: int):
def _token(client: TestClient, entity_id: int) -> str:
"""登录并返回指定账套的 token（get_token_for_user 固定 entity1，这里手动登录）"""
resp = client.post("/api/cma/auth/login", json={
"username": "testadmin", "password": "admin123", "entity_id": entity_id,
"username": "testadmin", "password": TEST_USER_PASSWORD, "entity_id": entity_id,
})
assert resp.status_code == 200, f"登录 entity{entity_id} 失败: {resp.text[:300]}"
return resp.json().get("token")
backend/tests/test_predict.py
+2 -2
View File
@@ -14,7 +14,7 @@ from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from app.models import CashForecast, ForecastAccuracy, Entity, User, UserEntity
from tests.conftest import create_test_user, get_token_for_user, auth_header
from tests.conftest import create_test_user, get_token_for_user, auth_header, TEST_USER_PASSWORD
BASE = "/api/cma/predict"
@@ -35,7 +35,7 @@ def _token_for_entity(client: TestClient, db: Session, eid: int) -> str:
db.add(UserEntity(user_id=user.id, entity_id=eid))
db.commit()
resp = client.post("/api/cma/auth/login", json={
"username": "testadmin", "password": "admin123", "entity_id": eid})
"username": "testadmin", "password": TEST_USER_PASSWORD, "entity_id": eid})
assert resp.status_code == 200, f"登录 entity{eid} 失败: {resp.text[:200]}"
return resp.json()["token"]
backend/tests/test_reports.py
+2 -2
View File
@@ -9,7 +9,7 @@ generate(weekly/monthly/special) / history / history/{id}
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from tests.conftest import create_test_user, get_token_for_user, auth_header
from tests.conftest import create_test_user, get_token_for_user, auth_header, TEST_USER_PASSWORD
from app.models import (
KPIDefinition, KPIValue, BudgetPlan, StrategicMap, KPIAlert, ActionPlan,
Subject, ReportHistory,
@@ -668,7 +668,7 @@ class TestDupont:
db.add(UserEntity(user_id=user.id, entity_id=entity_id))
db.commit()
resp = client.post("/api/cma/auth/login", json={
"username": "testadmin", "password": "admin123", "entity_id": entity_id,
"username": "testadmin", "password": TEST_USER_PASSWORD, "entity_id": entity_id,
})
assert resp.status_code == 200, f"账套登录失败({entity_id}): {resp.status_code} {resp.text[:200]}"
return resp.json()["token"]
backend/tests/test_security_knowledge_partition.py
+2 -2
View File
@@ -16,7 +16,7 @@ from app.models import KPIDefinition, KPIAlert, Entity
from app.models.knowledge import KnowledgeSummary, KnowledgeEvent
from app.models.knowledge_article import KnowledgeArticle
from app.services import knowledge_service
from tests.conftest import create_test_kpi, create_test_user, get_token_for_user, auth_header
from tests.conftest import create_test_kpi, create_test_user, get_token_for_user, auth_header, TEST_USER_PASSWORD
SUMMARY_FIELDS = dict(
level="daily", period_key="2026-09-01", content="测试摘要内容",
@@ -35,7 +35,7 @@ def _ensure_entity(db: Session, eid: int, name: str = None) -> Entity:
def _login_token(client: TestClient, username: str = "testadmin",
password: str = "admin123", entity_id: int = 1) -> str:
password: str = TEST_USER_PASSWORD, entity_id: int = 1) -> str:
"""登录指定账套返回 token（conftest.get_token_for_user 固定 entity1，这里参数化）"""
resp = client.post("/api/cma/auth/login", json={
"username": username, "password": password, "entity_id": entity_id,
backend/tests/test_security_multitenant.py
+9 -9
View File
@@ -13,7 +13,7 @@ from sqlalchemy.orm import Session
from app.models import KPIDefinition, KPIAlert, Entity, UserEntity
from app.api.alert_rules import AlertRule, _eval_threshold
from tests.conftest import create_test_kpi, create_test_user, get_token_for_user, auth_header
from tests.conftest import create_test_kpi, create_test_user, get_token_for_user, auth_header, TEST_USER_PASSWORD
BOT_KEY = {"X-BOT-KEY": "cma-bot-finance-2026"}
@@ -103,7 +103,7 @@ class TestAlertRulesIsolation:
k1 = _seed_entity_kpi(db, entity_id=1, code="AR_E1")
db.commit()
create_test_user(db)
token = get_token_for_user(client, "testadmin", "admin123")
token = get_token_for_user(client, "testadmin", TEST_USER_PASSWORD)
headers = auth_header(token)
resp = client.post("/api/cma/alert-rules", headers=headers, json={
@@ -128,7 +128,7 @@ class TestAlertRulesIsolation:
entity_id=1))
db.commit()
create_test_user(db)
token = get_token_for_user(client, "testadmin", "admin123")
token = get_token_for_user(client, "testadmin", TEST_USER_PASSWORD)
r1 = client.get(f"/api/cma/alert-rules/kpi/{k1.id}", headers=auth_header(token))
assert r1.status_code == 200
@@ -149,7 +149,7 @@ class TestReportsIsolation:
data_status="verified", entity_id=1))
db.commit()
create_test_user(db)
token = get_token_for_user(client, "testadmin", "admin123")
token = get_token_for_user(client, "testadmin", TEST_USER_PASSWORD)
r1 = client.get("/api/cma/reports/profit-summary?period=2026-06", headers=auth_header(token))
assert r1.status_code == 200
@@ -164,7 +164,7 @@ class TestReportsIsolation:
_seed_entity_kpi(db, entity_id=2, code="TR_E2")
db.commit()
create_test_user(db)
token = get_token_for_user(client, "testadmin", "admin123")
token = get_token_for_user(client, "testadmin", TEST_USER_PASSWORD)
r1 = client.get("/api/cma/reports/kpi-trends", headers=auth_header(token))
assert r1.status_code == 200
@@ -212,7 +212,7 @@ class TestCheckGovernanceRetired:
_seed_entity_kpi(db, entity_id=1, code="GOV_KPI")
db.commit()
create_test_user(db)
token = get_token_for_user(client, "testadmin", "admin123")
token = get_token_for_user(client, "testadmin", TEST_USER_PASSWORD)
resp = client.get("/api/cma/data-quality/check-governance", headers=auth_header(token))
assert resp.status_code == 404, f"应 404，实际 {resp.status_code}: {resp.text[:200]}"
@@ -225,7 +225,7 @@ class TestCheckGovernanceRetired:
_seed_entity_kpi(db, entity_id=1, code="GOV2_KPI")
db.commit()
create_test_user(db)
token = get_token_for_user(client, "testadmin", "admin123")
token = get_token_for_user(client, "testadmin", TEST_USER_PASSWORD)
resp = client.get("/api/cma/data-quality/governance-check", headers=auth_header(token))
assert resp.status_code == 200, resp.text[:300]
@@ -242,7 +242,7 @@ class TestCrossEntityEndpointAuth:
def _login(self, client, db):
_ensure_entity(db, 2)
create_test_user(db)
return get_token_for_user(client, "testadmin", "admin123") # 登录账套 entity_id=1
return get_token_for_user(client, "testadmin", TEST_USER_PASSWORD) # 登录账套 entity_id=1
# ── 1. data-quality/governance-check ──────────────────
def test_governance_check_cross_entity_403(self, client, db):
@@ -352,7 +352,7 @@ class TestDupontSessionScope:
def _login(self, client, db):
_ensure_entity(db, 2)
create_test_user(db)
return get_token_for_user(client, "testadmin", "admin123") # 登录账套 entity_id=1
return get_token_for_user(client, "testadmin", TEST_USER_PASSWORD) # 登录账套 entity_id=1
def test_dupont_cross_entity_bohai_403(self, client, db):
"""酣客会话显式 entity=bohai（博海真实KPI）→ 403（旧豁免已撤销）"""