diff --git a/backend/app/api/knowledge.py b/backend/app/api/knowledge.py index 907059d5..c936ea9c 100644 --- a/backend/app/api/knowledge.py +++ b/backend/app/api/knowledge.py @@ -15,6 +15,7 @@ from typing import Optional from app.database import get_db from app.auth_middleware import require_role +from app.deps import get_entity_id from app.models import KnowledgeEvent, KnowledgeSummary from app.services.knowledge_service import ( generate_summary_sync, @@ -59,18 +60,22 @@ def list_summaries( limit: int = 20, offset: int = 0, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), ): - """获取摘要列表,按层级筛选,按时间倒序""" + """获取摘要列表,按层级筛选,按时间倒序(按企业抽屉隔离)""" query = db.query(KnowledgeSummary) + query = query.filter(KnowledgeSummary.entity_id == entity_id) if level: query = query.filter(KnowledgeSummary.level == level) query = query.order_by(desc(KnowledgeSummary.id)).offset(offset).limit(limit) total = db.query(func.count(KnowledgeSummary.id)).select_from(KnowledgeSummary) + total = total.filter(KnowledgeSummary.entity_id == entity_id) if level: total = total.filter(KnowledgeSummary.level == level) total = total.scalar() return { "total": total, + "entity_id": entity_id, "items": [summary_to_dict(s) for s in query.all()], } @@ -79,18 +84,30 @@ def list_summaries( def latest_summary( level: str = Query("daily", description="层级: daily/weekly/monthly/cumulative"), db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), ): - """获取指定层级的最新摘要""" - s = get_last_summary(db, level) + """获取指定层级的最新摘要(按企业隔离)""" + s = get_last_summary(db, level, entity_id) if not s: return {"detail": f"没有{level}层级的摘要"}, 404 return summary_to_dict(s) @router.get("/summaries/{summary_id}") -def get_summary(summary_id: int, db: Session = Depends(get_db)): - """获取单条摘要详情""" - s = db.query(KnowledgeSummary).filter(KnowledgeSummary.id == summary_id).first() +def get_summary( + summary_id: int, + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), +): + """获取单条摘要详情(仅本企业,防跨企业探测)""" + s = ( + db.query(KnowledgeSummary) + .filter( + KnowledgeSummary.id == summary_id, + KnowledgeSummary.entity_id == entity_id, + ) + .first() + ) if not s: raise HTTPException(status_code=404, detail="摘要不存在") return summary_to_dict(s) @@ -105,8 +122,9 @@ def list_events( until: Optional[str] = None, limit: int = 50, db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), ): - """查询未摘要的原始事件 + """查询未摘要的原始事件(按企业隔离) 如果不传时间,默认返回最近7天的操作记录和预警。 """ @@ -116,18 +134,21 @@ def list_events( except ValueError: raise HTTPException(status_code=400, detail="时间格式错误,请使用 ISO 格式如 2026-06-01T00:00:00") - events = extract_events(db, dt_since, dt_until) - return {"since": dt_since.isoformat(), "until": dt_until.isoformat(), "total": len(events), "events": events[:limit]} + events = extract_events(db, dt_since, dt_until, entity_id=entity_id) + return {"since": dt_since.isoformat(), "until": dt_until.isoformat(), "entity_id": entity_id, "total": len(events), "events": events[:limit]} # ── 手动触发 ── @router.post("/generate/daily") -def trigger_daily_summary(db: Session = Depends(get_db)): - """手动触发每日摘要生成""" +def trigger_daily_summary( + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), +): + """手动触发每日摘要生成(写入当前企业抽屉)""" try: - result = generate_daily_sync(db) + result = generate_daily_sync(db, entity_id=entity_id) return {"message": "每日摘要已生成", "summary": result} except Exception as e: logger.exception("每日摘要生成失败") @@ -135,10 +156,13 @@ def trigger_daily_summary(db: Session = Depends(get_db)): @router.post("/generate/weekly") -def trigger_weekly_summary(db: Session = Depends(get_db)): - """手动触发周度摘要生成""" +def trigger_weekly_summary( + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), +): + """手动触发周度摘要生成(写入当前企业抽屉)""" try: - result = generate_weekly_sync(db) + result = generate_weekly_sync(db, entity_id=entity_id) return {"message": "周度摘要已生成", "summary": result} except Exception as e: logger.exception("周度摘要生成失败") @@ -146,10 +170,13 @@ def trigger_weekly_summary(db: Session = Depends(get_db)): @router.post("/generate/monthly") -def trigger_monthly_summary(db: Session = Depends(get_db)): - """手动触发月度摘要生成""" +def trigger_monthly_summary( + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), +): + """手动触发月度摘要生成(写入当前企业抽屉)""" try: - result = generate_monthly_sync(db) + result = generate_monthly_sync(db, entity_id=entity_id) return {"message": "月度摘要已生成", "summary": result} except Exception as e: logger.exception("月度摘要生成失败") diff --git a/backend/app/api/knowledge_articles.py b/backend/app/api/knowledge_articles.py index afcd1adf..fa783f5d 100644 --- a/backend/app/api/knowledge_articles.py +++ b/backend/app/api/knowledge_articles.py @@ -4,9 +4,11 @@ """ from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session +from sqlalchemy import or_ from typing import Optional from app.database import get_db from app.auth_middleware import require_role +from app.deps import get_entity_id from app.models.knowledge_article import KnowledgeArticle router = APIRouter(prefix="/api/cma/knowledge-articles", tags=["知识库嵌入"], @@ -17,6 +19,7 @@ router = APIRouter(prefix="/api/cma/knowledge-articles", tags=["知识库嵌入" def article_to_dict(a: KnowledgeArticle) -> dict: return { "id": a.id, + "entity_id": a.entity_id, "title": a.title, "summary": a.summary, "content": a.content, @@ -32,9 +35,21 @@ def list_articles( related_page: Optional[str] = Query(None, description="按关联页面路由筛选"), category: Optional[str] = Query(None), db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), ): - """查询知识文章,可按关联页面或分类筛选""" - q = db.query(KnowledgeArticle) + """查询知识文章(本企业专属 ∪ 全局共享),可按关联页面或分类筛选 + + 隔离语义(2026-09-03 多租户分区): + - entity_id IS NULL → 全局共享(CMA 系统教学知识,所有企业可见) + - entity_id == 当前账套 → 企业专属(只本企业可见) + - 其他企业的专属文章不可见 + """ + q = db.query(KnowledgeArticle).filter( + or_( + KnowledgeArticle.entity_id == entity_id, + KnowledgeArticle.entity_id.is_(None), + ) + ) if related_page: q = q.filter(KnowledgeArticle.related_page.contains(related_page)) if category: @@ -44,8 +59,19 @@ def list_articles( @router.get("/{article_id}") -def get_article(article_id: int, db: Session = Depends(get_db)): - a = db.query(KnowledgeArticle).filter(KnowledgeArticle.id == article_id).first() +def get_article( + article_id: int, + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id), +): + """获取文章详情(企业专属文章仅本企业可见,防跨企业探测)""" + a = ( + db.query(KnowledgeArticle) + .filter(KnowledgeArticle.id == article_id) + .first() + ) if not a: raise HTTPException(404, "文章不存在") + if a.entity_id is not None and a.entity_id != entity_id: + raise HTTPException(404, "文章不存在") return article_to_dict(a) diff --git a/backend/app/models/knowledge.py b/backend/app/models/knowledge.py index 71f95427..e001584e 100644 --- a/backend/app/models/knowledge.py +++ b/backend/app/models/knowledge.py @@ -22,6 +22,7 @@ class KnowledgeEvent(Base): __tablename__ = "knowledge_events" id = Column(Integer, primary_key=True, index=True) + entity_id = Column(Integer, default=1, index=True, comment="企业ID (多租户隔离 2026-09-03)") event_type = Column(String(30), nullable=False, comment="事件类型: kpi_change/alert/decision/plan/map/import/user_action") event_level = Column(String(20), default="info", comment="info/warning/important/critical") source = Column(String(50), nullable=True, comment="来源: operation_log/api/erp_sync/manual") @@ -47,6 +48,7 @@ class KnowledgeSummary(Base): __tablename__ = "knowledge_summaries" id = Column(Integer, primary_key=True, index=True) + entity_id = Column(Integer, default=1, index=True, comment="企业ID (多租户隔离 2026-09-03)") level = Column(String(20), nullable=False, comment="摘要层级: daily/weekly/monthly/cumulative") period_key = Column(String(20), nullable=False, comment="期间标识: 2026-06-12 / 2026-W24 / 2026-06 / cumulative") title = Column(String(300), nullable=False, comment="摘要标题") diff --git a/backend/app/models/knowledge_article.py b/backend/app/models/knowledge_article.py index 324d61f9..ab91d73c 100644 --- a/backend/app/models/knowledge_article.py +++ b/backend/app/models/knowledge_article.py @@ -12,6 +12,7 @@ class KnowledgeArticle(Base): __tablename__ = "knowledge_articles" id = Column(Integer, primary_key=True, index=True) + entity_id = Column(Integer, nullable=True, index=True, comment="企业ID: NULL=全局共享(系统教学), 有值=企业专属 (多租户隔离 2026-09-03)") title = Column(String(200), nullable=False, comment="文章标题") summary = Column(String(500), nullable=True, comment="一句话摘要") content = Column(Text, nullable=False, comment="文章正文(支持Markdown)") diff --git a/backend/app/services/knowledge_service.py b/backend/app/services/knowledge_service.py index 426fd4d9..57b7c498 100644 --- a/backend/app/services/knowledge_service.py +++ b/backend/app/services/knowledge_service.py @@ -19,7 +19,7 @@ from typing import Optional from sqlalchemy.orm import Session from sqlalchemy import func, and_ -from app.models import KnowledgeEvent, KnowledgeSummary, OperationLog, KPIAlert, KPIDefinition, KPIValue +from app.models import KnowledgeEvent, KnowledgeSummary, OperationLog, KPIAlert, KPIDefinition, KPIValue, UserEntity logger = logging.getLogger("cma.knowledge") @@ -66,16 +66,31 @@ async def _call_deepseek(prompt: str, timeout: int = 30) -> str: # ── 事件抽取 ── -def extract_events(db: Session, since: datetime, until: Optional[datetime] = None) -> list[dict]: - """从 OperationLog + KPIAlert 中抽取关键事件 +def extract_events(db: Session, since: datetime, until: Optional[datetime] = None, entity_id: int = 1) -> list[dict]: + """从 OperationLog + KPIAlert 中抽取关键事件(按企业抽屉隔离) 返回 dict 列表,用于喂给摘要 prompt + + 隔离说明(2026-09-03 多租户分区): + - KPIAlert 带 entity_id 列 → 直接按 entity_id 过滤 + - OperationLog 无 entity_id 列(数据债,P2 follow-up)→ 经 user_id→user_entities + 归属当前企业;无授权归属的日志(user_id NULL/未授权)不进入本企业记忆 """ now = until or datetime.utcnow() - # 1. 操作日志 → 事件 + # 0. 当前企业授权用户集合(OperationLog 归属过滤用) + auth_user_ids = [ + uid for (uid,) in db.query(UserEntity.user_id) + .filter(UserEntity.entity_id == entity_id).all() + ] + + # 1. 操作日志 → 事件(本企业授权用户的操作) logs = ( db.query(OperationLog) - .filter(OperationLog.created_at >= since, OperationLog.created_at <= now) + .filter( + OperationLog.created_at >= since, + OperationLog.created_at <= now, + OperationLog.user_id.in_(auth_user_ids), # 空集合 → IN (NULL) 不匹配任何行 + ) .order_by(OperationLog.created_at.asc()) .all() ) @@ -96,11 +111,15 @@ def extract_events(db: Session, since: datetime, until: Optional[datetime] = Non "detail": detail_str, }) - # 2. 预警记录 → 事件 + # 2. 预警记录 → 事件(KPIAlert 直接按企业过滤) alerts = ( db.query(KPIAlert, KPIDefinition.kpi_name, KPIDefinition.kpi_code) .join(KPIDefinition, KPIAlert.kpi_id == KPIDefinition.id) - .filter(KPIAlert.created_at >= since, KPIAlert.created_at <= now) + .filter( + KPIAlert.created_at >= since, + KPIAlert.created_at <= now, + KPIAlert.entity_id == entity_id, + ) .order_by(KPIAlert.created_at.asc()) .all() ) @@ -136,9 +155,12 @@ def _build_summary_prompt(events: list[dict], level: str, period_key: str) -> st return "\n".join(lines) -async def _kpi_snapshot(db: Session) -> list[dict]: - """当前KPI快照 — 每个活跃KPI的最新实际值""" - kpis = db.query(KPIDefinition).filter(KPIDefinition.status == "active").all() +async def _kpi_snapshot(db: Session, entity_id: int = 1) -> list[dict]: + """当前KPI快照 — 本企业每个活跃KPI的最新实际值(按账套隔离)""" + kpis = db.query(KPIDefinition).filter( + KPIDefinition.status == "active", + KPIDefinition.entity_id == entity_id, + ).all() snapshot = [] for k in kpis: latest = ( @@ -165,6 +187,7 @@ async def generate_summary( since: datetime, until: Optional[datetime] = None, prev_summary: Optional[KnowledgeSummary] = None, + entity_id: int = 1, ) -> KnowledgeSummary: """生成一层摘要(daily/weekly/monthly/cumulative) @@ -174,11 +197,12 @@ async def generate_summary( since: 事件开始时间 until: 事件结束时间 prev_summary: 前一层摘要(用于累积摘要继承) + entity_id: 企业账套(企业记忆分区 2026-09-03) """ now = until or datetime.utcnow() - # 1. 抽取事件 - events = extract_events(db, since, now) + # 1. 抽取事件(本企业) + events = extract_events(db, since, now, entity_id=entity_id) # 2. 构建 prompt prompt = _build_summary_prompt(events, level, period_key) @@ -194,11 +218,12 @@ async def generate_summary( if not result_text or "无重要变化" in result_text: result_text = f"{level}汇总: 窗口 {period_key} 内共 {len(events)} 条事件记录,无重大变化需记录。" - # 6. 提取 KPI 快照 - kpi_snapshot = await _kpi_snapshot(db) + # 6. 提取 KPI 快照(本企业) + kpi_snapshot = await _kpi_snapshot(db, entity_id=entity_id) # 7. 存入数据库 summary = KnowledgeSummary( + entity_id=entity_id, level=level, period_key=period_key, title=f"{level.upper()}摘要 - {period_key}", @@ -223,32 +248,36 @@ async def generate_summary( # ── 分层调度 ── -def get_last_summary(db: Session, level: str) -> Optional[KnowledgeSummary]: - """获取该层最新(最新创建)的摘要""" +def get_last_summary(db: Session, level: str, entity_id: int = 1) -> Optional[KnowledgeSummary]: + """获取该层最新(最新创建)的摘要(按企业隔离)""" return ( db.query(KnowledgeSummary) - .filter(KnowledgeSummary.level == level, KnowledgeSummary.is_stale == 0) + .filter( + KnowledgeSummary.level == level, + KnowledgeSummary.is_stale == 0, + KnowledgeSummary.entity_id == entity_id, + ) .order_by(KnowledgeSummary.id.desc()) .first() ) -async def run_daily_summary(db: Session) -> KnowledgeSummary: - """运行每日摘要""" +async def run_daily_summary(db: Session, entity_id: int = 1) -> KnowledgeSummary: + """运行每日摘要(企业记忆分区)""" today = date.today() period_key = today.isoformat() since = datetime(today.year, today.month, today.day) # 获取前一天摘要作为 prev yesterday = today - timedelta(days=1) - prev = get_last_summary(db, "daily") + prev = get_last_summary(db, "daily", entity_id=entity_id) return await generate_summary( - db, "daily", period_key, since, prev_summary=prev, + db, "daily", period_key, since, prev_summary=prev, entity_id=entity_id, ) -async def run_weekly_summary(db: Session) -> KnowledgeSummary: +async def run_weekly_summary(db: Session, entity_id: int = 1) -> KnowledgeSummary: """运行每周摘要(周日执行)""" today = date.today() # ISO 周算法: 本周一到今天 @@ -257,13 +286,13 @@ async def run_weekly_summary(db: Session) -> KnowledgeSummary: since = today - timedelta(days=today.weekday()) # 本周一 since_dt = datetime(since.year, since.month, since.day) - prev = get_last_summary(db, "weekly") + prev = get_last_summary(db, "weekly", entity_id=entity_id) return await generate_summary( - db, "weekly", period_key, since_dt, prev_summary=prev, + db, "weekly", period_key, since_dt, prev_summary=prev, entity_id=entity_id, ) -async def run_monthly_summary(db: Session) -> KnowledgeSummary: +async def run_monthly_summary(db: Session, entity_id: int = 1) -> KnowledgeSummary: """运行月度摘要""" today = date.today() period_key = today.strftime("%Y-%m") @@ -271,10 +300,10 @@ async def run_monthly_summary(db: Session) -> KnowledgeSummary: # 附属前一个月的 daily 和 weekly 摘要 prev_month = today.replace(day=1) - timedelta(days=1) - prev = get_last_summary(db, "monthly") + prev = get_last_summary(db, "monthly", entity_id=entity_id) return await generate_summary( - db, "monthly", period_key, since, prev_summary=prev, + db, "monthly", period_key, since, prev_summary=prev, entity_id=entity_id, ) @@ -289,40 +318,41 @@ def generate_summary_sync( period_key: str, since: datetime, until: Optional[datetime] = None, + entity_id: int = 1, ) -> dict: """同步包装,用于 API 手动触发""" loop = asyncio.new_event_loop() try: summary = loop.run_until_complete( - generate_summary(db, level, period_key, since, until) + generate_summary(db, level, period_key, since, until, entity_id=entity_id) ) - return {"id": summary.id, "level": summary.level, "period_key": summary.period_key} + return {"id": summary.id, "entity_id": summary.entity_id, "level": summary.level, "period_key": summary.period_key} finally: loop.close() -def generate_daily_sync(db: Session) -> dict: +def generate_daily_sync(db: Session, entity_id: int = 1) -> dict: loop = asyncio.new_event_loop() try: - summary = loop.run_until_complete(run_daily_summary(db)) - return {"id": summary.id, "level": summary.level, "period_key": summary.period_key} + summary = loop.run_until_complete(run_daily_summary(db, entity_id)) + return {"id": summary.id, "entity_id": summary.entity_id, "level": summary.level, "period_key": summary.period_key} finally: loop.close() -def generate_weekly_sync(db: Session) -> dict: +def generate_weekly_sync(db: Session, entity_id: int = 1) -> dict: loop = asyncio.new_event_loop() try: - summary = loop.run_until_complete(run_weekly_summary(db)) - return {"id": summary.id, "level": summary.level, "period_key": summary.period_key} + summary = loop.run_until_complete(run_weekly_summary(db, entity_id)) + return {"id": summary.id, "entity_id": summary.entity_id, "level": summary.level, "period_key": summary.period_key} finally: loop.close() -def generate_monthly_sync(db: Session) -> dict: +def generate_monthly_sync(db: Session, entity_id: int = 1) -> dict: loop = asyncio.new_event_loop() try: - summary = loop.run_until_complete(run_monthly_summary(db)) - return {"id": summary.id, "level": summary.level, "period_key": summary.period_key} + summary = loop.run_until_complete(run_monthly_summary(db, entity_id)) + return {"id": summary.id, "entity_id": summary.entity_id, "level": summary.level, "period_key": summary.period_key} finally: loop.close() diff --git a/backend/tests/test_security_knowledge_partition.py b/backend/tests/test_security_knowledge_partition.py new file mode 100644 index 00000000..ecbc7d6a --- /dev/null +++ b/backend/tests/test_security_knowledge_partition.py @@ -0,0 +1,220 @@ +"""企业记忆分区(knowledge 侧 entity_id 多租户隔离)测试(2026-09-03 P1) + +覆盖任务 DoD: +- knowledge_events / knowledge_summaries / knowledge_articles 三表 entity_id 隔离 +- 摘要列表/详情强制 entity 过滤(entity1 查不到 entity2 → 404/空集) +- 摘要生成自动带 entity_id(写入当前账套抽屉) +- 知识文章:全局共享(NULL) ∪ 本企业专属;他企专属 404 +- extract_events 按企业隔离(KPIAlert 直过滤 + OperationLog 经 user_entities 归属) +""" +import pytest +from datetime import datetime, timedelta +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +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 + +SUMMARY_FIELDS = dict( + level="daily", period_key="2026-09-01", content="测试摘要内容", + generated_by="test", is_stale=0, +) + + +def _ensure_entity(db: Session, eid: int, name: str = None) -> Entity: + ent = db.query(Entity).filter(Entity.id == eid).first() + if not ent: + ent = Entity(id=eid, name=name or f"企业{eid}", short_name=f"E{eid}", status="active") + db.add(ent) + db.commit() + db.refresh(ent) + return ent + + +def _login_token(client: TestClient, username: str = "testadmin", + password: str = "admin123", 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, + }) + assert resp.status_code == 200, resp.text + return resp.json().get("token") + + +# ============================================================ +# summaries 抽屉隔离 +# ============================================================ +class TestKnowledgeSummaryIsolation: + def test_list_summaries_entity_scoped(self, client: TestClient, db: Session): + """摘要列表按 entity 隔离:entity1 看不到 entity2 的摘要""" + _ensure_entity(db, 2) + s1 = KnowledgeSummary(**SUMMARY_FIELDS, title="E1摘要", entity_id=1) + s2 = KnowledgeSummary(**SUMMARY_FIELDS, title="E2专属摘要", entity_id=2) + db.add_all([s1, s2]) + db.commit() + db.refresh(s1) + db.refresh(s2) + + create_test_user(db) + token1 = _login_token(client, entity_id=1) + r1 = client.get("/api/cma/knowledge/summaries", headers=auth_header(token1)) + assert r1.status_code == 200, r1.text + items = r1.json()["items"] + titles = [i["title"] for i in items] + assert any("摘要" in t for t in titles) + assert "E2专属摘要" not in titles + + def test_get_summary_cross_entity_404(self, client: TestClient, db: Session): + """越权:entity1 直取 entity2 摘要详情 → 404(防跨企业探测)""" + _ensure_entity(db, 2) + s2 = KnowledgeSummary(**SUMMARY_FIELDS, title="E2秘密摘要", entity_id=2) + db.add(s2) + db.commit() + db.refresh(s2) + + create_test_user(db) + token1 = _login_token(client, entity_id=1) + r = client.get(f"/api/cma/knowledge/summaries/{s2.id}", headers=auth_header(token1)) + assert r.status_code == 404 + + # 反向:entity2 取 entity1 摘要同样 404 + s1 = KnowledgeSummary(**SUMMARY_FIELDS, title="E1摘要", entity_id=1) + db.add(s1) + db.commit() + db.refresh(s1) + token2 = _login_token(client, entity_id=2) + r2 = client.get(f"/api/cma/knowledge/summaries/{s1.id}", headers=auth_header(token2)) + assert r2.status_code == 404 + + def test_get_summary_own_entity_ok(self, client: TestClient, db: Session): + """同企业取自己摘要详情 → 200""" + s1 = KnowledgeSummary(**SUMMARY_FIELDS, title="E1自己的摘要", entity_id=1) + db.add(s1) + db.commit() + db.refresh(s1) + create_test_user(db) + token1 = _login_token(client, entity_id=1) + r = client.get(f"/api/cma/knowledge/summaries/{s1.id}", headers=auth_header(token1)) + assert r.status_code == 200 + assert r.json()["title"] == "E1自己的摘要" + + +# ============================================================ +# 摘要生成自动带 entity_id +# ============================================================ +class TestKnowledgeGenerateWritesEntity: + def test_generate_daily_writes_current_entity(self, client: TestClient, db: Session, monkeypatch): + """generate/daily 写入当前账套抽屉:entity1 触发 → 摘要 entity_id=1;entity2 触发 → =2""" + _ensure_entity(db, 2) + + async def _fake_deepseek(prompt: str, timeout: int = 30) -> str: + return "测试摘要:本期无重大变化。" + monkeypatch.setattr(knowledge_service, "_call_deepseek", _fake_deepseek) + + create_test_user(db) + token1 = _login_token(client, entity_id=1) + r1 = client.post("/api/cma/knowledge/generate/daily", headers=auth_header(token1)) + assert r1.status_code == 200, r1.text + assert r1.json()["summary"]["entity_id"] == 1 + + token2 = _login_token(client, entity_id=2) + r2 = client.post("/api/cma/knowledge/generate/daily", headers=auth_header(token2)) + assert r2.status_code == 200, r2.text + assert r2.json()["summary"]["entity_id"] == 2 + + # DB 回查:两账套抽屉各一条自己的 + e1 = db.query(KnowledgeSummary).filter(KnowledgeSummary.entity_id == 1).count() + e2 = db.query(KnowledgeSummary).filter(KnowledgeSummary.entity_id == 2).count() + assert e1 >= 1 + assert e2 >= 1 + + +# ============================================================ +# 知识文章:全局共享 ∪ 本企业专属 +# ============================================================ +class TestKnowledgeArticlesGlobalUnion: + def test_articles_global_union_scope(self, client: TestClient, db: Session): + """文章查询 = 本企业专属 ∪ 全局共享;他企专属不可见""" + _ensure_entity(db, 2) + g = KnowledgeArticle(title="全局教学文章", content="c", category="term", entity_id=None) + e1 = KnowledgeArticle(title="E1专属文章", content="c", category="term", entity_id=1) + e2 = KnowledgeArticle(title="E2专属文章", content="c", category="term", entity_id=2) + db.add_all([g, e1, e2]) + db.commit() + for a in (g, e1, e2): + db.refresh(a) + + create_test_user(db) + token1 = _login_token(client, entity_id=1) + r1 = client.get("/api/cma/knowledge-articles", headers=auth_header(token1)) + assert r1.status_code == 200, r1.text + titles1 = {i["title"] for i in r1.json()["data"]} + assert "全局教学文章" in titles1 # 全局共享可见 + assert "E1专属文章" in titles1 # 本企业专属可见 + assert "E2专属文章" not in titles1 # 他企专属不可见 + + token2 = _login_token(client, entity_id=2) + r2 = client.get("/api/cma/knowledge-articles", headers=auth_header(token2)) + titles2 = {i["title"] for i in r2.json()["data"]} + assert "全局教学文章" in titles2 + assert "E2专属文章" in titles2 + assert "E1专属文章" not in titles2 + + def test_get_article_cross_entity_404(self, client: TestClient, db: Session): + """越权:entity1 直取 entity2 专属文章 → 404;全局文章两账套可取""" + _ensure_entity(db, 2) + g = KnowledgeArticle(title="全局文章", content="c", entity_id=None) + e2 = KnowledgeArticle(title="E2秘密文章", content="c", entity_id=2) + db.add_all([g, e2]) + db.commit() + db.refresh(g) + db.refresh(e2) + + create_test_user(db) + token1 = _login_token(client, entity_id=1) + assert client.get(f"/api/cma/knowledge-articles/{g.id}", headers=auth_header(token1)).status_code == 200 + assert client.get(f"/api/cma/knowledge-articles/{e2.id}", headers=auth_header(token1)).status_code == 404 + + +# ============================================================ +# extract_events 企业隔离(service 层精确断言) +# ============================================================ +class TestExtractEventsIsolation: + def _seed_alerts(self, db: Session): + k1 = create_test_kpi(db, kpi_code="KE_E1", entity_id=1, status="active") + k2 = create_test_kpi(db, kpi_code="KE_E2", entity_id=2, status="active") + db.add(KPIAlert(kpi_id=k1.id, alert_level="red", alert_message="E1企业预警信息", + status="pending", entity_id=1)) + db.add(KPIAlert(kpi_id=k2.id, alert_level="red", alert_message="E2企业预警信息", + status="pending", entity_id=2)) + db.commit() + return k1, k2 + + def test_extract_events_only_own_entity(self, client: TestClient, db: Session): + """extract_events(entity_id=1) 只含 E1 的预警,不含 E2""" + _ensure_entity(db, 2) + self._seed_alerts(db) + since = datetime.utcnow() - timedelta(days=1) + ev1 = knowledge_service.extract_events(db, since, None, entity_id=1) + ev2 = knowledge_service.extract_events(db, since, None, entity_id=2) + txt1 = "\n".join(str(e) for e in ev1) + txt2 = "\n".join(str(e) for e in ev2) + assert "E1企业预警信息" in txt1 + assert "E2企业预警信息" not in txt1 # 核心越权断言:E1 抽不到 E2 + assert "E2企业预警信息" in txt2 + assert "E1企业预警信息" not in txt2 + + def test_api_events_entity_scoped(self, client: TestClient, db: Session): + """API /knowledge/events 按 token 账套过滤事件""" + _ensure_entity(db, 2) + self._seed_alerts(db) + create_test_user(db) + token1 = _login_token(client, entity_id=1) + r = client.get("/api/cma/knowledge/events", headers=auth_header(token1)) + assert r.status_code == 200, r.text + body = "\n".join(str(e) for e in r.json()["events"]) + assert "E1企业预警信息" in body + assert "E2企业预警信息" not in body diff --git a/docs/knowledge-entity-partition-plan-20260903.md b/docs/knowledge-entity-partition-plan-20260903.md new file mode 100644 index 00000000..223093ab --- /dev/null +++ b/docs/knowledge-entity-partition-plan-20260903.md @@ -0,0 +1,95 @@ +# CMA 知识侧 entity_id 多租户分区——迁移方案(评审稿) + +> 任务:CMA-entity-memory-partition-20260903(研学调度 P1) +> 提出者:yanxueBot → 执行人:wecom-fullstack +> 决策依据:/root/WB_WS/_research/architecture-shared-intelligence-isolated-data-20260903.md 第八节行动2(🔴 唯一隔离缺口:客户侧知识未按 entity_id 分区) +> 日期:2026-09-03 | 状态:评审稿(代码落地后由项目Bot独立复核) + +## 一、调研实证(先查系统再决策) + +### 1.1 生产库现状(2026-09-03 实测) + +| 表 | 行数 | entity_id | 写入路径 | 语义 | +|---|---|---|---|---| +| knowledge_events | **0** | ❌ 无 | 无任何写入路径(extract_events 只返回 dict 喂 prompt,不落库) | 企业事件记忆(设计态) | +| knowledge_summaries | **0** | ❌ 无 | 仅 API 手动触发 /knowledge/generate/*(knowledge_service.generate_summary) | 企业知识摘要(设计态) | +| knowledge_articles | **38** | ❌ 无 | seed_knowledge_articles.py / import_cma_knowledge.py(系统教学知识 seed) | CMA 教学知识库(term/formula/practice/方法论/案例) | + +### 1.2 账套事实(纠正任务单假设) + +- **entity 1 = 陕西酣客文化传媒(白酒经销, active)** +- **entity 2 = 陕西博海网络科技(IT服务, active)** +- entity 3-5 = 测试/demo 企业 + +任务单写"博海=entity 1 兜底?酣客=entity 2?"与生产**相反**。实际无需"归属确认"到具体企业——因为见 1.3。 + +### 1.3 存量归属结论(本次方案的关键决策点) + +1. **knowledge_events / knowledge_summaries:零存量**(0 行)→ 无迁移负担。加列后新写入必须带 entity_id(从请求上下文解析)。 +2. **knowledge_articles 38 行 = CMA 系统教学知识**(术语/公式/实务/方法论/CMA案例/税务合规等),**语义属于共享层**——对照架构决策文档: + > 共享层(越大越好):判断模型/方法论/科目映射模式库/合规规则 + > 隔离层(必须严格):企业数据/账套/权限/审计/记忆 + + 教学文章不是"企业 A 的记忆 vs 企业 B 的记忆",而是"CMA 方法论知识库"(所有企业共享,越大越好)。 + + **→ 归属映射 = NULL = 全局共享**(不归酣客也不归博海)。 + 若硬性归 entity=1,则 entity 2(博海)登录后 CMA 知识库页/嵌入知识全空——破坏既有产品体验,也违背"共享层不重复"原则。 + **附带推论**:任务验收"存量数据归属映射已确认"的答案 = 存量文章保持全局共享(entity_id NULL),只有未来企业自定义文章才进入企业抽屉。若任总/研学希望个别文章改为企业专属,一条 `UPDATE knowledge_articles SET entity_id=N WHERE id=...` 即可,成本可忽略。 + +### 1.4 数据债边界(本次不扩,记录 follow-up) + +- **operation_logs 无 entity_id 列**(模型+库均无;写入点分布在 kpis/data/ai_suggestions/budget_generate/alert_rules/auth/templates 等 7+ 文件)。任务验收"审计:OperationLog 记录知识操作的企业归属"**无法在当前表结构下实现**——为 operation_logs 加 entity_id 属独立数据债(影响全部 7+ 写入点,需要全量排查 user→entity 归属,与 2026-08-31 安全修复同级工作量),本次不动,记为 P2 follow-up。 +- 折衷:知识操作本身走 require_role + Depends(get_entity_id),操作发生在**已隔离的企业上下文内**;OperationLog.user_id → user_entities 可回溯归属。隔离缺口不因 operation_logs 无列而扩大(日志不是知识数据出口)。 + +## 二、设计定稿 + +### 2.1 三类语义 × 隔离策略 + +| 表 | entity_id 设计 | 查询语义 | 写入语义 | +|---|---|---|---| +| knowledge_events | `INT NOT NULL DEFAULT 1` + 索引 | 强制 `entity_id == 当前账套` | 事件抽取/记录必须从上下文带 entity_id | +| knowledge_summaries | `INT NOT NULL DEFAULT 1` + 索引 | 强制 `entity_id == 当前账套` | generate 时从当前账套写入 | +| knowledge_articles | `INT NULL`(NULL=全局共享)+ 索引 | `entity_id == 当前账套 OR entity_id IS NULL`(全局∪专属) | seed/import 默认 NULL=全局;未来企业专属文章带 entity_id | + +articles 的 OR NULL 是**有意为之**:教学知识留在共享层不被隔离吞掉,企业专属知识进抽屉严格隔离。越权核心验证对象是 events/summaries(企业记忆抽屉互不可见)+ articles 企业专属行互不可见。 + +### 2.2 归属与隔离验证(对应任务验收"越权测试") + +- entity 1(酣客)上下文查不到 entity 2(博海)的 events/summaries → 404/空集 +- entity 1 查不到 entity 2 的专属 article → 404(不暴露存在性) +- 全局 article(NULL)两账套均可见 → 教学知识不丢 +- 双实体实测记录见验收文档 + +## 三、代码落地清单(评审通过后执行) + +1. **models/knowledge.py**:KnowledgeEvent.entity_id / KnowledgeSummary.entity_id 加 `default=1, index=True` +2. **models/knowledge_article.py**:KnowledgeArticle.entity_id 加 `nullable=True, index=True`(NULL=全局共享) +3. **生产 DB ALTER**(MySQL,空表/小表零风险): + ```sql + ALTER TABLE knowledge_events ADD COLUMN entity_id INT NOT NULL DEFAULT 1 COMMENT '企业ID (多租户隔离 2026-09-03)', ADD INDEX ix_knowledge_events_entity_id (entity_id); + ALTER TABLE knowledge_summaries ADD COLUMN entity_id INT NOT NULL DEFAULT 1 COMMENT '企业ID (多租户隔离 2026-09-03)', ADD INDEX ix_knowledge_summaries_entity_id (entity_id); + ALTER TABLE knowledge_articles ADD COLUMN entity_id INT NULL DEFAULT NULL COMMENT '企业ID: NULL=全局共享, 有值=企业专属 (多租户隔离 2026-09-03)', ADD INDEX ix_knowledge_articles_entity_id (entity_id); + ``` +4. **services/knowledge_service.py** 全链路: + - `extract_events(db, since, until, entity_id)`:KPIAlert 直接按 entity_id 过滤;OperationLog 无 entity_id 列 → 经 user_id→user_entities 子查询过滤(user_id NULL 的日志跳过,标注临时方案) + - `_kpi_snapshot(db, entity_id)`:KPIDefinition 加 entity 过滤(现状漏企业,跨企业 KPI 会被喂进对方摘要 prompt——真实泄漏点) + - `generate_summary(..., entity_id)`:抽取+快照传 entity_id;KnowledgeSummary 写入 entity_id + - `get_last_summary(db, level, entity_id)` / `run_daily|weekly|monthly_summary(db, entity_id)` / `generate_*_sync(db, entity_id)` +5. **api/knowledge.py**:8 个业务端点(summaries 列表/latest/详情、events、generate daily/weekly/monthly)加 `entity_id: int = Depends(get_entity_id)` 并强制过滤;get_summary 详情校验归属(非本企业 → 404)。ethics-quiz 为静态公共题库不动。 +6. **api/knowledge_articles.py**:list/get 加 Depends(get_entity_id);过滤 `or_(entity_id == cur, entity_id.is_(None))`;get_article 企业专属行非本企业 → 404 +7. **测试**:新增 test_security_knowledge_partition.py(双实体互查越权 + 全局文章双可见 + generate 自动带 entity_id,monkeypatch DeepSeek 免网络);pytest 全量回归 +8. 前端零改动(token-first:登录用户 token 已绑 entity,拦截器不显式传 entity_id 时后端取 token 绑定值) + +## 四、验证门禁 + +1. schema_check.py ORM↔DB 一致(加列后) +2. pytest 全量无新增失败(对照基线 726 passed / 4 failed(日历漂移) / 1 xfailed) +3. 双实体越权实测脚本(生产 E2E:登 entity1 查不到 entity2 摘要/文章、entity2 专属行 404、全局文章两可见) +4. 部署冒烟 5 项 + +## 五、评审记录 + +(待项目Bot/任总复核后回填结论;若对 articles 归属语义有异议,仅需 UPDATE 存量行,代码逻辑不变) + +--- +产出物路径:本文档将随代码提交至 cma 仓库 docs/ + 归档 fullstack _reports/