fix(api-application): enforce processor allowed operation intersection and remove privilege escalation (LANG-224)

Co-authored-by: multica-agent <[email protected]>

src/langbot/pkg/api/application/granted_operations.py

@@ -11,7 +11,7 @@ from typing import Any, Literal
 from pydantic import TypeAdapter
 
 from ..authz import has_permission
-from ..context import PrincipalType, RequestContext
+from ..context import RequestContext
 from ..secrets import redact_secrets
 from .contracts import Operation
 from .errors import error_from_exception
@@ -141,18 +141,25 @@ def list_granted(
         return granted_ops
 
     # consumer == 'processor'
-    candidate_ids = granted
-    if candidate_ids is None and getattr(context, 'allowed_operation_ids', None) is not None:
-        candidate_ids = context.allowed_operation_ids
+    ctx_allowed = getattr(context, 'allowed_operation_ids', None)
+    ctx_allowed_set = frozenset(ctx_allowed) if ctx_allowed is not None else frozenset()
+
+    if granted is None:
+        candidate_ids = list(ctx_allowed) if ctx_allowed is not None else []
+    else:
+        candidate_ids = list(granted)
 
     seen: set[str] = set()
     granted_ops = []
     discarded: list[tuple[str, str]] = []
 
-    for op_id in candidate_ids or ():
+    for op_id in candidate_ids:
         if op_id in seen:
             continue
         seen.add(op_id)
+        if op_id not in ctx_allowed_set:
+            discarded.append((op_id, 'not_in_context_allowed_operation_ids'))
+            continue
         spec = operations_map.get(op_id)
         if spec is None:
             discarded.append((op_id, 'not_found'))
@@ -229,7 +236,7 @@ def _collect_input_values(data: Any) -> list[str]:
             values.extend(_collect_input_values(v))
     elif isinstance(data, (str, int, float)) and data is not None and data is not True and data is not False:
         s = str(data).strip()
-        if len(s) > 0:
+        if len(s) >= 4:
             values.append(s)
     return values
 
@@ -264,10 +271,12 @@ def _is_operation_granted(
     # consumer == 'processor'
     if doc.exposure != 'all':
         return False
-    candidates = granted
-    if candidates is None and getattr(context, 'allowed_operation_ids', None) is not None:
-        candidates = context.allowed_operation_ids
-    return candidates is not None and spec.operation_id in candidates
+    ctx_allowed = getattr(context, 'allowed_operation_ids', None)
+    if ctx_allowed is None or spec.operation_id not in ctx_allowed:
+        return False
+    if granted is not None and spec.operation_id not in set(granted):
+        return False
+    return True
 
 
 async def call(
@@ -296,21 +305,8 @@ async def call(
     clean_arguments = {k: v for k, v in (arguments or {}).items() if k != 'include_secret'}
     doc = get_operation_doc(spec.operation_id, side_effect=spec.side_effect)
 
-    effective_context = context
-    if context.principal.principal_type == PrincipalType.PROCESSOR:
-        perm_val = spec.permission.value if hasattr(spec.permission, 'value') else str(spec.permission)
-        curr_allowed = context.allowed_operation_ids or frozenset()
-        curr_perms = context.workspace.permissions or frozenset()
-        if operation_id not in curr_allowed or perm_val not in curr_perms:
-            new_ws = dataclasses.replace(context.workspace, permissions=curr_perms | {perm_val})
-            effective_context = dataclasses.replace(
-                context,
-                workspace=new_ws,
-                allowed_operation_ids=curr_allowed | {operation_id},
-            )
-
     try:
-        raw_result = await app_api.execute(operation_id, effective_context, clean_arguments)
+        raw_result = await app_api.execute(operation_id, context, clean_arguments)
         return {
             'ok': True,
             'result': redact_secrets(raw_result),

tests/integration/api/test_granted_operations_processor_sweep.py

@@ -11,8 +11,7 @@ import pytest
 from langbot.pkg.api.application.api import ApplicationAPI
 from langbot.pkg.api.application.granted_operations import call
 from langbot.pkg.api.application.operation_docs import OPERATION_DOCS, get_operation_doc
-from langbot.pkg.api.authz import Permission
-from langbot.pkg.api.context import PrincipalContext, PrincipalType, RequestContext, WorkspaceContext
+from langbot.pkg.api.application.processor_context import build_processor_request_context
 
 pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
 
@@ -83,6 +82,7 @@ def sweep_app(monkeypatch):
     ap.llm_model_service.create_llm_model = AsyncMock(return_value='m-1')
     ap.llm_model_service.update_llm_model = AsyncMock(return_value=None)
     ap.llm_model_service.delete_llm_model = AsyncMock(return_value=None)
+    ap.llm_model_service.test_llm_model = AsyncMock(return_value=None)
 
     # Embedding model service
     ap.embedding_models_service.get_embedding_models = AsyncMock(return_value=[])
@@ -101,6 +101,7 @@ def sweep_app(monkeypatch):
     ap.provider_service.create_provider = AsyncMock(return_value='pr-1')
     ap.provider_service.update_provider = AsyncMock(return_value=None)
     ap.provider_service.delete_provider = AsyncMock(return_value=None)
+    ap.provider_service.scan_provider_models = AsyncMock(return_value={'models': []})
 
     # Bot service
     ap.bot_service.get_bots = AsyncMock(return_value=[])
@@ -114,6 +115,7 @@ def sweep_app(monkeypatch):
     ap.bot_service.create_bot = AsyncMock(return_value='b-1')
     ap.bot_service.update_bot = AsyncMock(return_value=None)
     ap.bot_service.delete_bot = AsyncMock(return_value=None)
+    ap.bot_service.send_message = AsyncMock(return_value=None)
 
     # Agent service
     ap.agent_service.get_agents = AsyncMock(return_value=[])
@@ -124,6 +126,7 @@ def sweep_app(monkeypatch):
     ap.agent_service.create_agent = AsyncMock(return_value={'agent_uuid': 'a-1'})
     ap.agent_service.update_agent = AsyncMock(return_value=None)
     ap.agent_service.delete_agent = AsyncMock(return_value=None)
+    ap.agent_service.debug_agent = AsyncMock(return_value={'result': 'ok'})
 
     # Knowledge service
     ap.knowledge_service.get_knowledge_bases = AsyncMock(return_value=[])
@@ -137,6 +140,7 @@ def sweep_app(monkeypatch):
     ap.knowledge_service.create_knowledge_base = AsyncMock(return_value='kb-1')
     ap.knowledge_service.update_knowledge_base = AsyncMock(return_value=None)
     ap.knowledge_service.delete_knowledge_base = AsyncMock(return_value=None)
+    ap.knowledge_service.store_file = AsyncMock(return_value=1)
 
     # MCP service
     ap.mcp_service.get_mcp_servers = AsyncMock(return_value=[])
@@ -158,6 +162,7 @@ def sweep_app(monkeypatch):
     ap.skill_service.create_skill = AsyncMock(return_value={})
     ap.skill_service.update_skill = AsyncMock(return_value={})
     ap.skill_service.delete_skill = AsyncMock(return_value=True)
+    ap.skill_service.write_skill_file = AsyncMock(return_value={'revision': 'rev-1'})
 
     # Monitoring service
     ap.monitoring_service.get_overview_metrics = AsyncMock(return_value={})
@@ -206,28 +211,6 @@ def sweep_app(monkeypatch):
     return ap
 
 
-def make_processor_context(allowed_ids: frozenset[str]) -> RequestContext:
-    return RequestContext(
-        instance_uuid='sweep-instance',
-        placement_generation=1,
-        request_id='req-proc-sweep',
-        auth_type='processor',
-        principal=PrincipalContext(
-            principal_type=PrincipalType.PROCESSOR,
-            processor_kind='agent',
-            processor_uuid='agent-sweep',
-            run_id='run-sweep',
-        ),
-        workspace=WorkspaceContext(
-            workspace_uuid='ws-sweep',
-            membership_uuid=None,
-            role=None,
-            permissions=frozenset({p.value for p in Permission}),
-        ),
-        allowed_operation_ids=allowed_ids,
-    )
-
-
 MIN_READ_PARAMS = {
     'adapter.get': {'adapter_name': 'aiocqhttp'},
     'agent.get': {'agent_uuid': 'a-1'},
@@ -293,7 +276,15 @@ async def test_processor_sweep_all_in_scope_read_operations(sweep_app):
     ]
     assert len(read_op_ids) == 85
 
-    proc_ctx = make_processor_context(frozenset(read_op_ids))
+    proc_ctx = await build_processor_request_context(
+        sweep_app,
+        workspace_uuid='ws-sweep',
+        processor_kind='agent',
+        processor_uuid='agent-sweep',
+        run_id='run-sweep',
+        operation_ids=read_op_ids,
+    )
+    assert proc_ctx.allowed_operation_ids == frozenset(read_op_ids)
 
     with patch('langbot.pkg.api.application.plugin_operations._github_json', AsyncMock(return_value={'assets': []})):
         for op_id in read_op_ids:
@@ -309,14 +300,17 @@ WRITE_SAMPLES = [
     ('agent.create', {'agent_data': {'name': 'A1'}}),
     ('agent.update', {'agent_uuid': 'a-1', 'agent_data': {'name': 'A2'}}),
     ('agent.delete', {'agent_uuid': 'a-1'}),
+    ('agent.debug', {'agent_uuid': 'a-1', 'payload': {}}),
     # bot
     ('bot.create', {'bot_data': {'name': 'B1'}}),
     ('bot.update', {'bot_uuid': 'b-1', 'bot_data': {'name': 'B2'}}),
     ('bot.delete', {'bot_uuid': 'b-1'}),
+    ('bot.message.send', {'bot_uuid': 'b-1', 'target_type': 'group', 'target_id': 'g-1', 'message_chain_data': {}}),
     # knowledge
     ('knowledge.base.create', {'kb_data': {'name': 'K1'}}),
     ('knowledge.base.update', {'kb_uuid': 'kb-1', 'kb_data': {'name': 'K2'}}),
     ('knowledge.base.delete', {'kb_uuid': 'kb-1'}),
+    ('knowledge.file.store', {'kb_uuid': 'kb-1', 'file_id': 'f-1'}),
     # mcp (delete and test are exposure=='all')
     ('mcp.server.test', {'server_name': 's-1', 'server_data': {}}),
     ('mcp.server.delete', {'server_uuid': 's-1'}),
@@ -324,40 +318,67 @@ WRITE_SAMPLES = [
     ('model.llm.create', {'model_data': {'name': 'M1'}}),
     ('model.llm.update', {'model_uuid': 'm-1', 'model_data': {'name': 'M2'}}),
     ('model.llm.delete', {'model_uuid': 'm-1'}),
+    ('model.llm.test', {'model_uuid': 'm-1', 'model_data': {}}),
     # monitoring
     ('monitoring.session.reset', {'bot_id': 'b-1', 'session_id': 's-1'}),
     # pipeline
     ('pipeline.create', {'pipeline_data': {'name': 'P1'}}),
     ('pipeline.update', {'pipeline_uuid': 'p-1', 'pipeline_data': {'name': 'P2'}}),
     ('pipeline.delete', {'pipeline_uuid': 'p-1'}),
+    ('pipeline.run', {'pipeline_uuid': 'p-1', 'text': 'test pipeline run'}),
+    ('pipeline.chat.broadcast', {'pipeline_uuid': 'p-1', 'message': 'test broadcast'}),
     # provider
     ('provider.create', {'provider_data': {'name': 'PR1'}}),
     ('provider.update', {'provider_uuid': 'pr-1', 'provider_data': {'name': 'PR2'}}),
     ('provider.delete', {'provider_uuid': 'pr-1'}),
+    ('provider.models.scan', {'provider_uuid': 'pr-1'}),
     # skill
     ('skill.create', {'data': {}}),
     ('skill.update', {'skill_name': 'sk-1', 'data': {}}),
     ('skill.delete', {'skill_name': 'sk-1'}),
+    ('skill.file.write', {'skill_name': 'sk-1', 'path': 'test.py', 'content': 'print(1)', 'base_revision': None}),
 ]
 
 
 async def test_processor_sweep_sampled_write_operations(sweep_app):
-    """Sample create, update, delete write operations across all domains for processor."""
+    """Sample write operations across all domains for processor with minimal derived permissions."""
     write_op_ids = [op[0] for op in WRITE_SAMPLES]
-    assert len(write_op_ids) == 24
+    assert len(write_op_ids) == 32
 
-    proc_ctx = make_processor_context(frozenset(write_op_ids))
+    proc_ctx = await build_processor_request_context(
+        sweep_app,
+        workspace_uuid='ws-sweep',
+        processor_kind='agent',
+        processor_uuid='agent-sweep',
+        run_id='run-sweep',
+        operation_ids=write_op_ids,
+    )
+    assert proc_ctx.allowed_operation_ids == frozenset(write_op_ids)
 
-    for op_id, args in WRITE_SAMPLES:
-        result = await call(sweep_app, proc_ctx, op_id, args, consumer='processor', granted=write_op_ids)
-        assert result['ok'] is True, f'Write operation {op_id} failed: {result.get("error")}'
-        assert result['operation']['id'] == op_id
-        assert result['operation']['side_effect'] is True
+    with (
+        patch('langbot.pkg.api.http.service.pipeline_run.run_pipeline', AsyncMock(return_value={'status': 'ok'})),
+        patch(
+            'langbot.pkg.platform.sources.websocket_manager.ws_connection_manager.broadcast_to_pipeline',
+            AsyncMock(return_value=None),
+        ),
+    ):
+        for op_id, args in WRITE_SAMPLES:
+            result = await call(sweep_app, proc_ctx, op_id, args, consumer='processor', granted=write_op_ids)
+            assert result['ok'] is True, f'Write operation {op_id} failed: {result.get("error")}'
+            assert result['operation']['id'] == op_id
+            assert result['operation']['side_effect'] is True
 
 
 async def test_processor_cannot_call_assistant_only_operation(sweep_app):
     """Verify processor is denied from calling assistant-only operations."""
-    proc_ctx = make_processor_context(frozenset({'mcp.server.create', 'bot.list'}))
+    proc_ctx = await build_processor_request_context(
+        sweep_app,
+        workspace_uuid='ws-sweep',
+        processor_kind='agent',
+        processor_uuid='agent-sweep',
+        run_id='run-sweep',
+        operation_ids=['mcp.server.create', 'bot.list'],
+    )
 
     # mcp.server.create is exposure='assistant', not 'all'
     result = await call(
@@ -370,3 +391,4 @@ async def test_processor_cannot_call_assistant_only_operation(sweep_app):
     )
     assert result['ok'] is False
     assert result['error']['code'] == 'not_found'
+    assert result['error']['code'] == 'not_found'

tests/unit_tests/api/test_granted_operations.py

@@ -133,8 +133,8 @@ def test_processor_filtering_rules(mock_host, caplog):
     )
 
     requested_granted = [
-        'bot.list',  # exposure == 'all' -> kept
-        'agent.create',  # exposure == 'all' -> kept
+        'bot.list',  # exposure == 'all' and in allowed_operation_ids -> kept
+        'agent.create',  # exposure == 'all' but NOT in allowed_operation_ids -> discarded
         'mcp.server.create',  # exposure == 'assistant' -> discarded
         'pipeline.migration.preview',  # exposure == 'assistant' -> discarded
         'agent.debug.stream',  # streaming -> discarded
@@ -147,14 +147,94 @@ def test_processor_filtering_rules(mock_host, caplog):
         granted_ops = list_granted(mock_host, processor_context, consumer='processor', granted=requested_granted)
 
     granted_ids = [op.id for op in granted_ops]
-    assert granted_ids == ['bot.list', 'agent.create']
+    assert granted_ids == ['bot.list']
 
     # Verify warning logged for discarded operations
     assert 'Discarded' in caplog.text
+    assert 'agent.create' in caplog.text
     assert 'mcp.server.create' in caplog.text
     assert 'nonexistent.operation.id' in caplog.text
 
 
+def test_processor_granted_exceeding_context_is_discarded_with_warning(mock_host, caplog):
+    api = ApplicationAPI(mock_host)
+    mock_host.application_api = api
+
+    processor_context = RequestContext(
+        instance_uuid='test-instance',
+        placement_generation=1,
+        request_id='req-proc',
+        auth_type='processor',
+        principal=PrincipalContext(
+            principal_type=PrincipalType.PROCESSOR,
+            processor_kind='agent',
+            processor_uuid='agent-1',
+            run_id='run-1',
+        ),
+        workspace=WorkspaceContext(
+            workspace_uuid='ws-1',
+            membership_uuid=None,
+            role=None,
+            permissions=frozenset({Permission.RESOURCE_VIEW.value}),
+        ),
+        allowed_operation_ids=frozenset({'bot.list'}),
+    )
+
+    import logging
+
+    with caplog.at_level(logging.WARNING):
+        ops = list_granted(
+            mock_host,
+            processor_context,
+            consumer='processor',
+            granted=['bot.list', 'bot.delete', 'agent.create'],
+        )
+
+    assert [op.id for op in ops] == ['bot.list']
+    assert 'Discarded 2 invalid operation(s) for processor' in caplog.text
+    assert 'bot.delete' in caplog.text
+    assert 'agent.create' in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_processor_call_outside_context_allowed_operation_ids_returns_not_found(mock_host):
+    api = ApplicationAPI(mock_host)
+    mock_host.application_api = api
+
+    processor_context = RequestContext(
+        instance_uuid='test-instance',
+        placement_generation=1,
+        request_id='req-proc',
+        auth_type='processor',
+        principal=PrincipalContext(
+            principal_type=PrincipalType.PROCESSOR,
+            processor_kind='agent',
+            processor_uuid='agent-1',
+            run_id='run-1',
+        ),
+        workspace=WorkspaceContext(
+            workspace_uuid='ws-1',
+            membership_uuid=None,
+            role=None,
+            permissions=frozenset({Permission.RESOURCE_VIEW.value}),
+        ),
+        allowed_operation_ids=frozenset({'bot.list'}),
+    )
+
+    # Calling bot.delete with granted=['bot.delete'] must NOT elevate permissions or allow execution
+    res = await call(
+        mock_host,
+        processor_context,
+        'bot.delete',
+        {'bot_uuid': 'bot-1'},
+        consumer='processor',
+        granted=['bot.delete'],
+    )
+    assert res['ok'] is False
+    assert res['error']['code'] == 'not_found'
+    assert 'Operation not found' in res['error']['message']
+
+
 def test_processor_fallback_to_context_allowed_operation_ids(mock_host):
     api = ApplicationAPI(mock_host)
     mock_host.application_api = api
@@ -427,3 +507,15 @@ def test_granted_operation_serialization():
     assert op['id'] == 'test.op'
     assert op.get('domain') == 'test'
     assert dict(op)['permission'] == 'resource.view'
+
+
+def test_error_message_sanitization_does_not_corrupt_short_inputs():
+    from langbot.pkg.api.application.granted_operations import _sanitize_error_message
+
+    msg = 'Application resource 1 failed on shard a'
+    sanitized = _sanitize_error_message(msg, {'id': '1', 'shard': 'a', 'count': 2})
+    assert sanitized == 'Application resource 1 failed on shard a'
+
+    msg_with_secret = 'Application resource secret_token_xyz failed on shard a'
+    sanitized_secret = _sanitize_error_message(msg_with_secret, {'token': 'secret_token_xyz', 'id': '1'})
+    assert sanitized_secret == 'Application resource [REDACTED] failed on shard a'