"""User-isolation + table-isolation tests for the MAF roundtable experiment (MAF-01-07). These tests focus on the security boundary: a non-owner cannot read or mutate another user's session / run / action / lease / approval, and the nine MAF tables have no foreign-key relationship to the legacy ``roundtable_*`` tables (the experiment is fully isolated —see implementation plan §2.3 "不改原有的工程定义). """ from __future__ import annotations import pytest import pytest_asyncio from sqlalchemy import inspect from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.maf_roundtable.service import MafNotFoundError from deerflow.persistence.base import Base from deerflow.persistence.maf_roundtable import make_maf_roundtable_store from deerflow.persistence.maf_roundtable.model import ( MafRoundtableActionLeaseRow, MafRoundtableActionRow, MafRoundtableActionSecretRow, MafRoundtableApprovalRow, MafRoundtableConversationEventRow, MafRoundtableConversationRow, MafRoundtableEventRow, MafRoundtableMessageRow, MafRoundtableParticipantRow, MafRoundtableRunRow, MafRoundtableSeatRunRow, MafRoundtableSessionRow, ) @pytest_asyncio.fixture async def store(tmp_path): engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'maf_iso.db'}") async with engine.begin() as conn: import deerflow.persistence.maf_roundtable.model # noqa: F401 await conn.run_sync(Base.metadata.create_all) sf = async_sessionmaker(engine, expire_on_commit=False) yield make_maf_roundtable_store(sf), engine await engine.dispose() def _svc(store, *, enabled=True, admin_diagnostic_access=True): from maf_test_helpers import make_maf_service return make_maf_service(store, enabled=enabled, admin_diagnostic_access=admin_diagnostic_access) # ── user isolation: non-owner cannot read ──────────────────────────────── @pytest.mark.asyncio async def test_non_owner_cannot_read_session(store): svc, _ = store s = await svc.sessions.create_session({"user_id": "u1", "task_text": "secret"}) fetched = await svc.sessions.get_session(s["id"], user_id="u2") assert fetched is None # owner scope returns None, not the row @pytest.mark.asyncio async def test_non_owner_cannot_update_session(store): svc, _ = store s = await svc.sessions.create_session({"user_id": "u1", "task_text": "t"}) updated = await svc.sessions.update_session(s["id"], user_id="u2", fields={"title": "hijack"}) assert updated is None refreshed = await svc.sessions.get_session(s["id"], user_id="u1") assert refreshed["title"] == "" # unchanged @pytest.mark.asyncio async def test_non_owner_cannot_delete_session(store): svc, _ = store s = await svc.sessions.create_session({"user_id": "u1", "task_text": "t"}) ok = await svc.sessions.delete_session(s["id"], user_id="u2") assert ok is False refreshed = await svc.sessions.get_session(s["id"], user_id="u1") assert refreshed is not None # still there @pytest.mark.asyncio async def test_service_non_owner_get_raises_not_found(store): """At the service layer, an out-of-scope read surfaces as 404 (not 403) so the existence of another user's session is not leaked (MAF-01-07).""" svc = _svc(store[0]) s = await svc.create_session(user_id="u1", body={"task_text": "t"}) with pytest.raises(MafNotFoundError): await svc.get_session(session_id=s["id"], user_id="u2") @pytest.mark.asyncio async def test_non_owner_cannot_read_run_via_service(store): svc = _svc(store[0]) s = await svc.create_session(user_id="u1", body={"task_text": "t"}) run = await svc.create_run(session_id=s["id"], user_id="u1") with pytest.raises(MafNotFoundError): await svc.get_run(run_id=run["id"], user_id="u2") @pytest.mark.asyncio async def test_non_owner_cannot_append_event(store): svc = _svc(store[0]) s = await svc.create_session(user_id="u1", body={"task_text": "t"}) run = await svc.create_run(session_id=s["id"], user_id="u1") with pytest.raises(MafNotFoundError): await svc.append_event(run_id=run["id"], user_id="u2", event_type="run.started") @pytest.mark.asyncio async def test_non_owner_cannot_read_action(store): svc = _svc(store[0]) s = await svc.create_session(user_id="u1", body={"task_text": "t"}) run = await svc.create_run(session_id=s["id"], user_id="u1") action = await store[0].actions.create_action( { "session_id": s["id"], "run_id": run["id"], "tool_name": "update_task", "idempotency_key": "k1", "status": "pending", } ) with pytest.raises(MafNotFoundError): await svc.get_action(action_id=action["id"], user_id="u2") # ── admin diagnostic access ─────────────────────────────────────────────── @pytest.mark.asyncio async def test_admin_diagnostic_access_bypasses_owner_filter(store): svc = _svc(store[0], admin_diagnostic_access=True) s = await svc.create_session(user_id="u1", body={"task_text": "t"}) fetched = await svc.get_session(session_id=s["id"], user_id="admin-1", is_admin=True) assert fetched["id"] == s["id"] rows, total = await svc.list_sessions(user_id="admin-1", is_admin=True) assert total >= 1 @pytest.mark.asyncio async def test_admin_diagnostic_access_off_keeps_owner_scope(store): svc = _svc(store[0], admin_diagnostic_access=False) s = await svc.create_session(user_id="u1", body={"task_text": "t"}) with pytest.raises(MafNotFoundError): await svc.get_session(session_id=s["id"], user_id="admin-1", is_admin=True) # ── table isolation: no FK to legacy roundtable_* ───────────────────────── @pytest.mark.asyncio async def test_maf_tables_isolated_from_legacy_roundtable(store): """The nine MAF tables must exist and none of their FKs may target a legacy roundtable_* table (MAF-01-02 / 退出标净 MAF 表和旀roundtable 表完全无关联外键).""" _, engine = store async with engine.connect() as conn: table_names = await conn.run_sync(lambda sync_conn: set(inspect(sync_conn).get_table_names())) maf_tables = {n for n in table_names if n.startswith("maf_roundtable_")} assert maf_tables == { "maf_roundtable_sessions", "maf_roundtable_participants", "maf_roundtable_runs", "maf_roundtable_events", "maf_roundtable_actions", "maf_roundtable_action_leases", "maf_roundtable_approvals", "maf_roundtable_seat_runs", "maf_roundtable_action_secrets", "maf_roundtable_conversations", "maf_roundtable_messages", "maf_roundtable_conversation_events", } # Every FK defined on an MAF model must reference another maf_roundtable_* table. maf_models = [ MafRoundtableSessionRow, MafRoundtableParticipantRow, MafRoundtableRunRow, MafRoundtableEventRow, MafRoundtableActionRow, MafRoundtableActionLeaseRow, MafRoundtableActionSecretRow, MafRoundtableApprovalRow, MafRoundtableSeatRunRow, MafRoundtableConversationRow, MafRoundtableMessageRow, MafRoundtableConversationEventRow, ] for model in maf_models: for fk in model.__table__.foreign_keys: target_table = fk.column.table.name assert target_table.startswith("maf_roundtable_"), ( f"{model.__tablename__} FK targets non-MAF table {target_table}" ) # ── feature flag (MAF-01-04) ────────────────────────────────────────────── @pytest.mark.asyncio async def test_disabled_feature_rejects_writes(store): from app.maf_roundtable.service import MafDisabledError svc = _svc(store[0], enabled=False) with pytest.raises(MafDisabledError): await svc.create_session(user_id="u1", body={"task_text": "t"}) @pytest.mark.asyncio async def test_disabled_feature_keeps_reads_open(store): from app.maf_roundtable.service import MafDisabledError svc_on = _svc(store[0], enabled=True) s = await svc_on.create_session(user_id="u1", body={"task_text": "t"}) svc_off = _svc(store[0], enabled=False) # Reads do not raise —admin can review history while the feature is off. fetched = await svc_off.get_session(session_id=s["id"], user_id="u1") assert fetched["id"] == s["id"] # But a write through the disabled service still raises. with pytest.raises(MafDisabledError): await svc_off.update_session(session_id=s["id"], user_id="u1", fields={"title": "x"})