"""自主模式的真实沙箱执行、交付归档和权限边界；不调用外部模型或网关。""" import asyncio import json import socket from pathlib import Path from types import SimpleNamespace import pytest from PIL import Image import config from agents.workspace import session_root from db import main_fan_store from extensions.main_fan import exploration, service from extensions.main_fan.schemas import AgentAssessment from extensions.registry import load_extensions from runtime import python_sandbox @pytest.fixture def workbench(monkeypatch, tmp_path): monkeypatch.setattr(config, 'ROOT', tmp_path) monkeypatch.setenv('MAIN_FAN_TARGETS_FILE', '') monkeypatch.setenv('MAIN_FAN_RULES_FILE', '') user, session = 'sandbox-user', 'sandbox-session' workspace = session_root(user, session) / 'workspace' uploads = workspace / 'uploads' uploads.mkdir(parents=True) source = uploads / 'scan.pdf' with Image.new('RGB', (120, 160), 'white') as image: image.save(source, 'PDF') request = {'mine_code': 'sandbox-mine', 'mine_name': '沙箱测试矿', 'fan_id': 'fan-1', 'fan_name': '1号风机', 'fan_pair': ['fan-1', 'fan-2'], 'materials': [{'path': '/workspace/uploads/scan.pdf'}], 'measurements': {'values': [10, 20, 30]}} prepared = exploration.prepare(request, user, session) return SimpleNamespace(user=user, session=session, workspace=workspace, request=request, manifest=prepared, source=source) def require_sandbox(): if not python_sandbox.capability()['available']: pytest.skip(python_sandbox.capability()['reason']) def delivery_script(workbench, *, code='WARNING', summary='脚本实际计算均值20.0', status='completed'): """测试模型写出的程序：读原 PDF、渲染、计算并生成所有交付，不调用业务服务。""" identifier = workbench.manifest['workflow_id'] return f''' from pathlib import Path import json import statistics import pypdfium2 as pdfium from PIL import Image import numpy as np import cv2 base = Path('exploration/{identifier}') data = json.loads((base/'input.json').read_text()) mean = statistics.mean(data['measurements']['values']) with pdfium.PdfDocument('uploads/scan.pdf') as pdf: pages = len(pdf) page = pdf[0] try: textpage = page.get_textpage() try: text = textpage.get_text_range() finally: textpage.close() bitmap = page.render(scale=1) image = bitmap.to_pil() image.save('page.png') image.close() bitmap.close() finally: page.close() facts = {{'page_count': pages, 'text': text, 'mean': mean}} Path('facts.json').write_text(json.dumps(facts)) report = '# Agent 生成的测试报告\\n\\n均值：' + str(mean) + '；PDF页数：' + str(pages) Path('report.md').write_text(report) result = {{'status': {status!r}, 'result_code': {code!r}, 'summary': {summary!r}, 'data_detail': '测试原件及本轮明确提供的数据；非现场结果', 'evidence': [{{'source': '/workspace/exploration/{identifier}/input.json', 'pointer': '/measurements/values', 'excerpt': '10、20、30'}}], 'details': {{'mean': mean, 'pages': pages, 'custom_method': 'statistics.mean'}}}} Path('assessment.json').write_text(json.dumps(result, ensure_ascii=False)) print('computed', mean, 'pages', pages) ''' async def execute_delivery(workbench, **kwargs): require_sandbox() script = workbench.workspace / 'analyze.py' script.write_text(delivery_script(workbench, **kwargs), encoding='utf-8') receipt = await python_sandbox.run(workbench.user, workbench.session, '/workspace/analyze.py', output_paths=['/workspace/report.md', '/workspace/assessment.json', '/workspace/facts.json', '/workspace/page.png']) assert receipt['exit_code'] == 0, receipt['stderr'] assert not receipt['timed_out'] exploration.record_python(workbench.manifest['workflow_id'], workbench.user, workbench.session, receipt) return receipt def submit_delivery(workbench, run_ids, **kwargs): return exploration.submit(workbench.manifest['workflow_id'], workbench.user, workbench.session, '/workspace/report.md', '/workspace/assessment.json', run_ids, **kwargs) def test_only_gateway_batch_and_infrastructure_tools_are_registered(): catalog = load_extensions() names = set(catalog.agents['main_fan'].tools) assert names == {'prepare_main_fan_task', 'fetch_main_fan_data', 'query_fan_timeseries', 'query_fan_sensor_realtime', 'query_fan_equipment_realtime', 'query_fan_equipment_timeseries', 'run_python', 'submit_main_fan_report', 'submit_main_fan_mine_plan', 'query_main_fan_history', 'fetch_main_fan_reports', 'archive_main_fan_report'} assert names.isdisjoint({'inspect_main_fan_document', 'extract_main_fan_pages', 'register_main_fan_facts', 'compare_main_fan_data', 'analyze_main_fan'}) assert catalog.business('main_fan').test_runner is None @pytest.mark.asyncio async def test_direct_pdf_read_returns_recoverable_feedback_without_sending_binary(): from extensions.main_fan.middleware import MainFanSandboxMiddleware middleware = MainFanSandboxMiddleware() request = SimpleNamespace(tool_call={'name': 'read_file', 'id': 'pdf-read', 'args': {'file_path': '/workspace/uploads/report.pdf'}}) async def forbidden(request): pytest.fail('不应让框架发送 PDF 二进制附件') result = await middleware.awrap_tool_call(request, forbidden) assert result.status == 'error' and isinstance(result.content, str) assert 'run_python' in result.content and 'PNG' in result.content def test_prepare_exports_raw_files_and_frozen_delivery_schema(workbench): manifest = workbench.manifest assert manifest['mode'] == 'sandbox_exploration' assert manifest['python_paths']['input'] == manifest['input_path'].removeprefix('/workspace/') assert manifest['python_paths']['materials'] == ['uploads/scan.pdf'] assert 'facts_schema_path' not in manifest data = json.loads((workbench.workspace / manifest['data_path'].removeprefix('/workspace/')).read_text()) assert data['reports'] == [] schema = json.loads((workbench.workspace / manifest['submission_schema_path'].removeprefix('/workspace/')).read_text()) assert {'result_code', 'summary'} <= set(schema['required']) assert exploration.prepare(workbench.request, workbench.user, workbench.session)['workflow_id'] == manifest['workflow_id'] trusted = exploration.prepare({'mine_code': 'fake', 'mine_name': '伪造'}, workbench.user, workbench.session, trusted=workbench.request) assert trusted['mine_code'] == 'sandbox-mine' def test_test_lab_defaults_to_agent_and_cannot_use_legacy_pipeline(monkeypatch): from api.test_lab_routes import TestRequest as Request from runtime import business_tasks from runtime.test_lab import service as lab from db import business_store monkeypatch.setattr(business_tasks, '_controller', business_tasks.BusinessTaskController()) request = Request(agent_name='main_fan', input={'mine_code': 'test', 'mine_name': '测试矿'}) assert request.execution_mode == 'agent' result = lab.start_run(request, 'lab-default-test') assert business_store.get(result['id'])['source'] == 'test_agent' with pytest.raises(ValueError, match='完整智能体'): lab.start_run(request.model_copy(update={'execution_mode': 'pipeline'}), 'lab-default-test') @pytest.mark.asyncio async def test_real_sandbox_parses_pdf_computes_and_generates_report_without_legacy_pipeline(workbench, monkeypatch): def forbidden(*args, **kwargs): raise AssertionError('自主模式不能调用固定业务解析、计算或报告生成') from algorithms.main_fan import equivalence from extensions.main_fan import documents, pipeline, reporting for name in ('normalize_and_validate_fan_data', 'compare_basic_parameters', 'analyze_fan_switch_event', 'assess_fan_equivalent_capability'): monkeypatch.setattr(equivalence, name, forbidden) monkeypatch.setattr(documents, '_prepare', forbidden) monkeypatch.setattr(pipeline, 'run_pipeline', forbidden) monkeypatch.setattr(reporting, 'render_result', forbidden) monkeypatch.setattr(service, '_screen', forbidden) receipt = await execute_delivery(workbench) assert 'computed 20 pages 1' in receipt['stdout'] assert (workbench.workspace / 'page.png').is_file() result = submit_delivery(workbench, [receipt['run_id']], artifacts=['/workspace/facts.json']) assert result['status'] == 'completed' assert result['analysis'] is None assert result['result'] == (workbench.workspace / 'report.md').read_text() assert result['assessment']['agent_result']['details']['mean'] == 20 # 故意不生成旧七阶段结构；平台编码与摘要完全来自 Agent 交付，不被固定算法改成 UNDETERMINED。 judgment = result['platform_payload']['judgment'] assert judgment['resultCode'] == 'WARNING' assert judgment['result'] == '脚本实际计算均值20.0' assert result['assessment']['evidence'][0]['sha256'] from runtime.test_lab.spreadsheets import result_tables, write_workbook, preview_workbook workbook = workbench.workspace / 'export.xlsx' write_workbook(workbook, result_tables(result)) preview = preview_workbook(workbook) assert preview['sheets'][1]['name'] == 'Agent分析明细' assert ['mean', 20] in preview['sheets'][1]['rows'] assert len(main_fan_store.workflow(workbench.manifest['workflow_id'], workbench.user, workbench.session)['stages']) == 3 assert service.validate_result({**result, 'result': '篡改'}, workbench.user, workbench.session) == result assert submit_delivery(workbench, [receipt['run_id']]) == result with pytest.raises(ValueError, match='不属于'): service.validate_result(result, 'another-user', workbench.session) @pytest.mark.asyncio async def test_relative_evidence_path_binds_exact_same_trusted_source(workbench): require_sandbox() code = delivery_script(workbench).replace( '/workspace/exploration/' + workbench.manifest['workflow_id'] + '/input.json', 'exploration/' + workbench.manifest['workflow_id'] + '/input.json') (workbench.workspace / 'relative.py').write_text(code) receipt = await python_sandbox.run(workbench.user, workbench.session, '/workspace/relative.py', output_paths=['/workspace/report.md', '/workspace/assessment.json']) assert receipt['ok'], receipt['stderr'] exploration.record_python(workbench.manifest['workflow_id'], workbench.user, workbench.session, receipt) result = submit_delivery(workbench, [receipt['run_id']]) assert result['assessment']['evidence'][0]['source'] == workbench.manifest['input_path'] @pytest.mark.asyncio @pytest.mark.parametrize('target', ['report.md', 'assessment.json', 'facts.json']) async def test_changed_script_outputs_cannot_be_submitted(workbench, target): receipt = await execute_delivery(workbench) (workbench.workspace / target).write_text('tampered') with pytest.raises(ValueError): submit_delivery(workbench, [receipt['run_id']], artifacts=['/workspace/facts.json']) @pytest.mark.asyncio async def test_delivery_requires_successful_execution_from_same_workflow(workbench): receipt = await execute_delivery(workbench) with pytest.raises(ValueError, match='成功执行记录'): submit_delivery(workbench, []) second = exploration.prepare({**workbench.request, 'task_description': '独立任务'}, workbench.user, workbench.session) with pytest.raises(ValueError, match='不属于本轮'): exploration.submit(second['workflow_id'], workbench.user, workbench.session, '/workspace/report.md', '/workspace/assessment.json', [receipt['run_id']]) with pytest.raises(ValueError, match='不属于'): exploration._load(workbench.manifest['workflow_id'], 'someone-else', workbench.session) @pytest.mark.asyncio async def test_python_tool_returns_concrete_paths_when_agent_code_fails(workbench): require_sandbox() from extensions.main_fan.exploration_tools import run_python (workbench.workspace / 'wrong-path.py').write_text("from pathlib import Path\nPath('missing-input.json').read_text()") runtime = SimpleNamespace(config={'configurable': {'workspace_user': workbench.user, 'workspace_session': workbench.session}}) result = await run_python('/workspace/wrong-path.py', runtime, workflow_id=workbench.manifest['workflow_id']) assert result['ok'] is False assert 'FileNotFoundError' in result['stderr'] assert result['python_paths']['input'] == workbench.manifest['python_paths']['input'] assert '不要' in result['next_action'] @pytest.mark.asyncio async def test_preexisting_files_are_not_mislabeled_as_script_generated_outputs(workbench): require_sandbox() (workbench.workspace / 'existing.md').write_text('只是之前写入的文件') (workbench.workspace / 'noop.py').write_text("print('no output written')") receipt = await python_sandbox.run(workbench.user, workbench.session, '/workspace/noop.py', output_paths=['/workspace/existing.md']) assert receipt['ok'] assert receipt['outputs'] == [] assert receipt['unchanged_outputs'] == ['/workspace/existing.md'] @pytest.mark.parametrize('number', ['NaN', 'Infinity', '1e999']) def test_delivery_json_rejects_nonfinite_numbers(tmp_path, number): path = tmp_path / 'result.json' path.write_text('{"details":{"value":' + number + '}}') with pytest.raises(ValueError, match='非有限'): exploration._read_json(path) @pytest.mark.asyncio async def test_raw_sources_and_frozen_inputs_cannot_be_replaced(workbench): receipt = await execute_delivery(workbench) original = workbench.source.read_bytes() workbench.source.write_bytes(b'changed pdf') with pytest.raises(ValueError, match='已被修改'): submit_delivery(workbench, [receipt['run_id']]) workbench.source.write_bytes(original) rules = workbench.workspace / workbench.manifest['rules_path'].removeprefix('/workspace/') rules.write_text('{}') with pytest.raises(ValueError, match='已被修改'): submit_delivery(workbench, [receipt['run_id']]) @pytest.mark.asyncio async def test_unknown_evidence_is_rejected_and_agent_can_correct_and_resubmit(workbench): receipt = await execute_delivery(workbench) # 修改生成脚本后重新执行，保证失败来自来源归属而非哈希变化。 script = delivery_script(workbench).replace('/input.json', '/other-task.json') (workbench.workspace / 'analyze.py').write_text(script) bad = await python_sandbox.run(workbench.user, workbench.session, '/workspace/analyze.py', output_paths=['/workspace/report.md', '/workspace/assessment.json']) assert bad['exit_code'] == 0, bad['stderr'] exploration.record_python(workbench.manifest['workflow_id'], workbench.user, workbench.session, bad) with pytest.raises(ValueError, match='证据来源'): submit_delivery(workbench, [bad['run_id']]) corrected = await execute_delivery(workbench) assert submit_delivery(workbench, [corrected['run_id']])['analysis_id'] @pytest.mark.asyncio async def test_gateway_fetch_still_uses_host_connector_and_updates_raw_files(workbench, monkeypatch): import hashlib from extensions.main_fan import data_source inputs = {**workbench.request, 'use_gateway': True, 'materials': []} prepared = exploration.prepare(inputs, workbench.user, workbench.session) calls = [] class Gateway: async def download(self, agent_code, source, local): calls.append(('download', agent_code, source.file_id)) local.write_bytes(workbench.source.read_bytes()) return {'sha256': hashlib.sha256(local.read_bytes()).hexdigest()} gateway = Gateway() async def fetch(connector, agent_code, mine_code): assert connector is gateway calls.append(('fetch', agent_code, mine_code)) return {'devices': [], 'points': [], 'operating_series': [], 'switch_records': [], 'fan_state_records': [], 'files': [{'file_id': 'file-1', 'version': 'v1', 'filename': 'remote.pdf', 'path': '/files/1'}]} monkeypatch.setattr(exploration, 'get_service', lambda name: gateway) monkeypatch.setattr(data_source, 'fetch', fetch) first = await exploration.fetch(prepared['workflow_id'], workbench.user, workbench.session) second = await exploration.fetch(prepared['workflow_id'], workbench.user, workbench.session) assert len(first['materials']) == len(second['materials']) == 1 assert [call[0] for call in calls] == ['fetch', 'download'] material = first['materials'][0] assert (workbench.workspace / material['path'].removeprefix('/workspace/')).is_file() raw = json.loads((workbench.workspace / first['input_path'].removeprefix('/workspace/')).read_text()) assert raw['materials'][0]['sha256'] == material['sha256'] @pytest.mark.asyncio async def test_sandbox_denies_host_secrets_network_and_process_execution(workbench, monkeypatch): require_sandbox() secret = workbench.workspace.parent / 'host-secret.txt' secret.write_text('host-only-value') monkeypatch.setenv('SANDBOX_TEST_SECRET', 'not-for-child') listener = socket.socket() listener.bind(('127.0.0.1', 0)) listener.listen() port = listener.getsockname()[1] program = ''' from pathlib import Path import os, socket, subprocess, sys assert os.getenv('SANDBOX_TEST_SECRET') is None try: Path('../host-secret.txt').read_text() except (PermissionError, FileNotFoundError): print('host-file-denied') else: raise AssertionError('host file readable') try: socket.create_connection(('127.0.0.1', TEST_PORT), timeout=1) except OSError: print('network-denied') else: raise AssertionError('network accessible') if sys.platform == 'darwin': try: subprocess.run(['/bin/sh', '-c', 'true'], check=True) except PermissionError: print('process-denied') else: raise AssertionError('arbitrary process allowed') else: print('processes-remain-in-namespace') ''' (workbench.workspace / 'boundary.py').write_text(program.replace('TEST_PORT', str(port))) try: receipt = await python_sandbox.run(workbench.user, workbench.session, '/workspace/boundary.py') listener.settimeout(.1) with pytest.raises(TimeoutError): listener.accept() finally: listener.close() assert receipt['exit_code'] == 0, receipt['stderr'] assert 'host-file-denied' in receipt['stdout'] assert 'network-denied' in receipt['stdout'] assert 'process-denied' in receipt['stdout'] or 'processes-remain-in-namespace' in receipt['stdout'] @pytest.mark.asyncio async def test_sandbox_failure_timeout_and_path_escape_are_real(workbench): require_sandbox() script = workbench.workspace / 'failure.py' script.write_text("raise RuntimeError('real-script-failure')") failed = await python_sandbox.run(workbench.user, workbench.session, '/workspace/failure.py') assert failed['exit_code'] != 0 and 'real-script-failure' in failed['stderr'] with pytest.raises(ValueError, match='失败或超时'): python_sandbox.execution_receipt(workbench.user, workbench.session, failed['run_id']) script.write_text('import time\ntime.sleep(10)') timed = await python_sandbox.run(workbench.user, workbench.session, '/workspace/failure.py', timeout_seconds=1) assert timed['timed_out'] with pytest.raises(ValueError, match='越出'): python_sandbox.workspace_file(workbench.user, workbench.session, '/workspace/../../secret') (workbench.workspace / 'escape.py').symlink_to(workbench.workspace.parent / 'outside.py') with pytest.raises(ValueError, match='越出'): await python_sandbox.run(workbench.user, workbench.session, '/workspace/escape.py') @pytest.mark.asyncio async def test_missing_sandbox_never_falls_back_to_host_execution(workbench, monkeypatch): monkeypatch.setattr(python_sandbox.shutil, 'which', lambda name: None) (workbench.workspace / 'touch.py').write_text("from pathlib import Path\nPath('unsafe-marker').touch()") with pytest.raises(ValueError, match='沙箱执行'): await python_sandbox.run(workbench.user, workbench.session, '/workspace/touch.py') assert not (workbench.workspace / 'unsafe-marker').exists() @pytest.mark.parametrize('value', [ {'result_code': 'NORMAL', 'summary': '无来源'}, {'result_code': 'BAD', 'summary': '未知编码'}, {'result_code': 'UNDETERMINED', 'summary': ' '}, {'status': 'needs_materials', 'result_code': 'WARNING', 'summary': '矛盾状态', 'evidence': [{'source': '/workspace/input.json'}]}, ]) def test_delivery_contract_rejects_invalid_metadata(value): with pytest.raises(ValueError): AgentAssessment.model_validate(value)