| """完整诊断、分片流、持久状态和异常恢复出口的脱敏回归。""" |
| import asyncio |
| import json |
| import os |
| import sys |
| import tempfile |
| import unittest |
| from pathlib import Path |
| from types import SimpleNamespace |
| from unittest.mock import AsyncMock, patch |
|
|
| ROOT = Path(__file__).resolve().parents[2] |
| sys.path.insert(0, str(ROOT / "apps/market-api")) |
| from app.native.config import Settings |
| from app.native.operations import Operations |
| from app.native.store import Store |
| from app.native.supervisor import Supervisor |
|
|
|
|
| class RedactionTests(unittest.IsolatedAsyncioTestCase): |
| def setUp(self): |
| self.temporary = tempfile.TemporaryDirectory(prefix="message-redaction-") |
| self.root = Path(self.temporary.name) |
| self.settings = Settings(data_dir=self.root, admin_token="fixture-control-admin", mcp_tokens={"a": "fixture-control-mcp-a"}, allowed_hosts=("localhost",), allowed_origins=("http://localhost",), plugin_env_allowlist={"a": ["REDACTION_FIXTURE_SECRET"]}) |
| self.store = Store(self.root) |
| self.supervisor = Supervisor(self.settings, self.store) |
| self.secret = 'fixture-中文-quoted"-secret' |
|
|
| def tearDown(self): |
| self.store.close() |
| self.temporary.cleanup() |
|
|
| async def test_every_byte_boundary_is_redacted_after_drain(self): |
| raw = self.secret.encode() |
| with patch.dict(os.environ, {"REDACTION_FIXTURE_SECRET": self.secret}): |
| for boundary in range(1, len(raw)): |
| stream = asyncio.StreamReader() |
| worker = SimpleNamespace(diagnostics="") |
| draining = asyncio.create_task(self.supervisor.drain("a", worker, stream)) |
| stream.feed_data(raw[:boundary]) |
| await asyncio.sleep(0) |
| stream.feed_data(raw[boundary:]) |
| stream.feed_eof() |
| await draining |
| self.assertEqual(worker.diagnostics, "[REDACTED]") |
|
|
| async def test_store_and_error_reconciliation_have_final_boundary(self): |
| with patch.dict(os.environ, {"REDACTION_FIXTURE_SECRET": self.secret}): |
| self.store.save_profile({"id": "a", "last_error": self.secret}) |
| self.supervisor.mutate = AsyncMock(side_effect=RuntimeError("operation: " + self.secret)) |
| self.supervisor.refresh = AsyncMock(side_effect=RuntimeError("refresh: " + self.secret + " " + self.settings.admin_token)) |
| operations = Operations(self.store, self.supervisor) |
| queued = operations.submit("a", "restart", {}, None) |
| await operations.tasks[queued["id"]] |
| completed = self.store.operation(queued["id"]) |
| self.assertEqual(completed["status"], "failed") |
| self.assertNotIn(self.secret, json.dumps(completed, ensure_ascii=False)) |
| self.assertIn("[REDACTED]", completed["error"]) |
| self.assertIn("[REDACTED]", self.store.profile("a")["last_error"]) |
| |
| completed.update(error=self.secret, result={"diagnostic": self.secret, "token": self.settings.mcp_tokens["a"]}) |
| self.store.save_operation(completed) |
| for forbidden in (self.secret, self.settings.admin_token, self.settings.mcp_tokens["a"]): |
| for path in self.root.glob("control.sqlite3*"): |
| self.assertNotIn(forbidden.encode(), path.read_bytes()) |
|
|
| async def test_cached_values_and_json_escaping_are_redacted(self): |
| with patch.dict(os.environ, {"REDACTION_FIXTURE_SECRET": self.secret}): |
| self.supervisor.redact("a", "") |
| for value in (self.secret, json.dumps(self.secret, ensure_ascii=True)[1:-1], json.dumps(self.secret, ensure_ascii=False)[1:-1]): |
| self.assertEqual(self.supervisor.redact("a", value), "[REDACTED]") |
| self.assertEqual(self.supervisor.redact("b", self.secret), self.secret) |
| self.assertEqual(self.supervisor.redact(None, self.secret), "[REDACTED]") |
|
|
