From 23f4abb65b5ca8b53204b1d6df1313d3564193b9 Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <60681390+huanghuoguoguo@users.noreply.github.com> Date: Thu, 29 Oct 2026 15:26:56 +0800 Subject: [PATCH] fix(api): preserve scoped writes, streaming readiness and audit contracts --- dev-docs/architecture/application-api.md | 5 +- .../pkg/api/application/service_operations.py | 98 +++++++++++++ src/langbot/pkg/api/application/transport.py | 1 + .../pkg/api/application/workspace_streams.py | 2 + src/langbot/pkg/api/http/controller/group.py | 10 +- .../api/http/controller/groups/assistant.py | 11 +- .../http/controller/groups/knowledge/base.py | 13 +- .../controller/groups/pipelines/pipelines.py | 49 +------ .../http/controller/groups/provider/models.py | 11 +- .../controller/groups/provider/providers.py | 13 +- .../http/controller/groups/resources/mcp.py | 21 ++- .../pkg/api/http/controller/groups/skills.py | 11 +- tests/integration/api/test_mcp_write_scope.py | 102 +++++++++++++ .../api/test_resource_write_audit.py | 134 ++++++++++++++++++ tests/integration/api/test_workspaces.py | 32 +++++ .../api/test_application_workspace_streams.py | 1 + .../api/test_management_assistant.py | 66 +++++++++ 17 files changed, 468 insertions(+), 112 deletions(-) create mode 100644 tests/integration/api/test_mcp_write_scope.py create mode 100644 tests/integration/api/test_resource_write_audit.py diff --git a/dev-docs/architecture/application-api.md b/dev-docs/architecture/application-api.md index d303b297a..0317d8a91 100644 --- a/dev-docs/architecture/application-api.md +++ b/dev-docs/architecture/application-api.md @@ -71,7 +71,10 @@ that execution failed or was rolled back. Streaming scopes live in isolated producer tasks with bounded queues. Closing an Agent debug stream cancels its execution. Management-assistant streaming retains its existing detached-run behavior; use the explicit stop operation to -cancel an assistant run. Consumers should close partially consumed iterators +cancel an assistant run. `assistant.stream` first emits +`{"kind": "started", "data": {"conversation_id": "..."}}` after admission and +before resolving a model. HTTP consumes this frame internally, returns headers +immediately, and keeps its existing business frames and 15-second heartbeats. Consumers should close partially consumed iterators with `contextlib.aclosing`. ## Execution-context operations diff --git a/src/langbot/pkg/api/application/service_operations.py b/src/langbot/pkg/api/application/service_operations.py index a77f3e3a4..61cfd6931 100644 --- a/src/langbot/pkg/api/application/service_operations.py +++ b/src/langbot/pkg/api/application/service_operations.py @@ -802,6 +802,49 @@ def _contract(operation_id, method): return model, output +_UPDATE_AUDITS = { + 'pipeline.update': ('get_pipeline', 'pipeline_uuid', 'pipeline_data', False), + 'pipeline.extensions.update': ('get_pipeline', 'pipeline_uuid', None, False), + 'model.llm.update': ('get_llm_model', 'model_uuid', 'model_data', False), + 'provider.update': ('get_provider', 'provider_uuid', 'provider_data', True), + 'knowledge.base.update': ('get_knowledge_base', 'kb_uuid', 'kb_data', True), + 'skill.update': ('get_skill', 'skill_name', 'data', False), +} + + +async def _update_audit(service, context, operation_id, payload): + from ..http.service.secrets import redact_secrets + from ...operation_trace.service import changed_fields + from ...pipeline.extension_preferences import normalize_extension_preferences + + reader, identity, data_field, include_secret = _UPDATE_AUDITS[operation_id] + try: + previous = await getattr(service, reader)( + context, getattr(payload, identity), **({'include_secret': True} if include_secret else {}) + ) + except Exception as exc: + if operation_id == 'skill.update' and not isinstance(exc, ValueError): + raise + previous = None + before = previous if isinstance(previous, dict) else {} + ignore = ('uuid', 'created_at', 'updated_at') + if operation_id == 'pipeline.extensions.update': + before = normalize_extension_preferences(before.get('extensions_preferences')) + requested = { + field.removeprefix('bound_') if field.startswith('bound_') else field: value + for field, value in payload.model_dump().items() + if field != 'pipeline_uuid' and value is not None + } + after = {**before, **requested} + else: + after = getattr(payload, data_field) + if operation_id == 'provider.update': + ignore += ('llm_count', 'embedding_count', 'rerank_count') + elif operation_id == 'skill.update': + ignore = ('files', 'package_root', 'created_at', 'updated_at') + return changed_fields(redact_secrets(before), redact_secrets(after), ignore=ignore) + + def _operation(operation_id, service_name, method, permission, side_effect): input_type, output_type = _contract(operation_id, method) @@ -815,7 +858,15 @@ def _operation(operation_id, service_name, method, permission, side_effect): ) require_permission(context, secret_permission) service = getattr(ap, service_name) + from .transport import application_audit_details, transport_records_audit + + audit_details = application_audit_details.get() + changes = None + if transport_records_audit.get() and audit_details is not None and operation_id in _UPDATE_AUDITS: + changes = await _update_audit(service, context, operation_id, payload) result = await getattr(service, method.__name__)(context, **arguments) + if changes is not None: + audit_details['changes'] = changes if result is None and operation_id.endswith('.get') and output_type != type(None): # Optional defaults (e.g. starred model) are intentionally nullable. if operation_id not in {'model.llm.starred.get', 'model.llm.default.get'}: @@ -916,6 +967,37 @@ class MigrationInput(Input): body: dict +class MCPServerNameInput(Input): + server_name: str = Field(min_length=1) + + +class MCPServerUpdateByNameInput(MCPServerNameInput): + server_data: dict + + +async def _mcp_server_update_by_name(ap, context, payload): + from ..http.service.mcp import redact_mcp_secrets + from ...operation_trace.service import changed_fields + + previous = await ap.mcp_service.get_mcp_server_by_name(context, payload.server_name) + if previous is None: + return None + await ap.mcp_service.update_mcp_server(context, previous['uuid'], payload.server_data) + return { + 'changes': changed_fields( + redact_mcp_secrets(previous), redact_mcp_secrets(payload.server_data), ignore=('uuid',) + ) + } + + +async def _mcp_server_delete_by_name(ap, context, payload): + server = await ap.mcp_service.get_mcp_server_by_name(context, payload.server_name) + if server is None: + return None + await ap.mcp_service.delete_mcp_server(context, server['uuid']) + return {} + + def _migration_service(ap): from ..http.service.pipeline_migration import PipelineMigrationService @@ -946,6 +1028,22 @@ def _special_operations(): result = [ Operation('pipeline.run', Permission.RUNTIME_OPERATE, _pipeline_run, PipelineRunInput, dict, True), Operation('bot.routes.preview', Permission.RESOURCE_VIEW, _route_preview, RoutePreviewInput, dict), + Operation( + 'mcp.server.update_by_name', + Permission.RESOURCE_MANAGE, + _mcp_server_update_by_name, + MCPServerUpdateByNameInput, + dict | None, + True, + ), + Operation( + 'mcp.server.delete_by_name', + Permission.RESOURCE_MANAGE, + _mcp_server_delete_by_name, + MCPServerNameInput, + dict | None, + True, + ), Operation( 'pipeline.migration.preview', Permission.RESOURCE_VIEW, _migration_preview, Input, dict, principals=account ), diff --git a/src/langbot/pkg/api/application/transport.py b/src/langbot/pkg/api/application/transport.py index 86885a3c4..aa723def6 100644 --- a/src/langbot/pkg/api/application/transport.py +++ b/src/langbot/pkg/api/application/transport.py @@ -7,6 +7,7 @@ from .contracts import ApplicationAPIException # HTTP already records the request's route, resource and before/after changes. transport_records_audit = ContextVar('transport_records_application_audit', default=False) +application_audit_details = ContextVar('application_audit_details', default=None) class ApplicationInputError(ValueError): diff --git a/src/langbot/pkg/api/application/workspace_streams.py b/src/langbot/pkg/api/application/workspace_streams.py index cd8fc40c8..69cc2daa3 100644 --- a/src/langbot/pkg/api/application/workspace_streams.py +++ b/src/langbot/pkg/api/application/workspace_streams.py @@ -71,6 +71,8 @@ async def assistant_turn(ap, context, payload): queue = service.start_run(context, payload.conversation_id, payload) async def frames(): + # Acknowledge admission before model resolution or other background work. + yield {'kind': 'started', 'data': {'conversation_id': payload.conversation_id}} # start_run owns a bounded queue and the run survives a subscriber disconnect. while True: frame = await queue.get() diff --git a/src/langbot/pkg/api/http/controller/group.py b/src/langbot/pkg/api/http/controller/group.py index 523daacdb..a11dbb15a 100644 --- a/src/langbot/pkg/api/http/controller/group.py +++ b/src/langbot/pkg/api/http/controller/group.py @@ -88,12 +88,18 @@ class RouterGroup(abc.ABC): self.quart_app = quart_app async def call_application(self, operation_id, context, payload): - from ...application.transport import invoke_application, transport_records_audit + from ...application.transport import application_audit_details, invoke_application, transport_records_audit token = transport_records_audit.set(True) + details = {} + details_token = application_audit_details.set(details) try: - return await invoke_application(self.ap, operation_id, context, payload) + result = await invoke_application(self.ap, operation_id, context, payload) + if 'changes' in details: + quart.g.operation_log_changes = details['changes'] + return result finally: + application_audit_details.reset(details_token) transport_records_audit.reset(token) async def stream_application(self, operation_id, context, payload): diff --git a/src/langbot/pkg/api/http/controller/groups/assistant.py b/src/langbot/pkg/api/http/controller/groups/assistant.py index 97abf3485..dcd6eaa2d 100644 --- a/src/langbot/pkg/api/http/controller/groups/assistant.py +++ b/src/langbot/pkg/api/http/controller/groups/assistant.py @@ -120,16 +120,12 @@ class AssistantRouterGroup(group.RouterGroup): async def assistant_stream_response(source): """Keep the run alive after disconnect; only an explicit stop cancels it.""" - # Resolve startup failures before sending HTTP headers. - first = await anext(source) + # Consume the admission frame so startup errors remain HTTP errors without waiting for a model. + await anext(source) async def stream(): async with aclosing(source): - frame = first while True: - yield json.dumps(frame, ensure_ascii=False) + '\n' - if frame['kind'] in {'completed', 'error'}: - return pending = asyncio.create_task(anext(source)) try: while not (await asyncio.wait({pending}, timeout=15))[0]: @@ -141,6 +137,9 @@ async def assistant_stream_response(source): finally: pending.cancel() await asyncio.gather(pending, return_exceptions=True) + yield json.dumps(frame, ensure_ascii=False) + '\n' + if frame['kind'] in {'completed', 'error'}: + return response = quart.Response(stream(), content_type='application/x-ndjson; charset=utf-8') response.timeout = None diff --git a/src/langbot/pkg/api/http/controller/groups/knowledge/base.py b/src/langbot/pkg/api/http/controller/groups/knowledge/base.py index 0a3a33f93..9de751586 100644 --- a/src/langbot/pkg/api/http/controller/groups/knowledge/base.py +++ b/src/langbot/pkg/api/http/controller/groups/knowledge/base.py @@ -76,22 +76,11 @@ class KnowledgeBaseRouterGroup(group.RouterGroup): quart.g.operation_log_resource_id = knowledge_base_uuid if quart.request.method == 'PUT': json_data = await quart.request.json - try: - previous = await self.call_application( - 'knowledge.base.get', request_context, {'kb_uuid': knowledge_base_uuid, 'include_secret': True} - ) - except Exception: - previous = None await self.call_application( 'knowledge.base.update', request_context, {'kb_uuid': knowledge_base_uuid, 'kb_data': json_data} ) - changes = settings_service.changed_fields( - previous if isinstance(previous, dict) else {}, - json_data if isinstance(json_data, dict) else {}, - ignore=('uuid', 'created_at', 'updated_at'), - ) + changes = quart.g.operation_log_changes rule = settings_service.ACTION_RULES_BY_ACTION.get('knowledge_base_update') - quart.g.operation_log_changes = changes if rule is not None and changes: quart.g.operation_log_summary = settings_service.build_summary(rule, changes) return self.success(data={'uuid': knowledge_base_uuid}) diff --git a/src/langbot/pkg/api/http/controller/groups/pipelines/pipelines.py b/src/langbot/pkg/api/http/controller/groups/pipelines/pipelines.py index 340938f65..2ac9d65f0 100644 --- a/src/langbot/pkg/api/http/controller/groups/pipelines/pipelines.py +++ b/src/langbot/pkg/api/http/controller/groups/pipelines/pipelines.py @@ -8,7 +8,6 @@ from ......operation_trace import service as settings_service from ......workspace.errors import WorkspaceNotFoundError from ... import group from ......pipeline.extension_preferences import ( - normalize_extension_preferences, validate_extension_preferences, ) @@ -91,25 +90,14 @@ class PipelinesRouterGroup(group.RouterGroup): quart.g.operation_log_resource_id = pipeline_uuid if quart.request.method == 'PUT': json_data = await quart.request.json - try: - previous = await self.call_application( - 'pipeline.get', request_context, {'pipeline_uuid': pipeline_uuid} - ) - except Exception: - previous = None try: await self.call_application( 'pipeline.update', request_context, {'pipeline_uuid': pipeline_uuid, 'pipeline_data': json_data} ) except ValueError as exc: return self.http_status(400, -1, str(exc)) - changes = settings_service.changed_fields( - previous if isinstance(previous, dict) else {}, - json_data if isinstance(json_data, dict) else {}, - ignore=('uuid', 'created_at', 'updated_at'), - ) + changes = quart.g.operation_log_changes rule = settings_service.ACTION_RULES_BY_ACTION.get('update') - quart.g.operation_log_changes = changes if rule is not None and changes: quart.g.operation_log_summary = settings_service.build_summary(rule, changes) else: @@ -155,41 +143,6 @@ class PipelinesRouterGroup(group.RouterGroup): async def _(pipeline_uuid: str, request_context: RequestContext) -> str: quart.g.operation_log_resource_id = pipeline_uuid json_data = await quart.request.json - # Capture "what was bound before" so the trace answers "what changed - # into what": without this the log could only say an extension was - # configured, never which binding was added or removed. - try: - previous = await self.call_application( - 'pipeline.get', request_context, {'pipeline_uuid': pipeline_uuid} - ) - except Exception: - previous = None - # The request uses ``bound_*`` keys while the stored preferences use - # the canonical names, so both sides are normalized before diffing. - requested = { - canonical: json_data[field] - for field, canonical in { - 'bound_plugins': 'plugins', - 'bound_mcp_servers': 'mcp_servers', - 'bound_skills': 'skills', - 'bound_mcp_resources': 'mcp_resources', - 'enable_all_plugins': 'enable_all_plugins', - 'enable_all_mcp_servers': 'enable_all_mcp_servers', - 'enable_all_skills': 'enable_all_skills', - 'mcp_resource_agent_read_enabled': 'mcp_resource_agent_read_enabled', - }.items() - if field in json_data - } - if isinstance(previous, dict): - current = { - **normalize_extension_preferences(previous.get('extensions_preferences')), - **requested, - } - else: - current = requested - changes = settings_service.changed_fields(current, requested) - if changes: - quart.g.operation_log_changes = changes try: validate_extension_preferences( { diff --git a/src/langbot/pkg/api/http/controller/groups/provider/models.py b/src/langbot/pkg/api/http/controller/groups/provider/models.py index 1e80a5a55..2b4caf12e 100644 --- a/src/langbot/pkg/api/http/controller/groups/provider/models.py +++ b/src/langbot/pkg/api/http/controller/groups/provider/models.py @@ -107,23 +107,14 @@ class LLMModelsRouterGroup(group.RouterGroup): ) async def _(model_uuid: str, request_context: RequestContext) -> str: json_data = await quart.request.json - try: - previous = await self.call_application('model.llm.get', request_context, {'model_uuid': model_uuid}) - except Exception: - previous = None try: await self.call_application( 'model.llm.update', request_context, {'model_uuid': model_uuid, 'model_data': json_data} ) except ValueError as exc: return self.http_status(400, -1, str(exc)) - changes = settings_service.changed_fields( - previous if isinstance(previous, dict) else {}, - json_data if isinstance(json_data, dict) else {}, - ignore=('uuid', 'created_at', 'updated_at'), - ) + changes = quart.g.operation_log_changes rule = settings_service.ACTION_RULES_BY_ACTION.get('update') - quart.g.operation_log_changes = changes if rule is not None and changes: quart.g.operation_log_summary = settings_service.build_summary(rule, changes) return self.success() diff --git a/src/langbot/pkg/api/http/controller/groups/provider/providers.py b/src/langbot/pkg/api/http/controller/groups/provider/providers.py index b824347a8..887c3632c 100644 --- a/src/langbot/pkg/api/http/controller/groups/provider/providers.py +++ b/src/langbot/pkg/api/http/controller/groups/provider/providers.py @@ -169,12 +169,6 @@ class ModelProvidersRouterGroup(group.RouterGroup): ) async def _(provider_uuid: str, request_context: RequestContext) -> str: json_data = await quart.request.json - try: - previous = await self.call_application( - 'provider.get', request_context, {'provider_uuid': provider_uuid, 'include_secret': True} - ) - except Exception: - previous = None try: await self.call_application( 'provider.update', request_context, {'provider_uuid': provider_uuid, 'provider_data': json_data} @@ -183,13 +177,8 @@ class ModelProvidersRouterGroup(group.RouterGroup): return self.http_status(400, -1, str(exc)) # Record which provider was reconfigured and which fields moved; # credential-looking keys are redacted by ``changed_fields``. - changes = settings_service.changed_fields( - previous if isinstance(previous, dict) else {}, - json_data if isinstance(json_data, dict) else {}, - ignore=('uuid', 'created_at', 'updated_at', 'llm_count', 'embedding_count', 'rerank_count'), - ) + changes = quart.g.operation_log_changes rule = settings_service.ACTION_RULES_BY_ACTION.get('update') - quart.g.operation_log_changes = changes if rule is not None and changes: quart.g.operation_log_summary = settings_service.build_summary(rule, changes) return self.success() diff --git a/src/langbot/pkg/api/http/controller/groups/resources/mcp.py b/src/langbot/pkg/api/http/controller/groups/resources/mcp.py index efc3546e0..418dcd5e9 100644 --- a/src/langbot/pkg/api/http/controller/groups/resources/mcp.py +++ b/src/langbot/pkg/api/http/controller/groups/resources/mcp.py @@ -60,9 +60,6 @@ class MCPRouterGroup(group.RouterGroup): ) async def _(server_name: str, request_context: RequestContext) -> str: server_name = unquote(server_name) - server_data = await self.call_application('mcp.server.get', request_context, {'server_name': server_name}) - if server_data is None: - return self.http_status(404, -1, 'Server not found') # Name the MCP server in the trace, and for an update record which # fields actually moved so the card answers "what changed" rather # than only "the MCP config was touched". @@ -70,24 +67,26 @@ class MCPRouterGroup(group.RouterGroup): if quart.request.method == 'PUT': data = await quart.request.json try: - await self.call_application( - 'mcp.server.update', request_context, {'server_uuid': server_data['uuid'], 'server_data': data} + result = await self.call_application( + 'mcp.server.update_by_name', request_context, {'server_name': server_name, 'server_data': data} ) except MCPStdioDisabledError as exc: return self.http_status(403, exc.code, str(exc)) except ValueError as exc: return self.http_status(400, -1, str(exc)) - changes = settings_service.changed_fields( - server_data, - data if isinstance(data, dict) else {}, - ignore=('uuid',), - ) + if result is None: + return self.http_status(404, -1, 'Server not found') + changes = result['changes'] rule = settings_service.ACTION_RULES_BY_ACTION.get('mcp_config') quart.g.operation_log_changes = changes if rule is not None and changes: quart.g.operation_log_summary = settings_service.build_summary(rule, changes) else: - await self.call_application('mcp.server.delete', request_context, {'server_uuid': server_data['uuid']}) + result = await self.call_application( + 'mcp.server.delete_by_name', request_context, {'server_name': server_name} + ) + if result is None: + return self.http_status(404, -1, 'Server not found') return self.success() @self.route( diff --git a/src/langbot/pkg/api/http/controller/groups/skills.py b/src/langbot/pkg/api/http/controller/groups/skills.py index cb7c443ce..556cab95d 100644 --- a/src/langbot/pkg/api/http/controller/groups/skills.py +++ b/src/langbot/pkg/api/http/controller/groups/skills.py @@ -71,10 +71,6 @@ class SkillsRouterGroup(group.RouterGroup): quart.g.operation_log_resource_id = skill_name if quart.request.method == 'PUT': data = await quart.request.json - try: - previous = await self.call_application('skill.get', request_context, {'skill_name': skill_name}) - except ValueError: - previous = None try: skill = await self.call_application( 'skill.update', request_context, {'skill_name': skill_name, 'data': data} @@ -84,13 +80,8 @@ class SkillsRouterGroup(group.RouterGroup): except ValueError as exc: return self.http_status(400, -1, str(exc)) - changes = settings_service.changed_fields( - previous if isinstance(previous, dict) else {}, - data if isinstance(data, dict) else {}, - ignore=('files', 'package_root', 'created_at', 'updated_at'), - ) + changes = quart.g.operation_log_changes rule = settings_service.ACTION_RULES_BY_ACTION.get('skill_view') - quart.g.operation_log_changes = changes if rule is not None and changes: quart.g.operation_log_summary = settings_service.build_summary(rule, changes) return self.success(data={'skill': skill}) diff --git a/tests/integration/api/test_mcp_write_scope.py b/tests/integration/api/test_mcp_write_scope.py new file mode 100644 index 000000000..55de2f11c --- /dev/null +++ b/tests/integration/api/test_mcp_write_scope.py @@ -0,0 +1,102 @@ +"""MCP management keys keep their write contract without resource-view scope.""" + +from types import SimpleNamespace + +import pytest +import quart +import sqlalchemy as sa + +from langbot.pkg.api.http.controller.groups.resources.mcp import MCPRouterGroup +from langbot.pkg.api.http.service.mcp import MCPService +from langbot.pkg.entity.persistence.mcp import MCPServer +from tests.integration.api.test_application_foundation import context, foundation # noqa: F401 + + +pytestmark = [pytest.mark.integration, pytest.mark.asyncio] + + +@pytest.fixture +async def mcp_write_api(foundation): # noqa: F811 + api, ap, engine = foundation + ap.application_api = api + ap.tool_mgr = SimpleNamespace( + mcp_tool_loader=SimpleNamespace(get_session=lambda *_: None, has_session=lambda *_: False) + ) + ap.mcp_service = MCPService(ap) + keys = {} + for name, scopes in [('manager', ['resource.manage']), ('empty', [])]: + keys[name] = (await ap.apikey_service.create_api_key(context(), name, scopes=scopes))['key'] + async with engine.begin() as connection: + await connection.execute( + sa.insert(MCPServer), + [ + { + 'uuid': identity, + 'workspace_uuid': workspace, + 'name': name, + 'mode': 'remote', + 'enable': False, + 'readme': 'before', + 'extra_args': {'api_key': 'old-secret'}, + } + for identity, workspace, name in [ + ('owned-a', 'workspace-a', 'owned'), + ('owned-b', 'workspace-b', 'owned'), + ('foreign-b', 'workspace-b', 'foreign'), + ] + ], + ) + app = quart.Quart(__name__) + audit_changes = [] + + @app.after_request + async def capture_audit(response): + audit_changes.append(getattr(quart.g, 'operation_log_changes', [])) + return response + + await MCPRouterGroup(ap, app).initialize() + return app.test_client(), engine, keys, audit_changes + + +async def test_manage_only_key_updates_and_deletes_without_read_access(mcp_write_api): + client, engine, keys, audit_changes = mcp_write_api + headers = {'X-API-Key': keys['manager'], 'X-Workspace-Id': 'workspace-b'} + endpoint = '/api/v1/mcp/servers/owned' + assert (await client.get(endpoint, headers=headers)).status_code == 403 + response = await client.put( + endpoint, headers=headers, json={'readme': 'after', 'extra_args': {'api_key': 'new-secret'}} + ) + assert response.status_code == 200 + assert await response.get_json() == {'code': 0, 'msg': 'ok', 'data': None} + assert {'field': 'readme', 'before': 'before', 'after': 'after'} in audit_changes[-1] + assert 'old-secret' not in str(audit_changes[-1]) and 'new-secret' not in str(audit_changes[-1]) + async with engine.connect() as connection: + rows = (await connection.execute(sa.select(MCPServer))).mappings().all() + assert next(row for row in rows if row['uuid'] == 'owned-a')['extra_args'] == {'api_key': 'new-secret'} + assert all(row['readme'] == 'before' for row in rows if row['workspace_uuid'] == 'workspace-b') + assert (await client.delete(endpoint, headers=headers)).status_code == 200 + async with engine.connect() as connection: + assert set(await connection.scalars(sa.select(MCPServer.uuid))) == {'owned-b', 'foreign-b'} + assert (await client.get(endpoint, headers=headers)).status_code == 403 + + +@pytest.mark.parametrize('method', ['put', 'delete']) +async def test_write_scopes_and_foreign_names_fail_without_mutation(mcp_write_api, method): + client, engine, keys, _ = mcp_write_api + request = getattr(client, method) + kwargs = {'json': {'readme': 'stolen'}} if method == 'put' else {} + assert ( + await request('/api/v1/mcp/servers/owned', headers={'X-API-Key': keys['empty']}, **kwargs) + ).status_code == 403 + for name in ('foreign', 'missing'): + response = await request( + '/api/v1/mcp/servers/' + name, + headers={'X-API-Key': keys['manager'], 'X-Workspace-Id': 'workspace-b'}, + **kwargs, + ) + assert response.status_code == 404 + assert await response.get_json() == {'code': -1, 'msg': 'Server not found'} + async with engine.connect() as connection: + rows = (await connection.execute(sa.select(MCPServer))).mappings().all() + assert len(rows) == 3 + assert all(row['readme'] == 'before' for row in rows) diff --git a/tests/integration/api/test_resource_write_audit.py b/tests/integration/api/test_resource_write_audit.py new file mode 100644 index 000000000..f8daa79ff --- /dev/null +++ b/tests/integration/api/test_resource_write_audit.py @@ -0,0 +1,134 @@ +"""Narrow write scopes retain HTTP audit snapshots without acquiring read access.""" + +import json + +import pytest +import quart +import sqlalchemy as sa + +from langbot.pkg.api.http.controller.groups.knowledge.base import KnowledgeBaseRouterGroup +from langbot.pkg.api.http.controller.groups.pipelines.pipelines import PipelinesRouterGroup +from langbot.pkg.api.http.controller.groups.provider.models import LLMModelsRouterGroup +from langbot.pkg.api.http.controller.groups.provider.providers import ModelProvidersRouterGroup +from langbot.pkg.api.http.controller.groups.skills import SkillsRouterGroup +from langbot.pkg.api.http.service.skill import SkillService +from langbot.pkg.entity.persistence.operation_log import WorkspaceOperationLog +from langbot.pkg.operation_trace import service as tracing +from langbot.pkg.skill.repository import SkillRepository +from tests.integration.api.test_application_foundation import context, foundation # noqa: F401 +from tests.integration.api.test_application_resources import resources # noqa: F401 + + +pytestmark = [pytest.mark.integration, pytest.mark.asyncio] + + +@pytest.mark.parametrize('kind', ['pipeline', 'pipeline_extensions', 'provider', 'model', 'knowledge', 'skill']) +async def test_write_only_keys_preserve_before_values_and_one_redacted_audit(resources, tmp_path, monkeypatch, kind): # noqa: F811 + api, ap, engine = resources + ap.application_api = api + scope = 'provider_secret.manage' if kind in {'provider', 'model'} else 'resource.manage' + key = (await ap.apikey_service.create_api_key(context(), 'writer', scopes=[scope]))['key'] + empty_key = (await ap.apikey_service.create_api_key(context(), 'empty', scopes=[]))['key'] + body = {'name': 'after'} + expected_change = {'field': 'name', 'before': 'before', 'after': 'after'} + if kind.startswith('pipeline'): + identity = await api.execute( + 'pipeline.create', + context(), + { + 'pipeline_data': { + 'name': 'before', + 'config': {'ai': {'api_key': 'old-secret'}}, + 'extensions_preferences': {'enable_all_plugins': False, 'plugins': []}, + } + }, + ) + endpoint = '/api/v1/pipelines/' + identity + router = PipelinesRouterGroup + if kind == 'pipeline_extensions': + endpoint += '/extensions' + body = {'enable_all_plugins': True} + expected_change = {'field': 'enable_all_plugins', 'before': False, 'after': True} + else: + body['config'] = {'ai': {'api_key': 'new-secret'}} + elif kind in {'provider', 'model'}: + provider = await api.execute( + 'provider.create', + context(), + { + 'provider_data': { + 'name': 'before', + 'requester': 'test-requester', + 'base_url': 'https://example.test/v1', + 'api_keys': ['old-secret'], + } + }, + ) + if kind == 'provider': + endpoint = '/api/v1/provider/providers/' + provider + router = ModelProvidersRouterGroup + body['api_keys'] = ['new-secret'] + else: + identity = await api.execute( + 'model.llm.create', + context(), + {'model_data': {'name': 'before', 'provider_uuid': provider, 'extra_args': {'api_key': 'old-secret'}}}, + ) + endpoint = '/api/v1/provider/models/llm/' + identity + router = LLMModelsRouterGroup + body['extra_args'] = {'api_key': 'new-secret'} + elif kind == 'knowledge': + identity = await api.execute( + 'knowledge.base.create', + context(), + { + 'kb_data': { + 'name': 'before', + 'knowledge_engine_plugin_id': 'test/engine', + 'defer_initialization': True, + 'creation_settings': {'api_key': 'old-secret'}, + } + }, + ) + endpoint = '/api/v1/knowledge/bases/' + identity + router = KnowledgeBaseRouterGroup + else: + ap.instance_config.data['skills'] = {'root': str(tmp_path / 'skills')} + ap.skill_repository = SkillRepository(ap) + ap.skill_service = SkillService(ap) + skill = await api.execute( + 'skill.create', + context(), + {'data': {'name': 'audit-skill', 'description': 'before', 'instructions': 'Read me'}}, + ) + endpoint = '/api/v1/skills/audit-skill' + router = SkillsRouterGroup + body = {'description': 'after', 'base_revision': skill['revision']} + expected_change = {'field': 'description', 'before': 'before', 'after': 'after'} + monkeypatch.setattr(tracing, '_global_enabled', False) + audit = ap.workspace_settings_service = tracing.WorkspaceSettingsService(ap) + await audit.set_operation_level('workspace-a', tracing.OPERATION_LEVEL_MUTATION, dedupe_window_seconds=0) + app = quart.Quart(__name__) + await router(ap, app).initialize() + client = app.test_client() + headers = {'X-API-Key': key} + assert (await client.get(endpoint, headers=headers)).status_code == 403 + assert (await client.put(endpoint, headers={'X-API-Key': empty_key}, json=body)).status_code == 403 + response = await client.put(endpoint, headers={**headers, 'X-Request-Id': 'write-audit'}, json=body) + assert response.status_code == 200, await response.get_data(as_text=True) + await audit.flush_pending() + async with engine.connect() as connection: + rows = ( + ( + await connection.execute( + sa.select(WorkspaceOperationLog).where(WorkspaceOperationLog.request_id == 'write-audit') + ) + ) + .mappings() + .all() + ) + assert len(rows) == 1 + assert rows[0]['status_code'] == 200 and rows[0]['outcome'] == 'ok' + assert expected_change in json.loads(rows[0]['changes']) + assert 'old-secret' not in str(rows[0]) and 'new-secret' not in str(rows[0]) + assert (await client.get(endpoint, headers=headers)).status_code == 403 diff --git a/tests/integration/api/test_workspaces.py b/tests/integration/api/test_workspaces.py index 9579289d0..bd8578887 100644 --- a/tests/integration/api/test_workspaces.py +++ b/tests/integration/api/test_workspaces.py @@ -1095,3 +1095,35 @@ async def test_migrated_workspace_mutations_record_one_rich_audit_each(workspace assert exported.status_code == 200 assert 'text/csv' in exported.content_type assert 'member_role_update' in await exported.get_data(as_text=True) + + +@pytest.mark.parametrize('failure,expected_status', [('invalid_input', 400), ('invalid_output', 500)]) +async def test_application_failure_audit_matches_http_status(workspace_api, monkeypatch, failure, expected_status): + from langbot.pkg.operation_trace import service as tracing + from langbot.pkg.entity.persistence.operation_log import WorkspaceOperationLog + + application, client, engine, token = workspace_api + monkeypatch.setattr(tracing, '_global_enabled', False) + service = application.workspace_settings_service = tracing.WorkspaceSettingsService(application) + current = await client.get('/api/v1/workspaces/current', headers=_auth(token)) + workspace_uuid = (await current.get_json())['data']['workspace']['uuid'] + await service.set_operation_level(workspace_uuid, tracing.OPERATION_LEVEL_READ, dedupe_window_seconds=0) + headers = {**_auth(token), 'X-Request-Id': failure} + if failure == 'invalid_input': + response = await client.put('/api/v1/system/wizard/progress', headers=headers, json={'step': 'invalid'}) + assert (await response.get_json())['msg'] == 'Invalid fields: step' + else: + application.space_service = SimpleNamespace(get_recommended_chat_model=AsyncMock(return_value=[])) + response = await client.get('/api/v1/system/wizard/recommended-model', headers=headers) + assert (await response.get_json())['code'] == 'internal_error' + assert response.status_code == expected_status + await service.flush_pending() + async with engine.connect() as connection: + rows = ( + await connection.execute( + sqlalchemy.select(WorkspaceOperationLog).where(WorkspaceOperationLog.request_id == failure) + ) + ).all() + assert len(rows) == 1 + assert rows[0].status_code == expected_status + assert rows[0].outcome == 'error' diff --git a/tests/unit_tests/api/test_application_workspace_streams.py b/tests/unit_tests/api/test_application_workspace_streams.py index 8f68ebb0c..533f66371 100644 --- a/tests/unit_tests/api/test_application_workspace_streams.py +++ b/tests/unit_tests/api/test_application_workspace_streams.py @@ -105,6 +105,7 @@ async def test_assistant_stream_disconnect_preserves_run_and_bounded_queue(assis conversation_id = (await service.create(ctx))['uuid'] stream = await assistant_turn(ap, ctx, TurnInput(conversation_id=conversation_id, revision=0, text='Hi')) try: + assert await anext(stream) == {'kind': 'started', 'data': {'conversation_id': conversation_id}} while (await asyncio.wait_for(anext(stream), 2))['kind'] != 'text': pass task = service._runs[conversation_id]['task'] diff --git a/tests/unit_tests/api/test_management_assistant.py b/tests/unit_tests/api/test_management_assistant.py index 4b65d1dec..039220461 100644 --- a/tests/unit_tests/api/test_management_assistant.py +++ b/tests/unit_tests/api/test_management_assistant.py @@ -583,3 +583,69 @@ async def test_background_assistant_tools_keep_their_audit_after_http_dispatch(a assert record['route'] == 'pipeline.create' assert record['outcome'] == 'ok' assert (await service.get(ctx, cid))['status'] == 'ready' + + +@pytest.mark.asyncio +async def test_http_stream_headers_and_heartbeats_do_not_wait_for_model_resolution(assistant): + import json + from quart import Quart + from quart.testing.connections import TestHTTPConnection + from langbot_plugin.api.entities.builtin.provider.message import MessageChunk + from langbot.pkg.api.http.controller.groups.assistant import AssistantRouterGroup + + service, ap, provider = assistant + blocked, release, headers_ready = asyncio.Event(), asyncio.Event(), asyncio.Event() + + async def recommend(*args): + blocked.set() + await release.wait() + return {'uuid': 'model'} + + async def model_stream(**kwargs): + yield MessageChunk(role='assistant', content='Finished', is_final=True) + + ap.space_service.get_recommended_assistant_model.side_effect = recommend + provider.invoke_llm_stream = model_stream + ctx = context() + cid = (await service.create(ctx))['uuid'] + app = Quart(__name__) + router = AssistantRouterGroup(ap, app) + router._authenticate_support_admin = AsyncMock(return_value=None) + router._authenticate_account = AsyncMock(return_value=(SimpleNamespace(uuid='alice'), 'alice@example.invalid')) + router._resolve_account_context = AsyncMock(return_value=ctx) + await router.initialize() + + class Connection(TestHTTPConnection): + async def _asgi_send(self, message): + await super()._asgi_send(message) + if message['type'] == 'http.response.start': + headers_ready.set() + + client = app.test_client() + client.http_connection_class = Connection + async with client.request( + f'/api/v1/assistant/conversations/{cid}/turn/stream', + method='POST', + headers={'Authorization': 'Bearer synthetic-test', 'Content-Type': 'application/json'}, + ) as connection: + try: + await connection.send(json.dumps({'revision': 0, 'text': 'Hi'}).encode()) + await connection.send_complete() + await asyncio.wait_for(headers_ready.wait(), 1) + await asyncio.wait_for(blocked.wait(), 1) + assert connection.status_code == 200 + for _ in range(2): + assert await asyncio.wait_for(connection.receive(), 16) == b'\n' + assert not release.is_set() + assert not service._runs[cid]['task'].done() + release.set() + frames = [] + while not any(frame['kind'] == 'completed' for frame in frames): + chunk = await asyncio.wait_for(connection.receive(), 2) + frames.extend(json.loads(line) for line in chunk.decode().splitlines() if line.strip()) + assert all(frame['kind'] != 'started' for frame in frames) + assert frames[-1]['data']['messages'][-1]['content'] == 'Finished' + finally: + release.set() + for run in list(service._runs.values()): + await asyncio.wait_for(run['task'], 2)