File size: 4,013 Bytes
1140c43 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 | """完整诊断、分片流、持久状态和异常恢复出口的脱敏回归。"""
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]")
|