test: 내부 챗봇 사용자별 Knowledge 권한 및 비노출 검증

apps/gateway/tests/api/test_deployment_permissions.py

@@ -484,6 +484,51 @@ async def run_service(**kwargs):
     assert captured["request_id"] == "req-1"
 
 
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("user_id", str(uuid.uuid4())),
+        ("organization_id", str(uuid.uuid4())),
+        (
+            "execution_subject",
+            {"type": "user", "id": str(uuid.uuid4())},
+        ),
+    ],
+)
+def test_authenticated_run_rejects_top_level_execution_context_override(
+    monkeypatch,
+    field,
+    value,
+):
+    current_user = SimpleNamespace(id=uuid.uuid4())
+    test_app = FastAPI()
+    test_app.include_router(deployment_endpoint.router, prefix="/deployments")
+    test_app.dependency_overrides[deployment_endpoint.get_db] = lambda: object()
+    test_app.dependency_overrides[deployment_endpoint.get_current_user] = (
+        lambda: current_user
+    )
+    test_app.dependency_overrides[get_deployment_runtime_policy] = (
+        lambda: DEFAULT_DEPLOYMENT_RUNTIME_POLICY
+    )
+
+    async def fail_run_service(**_kwargs):
+        raise AssertionError("invalid request must not reach deployment execution")
+
+    monkeypatch.setattr(
+        deployment_endpoint.DeploymentService,
+        "run_authenticated_deployment",
+        fail_run_service,
+    )
+
+    response = TestClient(test_app).post(
+        f"/deployments/{uuid.uuid4()}/run",
+        json={"inputs": {"question": "safe"}, field: value},
+    )
+
+    assert response.status_code == 422
+    assert response.json()["detail"][0]["type"] == "extra_forbidden"
+
+
 def test_run_authenticated_deployment_forwards_middleware_request_id(
     monkeypatch,
 ):

apps/gateway/tests/services/test_chatbot_deployment_run.py

@@ -315,6 +315,8 @@ def test_authenticated_run_uses_current_user_execution_subject(
     app_row, deployment_row = _deployed_app(deployment_type)
     current_user_id = uuid4()
     client_conversation_id = str(uuid4())
+    forged_user_id = str(uuid4())
+    forged_organization_id = str(uuid4())
     db = _Db(rows=[app_row, deployment_row])
     budget_calls = []
     evaluated_surfaces = []
@@ -341,7 +343,12 @@ def capture_budget_call(db, **kwargs):
         db,
         deployment_row.id,
         current_user_id,
-        {"question": "안녕"},
+        {
+            "question": "안녕",
+            "user_id": forged_user_id,
+            "organization_id": forged_organization_id,
+            "execution_subject": {"type": "user", "id": forged_user_id},
+        },
         monkeypatch,
         client_conversation_id=client_conversation_id,
     )
@@ -357,7 +364,13 @@ def capture_budget_call(db, **kwargs):
     assert ctx["memory_mode"] is True
     assert ctx["conversation_id"].startswith("auth:v1:")
     assert client_conversation_id not in ctx["conversation_id"]
-    assert sent_inputs == {"question": "안녕"}
+    assert sent_inputs == {
+        "question": "안녕",
+        "user_id": forged_user_id,
+        "organization_id": forged_organization_id,
+        "execution_subject": {"type": "user", "id": forged_user_id},
+    }
+    assert ctx["organization_id"] == str(app_row.organization_id)
     assert budget_calls == [
         {
             "workflow_id": app_row.workflow_id,