fix(buzz): redact media paths before bounding errors

This commit is contained in:
EmpireOperating
2026-08-06 10:01:20 -06:00
committed by Teknium
parent fcd34e57a2
commit a55d66b79a
2 changed files with 48 additions and 5 deletions

View File

@@ -666,14 +666,21 @@ async def _exec_buzz(
_MAX_CLI_MESSAGE_CHARS = 900
def _bounded_cli_message(message: str) -> str:
def _bounded_cli_message(message: str, redact_path: Optional[Path] = None) -> str:
"""Keep untrusted CLI detail useful without exposing unbounded output."""
if redact_path is not None:
message = message.replace(str(redact_path), redact_path.name)
if len(message) <= _MAX_CLI_MESSAGE_CHARS:
return message
return f"{message[: _MAX_CLI_MESSAGE_CHARS - 3]}..."
def _cli_error_message(stderr: str, returncode: int) -> str:
def _cli_error_message(
stderr: str,
returncode: int,
*,
redact_path: Optional[Path] = None,
) -> str:
"""Extract a bounded human-readable message from the CLI error contract."""
text = (stderr or "").strip()
try:
@@ -684,11 +691,15 @@ def _cli_error_message(stderr: str, returncode: int) -> str:
if isinstance(detail, str) and detail.strip():
label = category.strip() if isinstance(category, str) and category.strip() else "error"
return _bounded_cli_message(
f"{label}: {detail.strip()} (exit {returncode})"
f"{label}: {detail.strip()} (exit {returncode})",
redact_path,
)
except ValueError:
pass
return _bounded_cli_message(text or f"buzz CLI failed with exit code {returncode}")
return _bounded_cli_message(
text or f"buzz CLI failed with exit code {returncode}",
redact_path,
)
def _parse_send_receipt(stdout: str) -> Tuple[Optional[str], Optional[str]]:
@@ -1501,7 +1512,7 @@ class BuzzAdapter(BasePlatformAdapter):
if code != 0:
return SendResult(
success=False,
error=_cli_error_message(err, code),
error=_cli_error_message(err, code, redact_path=local),
retryable=code == 2,
)
event_id, receipt_error = _parse_send_receipt(out)

View File

@@ -3057,6 +3057,38 @@ class TestInboundMediaAuthorizationGate:
assert result.error == "invalid CLI response"
assert result.raw_response is None
@pytest.mark.asyncio
async def test_live_media_redacts_long_path_before_bounding(self, tmp_path):
parent = tmp_path
private_parts = []
for index in range(6):
part = f"private-{index}-" + ("x" * 150)
private_parts.append(part)
parent = parent / part
parent.mkdir()
media = parent / "handoff.txt"
media.write_text("safe handoff", encoding="utf-8")
adapter = _make_adapter()
adapter._run_cli = AsyncMock(
return_value=(
2,
"",
json.dumps(
{
"error": "network",
"message": f"upload failed for {media}: " + ("z" * 1_000),
}
),
)
)
result = await adapter.send_document(CHANNEL, str(media))
assert result.success is False
assert all(part not in result.error for part in private_parts)
assert "handoff.txt" in result.error
assert len(result.error) <= 900
@pytest.mark.asyncio
async def test_send_to_platform_live_buzz_delivers_all_media(self, monkeypatch, tmp_path):
from gateway.config import Platform