didericis-claude
5ad3449e3b
Install bot_bottle via pip in Dockerfile.gateway instead of COPYing individual .py files flat under /app/. This eliminates the try/except import shims in egress_addon_core, dlp_detectors, egress_addon, supervise, supervise_server, and git_http_backend that existed only to support the flat-bundle layout. Adds bot_bottle/constants.py as a single source of truth for IDENTITY_HEADER and GIT_GATE_TIMEOUT_SECS, removing the duplicated literal definitions in egress_addon.py, supervise_server.py, git_http_backend.py, and git_gate_render.py. Test files updated to match: test_supervise_server.py drops the sys.path.insert hack in favour of direct package imports; the egress_addon test shims no longer pre-populate sys.modules with a bare egress_addon_core alias.
272 lines
9.6 KiB
Python
272 lines
9.6 KiB
Python
"""Unit: LOG_FULL credential redaction in _log_request / _log_response (issue #257).
|
|
|
|
egress_addon.py is gateway-only code that depends on mitmproxy, which is
|
|
not installed on the host. This file pre-populates sys.modules with the
|
|
minimum mocks needed so EgressAddon can be imported and tested without the
|
|
real mitmproxy package."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import types
|
|
import unittest
|
|
from io import StringIO
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# mitmproxy stub — must run before importing egress_addon
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _ensure_shims() -> None:
|
|
# Resolver-only egress: importing the module builds the `addons` singleton,
|
|
# which requires an orchestrator URL. These tests exercise the log helpers
|
|
# on a __new__-built addon, so the value is never dialed.
|
|
os.environ.setdefault("BOT_BOTTLE_ORCHESTRATOR_URL", "http://127.0.0.1:0")
|
|
if "mitmproxy" not in sys.modules:
|
|
_mm = types.ModuleType("mitmproxy")
|
|
_mh = types.ModuleType("mitmproxy.http")
|
|
setattr(_mm, "http", _mh)
|
|
sys.modules["mitmproxy"] = _mm
|
|
sys.modules["mitmproxy.http"] = _mh
|
|
|
|
|
|
_ensure_shims()
|
|
|
|
from bot_bottle.egress_addon import EgressAddon # noqa: E402 (import after shims)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _addon() -> EgressAddon:
|
|
"""A bare EgressAddon for exercising the log helpers directly. The redaction
|
|
log methods take their env explicitly, so no resolver/config wiring is
|
|
needed here."""
|
|
return EgressAddon.__new__(EgressAddon)
|
|
|
|
|
|
class _Headers:
|
|
def __init__(self, d: dict[str, str]) -> None:
|
|
self._d = d
|
|
|
|
def items(self) -> list[tuple[str, str]]:
|
|
return list(self._d.items())
|
|
|
|
|
|
class _Request:
|
|
def __init__(
|
|
self,
|
|
host: str = "api.example.com",
|
|
method: str = "POST",
|
|
path: str = "/v1/messages",
|
|
headers: dict[str, str] | None = None,
|
|
body: str = "",
|
|
) -> None:
|
|
self.pretty_host = host
|
|
self.method = method
|
|
self.path = path
|
|
self.headers = _Headers(headers or {})
|
|
self._body = body
|
|
|
|
def get_text(self, *, strict: bool = True) -> str:
|
|
return self._body
|
|
|
|
|
|
class _Response:
|
|
def __init__(
|
|
self,
|
|
status_code: int = 200,
|
|
headers: dict[str, str] | None = None,
|
|
body: str = "",
|
|
) -> None:
|
|
self.status_code = status_code
|
|
self.headers = _Headers(headers or {})
|
|
self._body = body
|
|
|
|
def get_text(self, *, strict: bool = True) -> str:
|
|
return self._body
|
|
|
|
|
|
class _Flow:
|
|
def __init__(
|
|
self,
|
|
request: _Request | None = None,
|
|
response: _Response | None = None,
|
|
) -> None:
|
|
self.request = request or _Request()
|
|
self.response = response or _Response()
|
|
|
|
|
|
def _log_request(addon: EgressAddon, flow: _Flow) -> dict[str, Any]:
|
|
buf = StringIO()
|
|
with patch("sys.stderr", buf):
|
|
addon._log_request(flow, os.environ) # type: ignore[arg-type]
|
|
return json.loads(buf.getvalue())
|
|
|
|
|
|
def _log_response(addon: EgressAddon, flow: _Flow) -> dict[str, Any]:
|
|
buf = StringIO()
|
|
with patch("sys.stderr", buf):
|
|
addon._log_response(flow, os.environ) # type: ignore[arg-type]
|
|
return json.loads(buf.getvalue())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _log_request — authorization header stripped
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestLogRequestAuthorizationStripped(unittest.TestCase):
|
|
def test_lowercase_authorization_excluded(self) -> None:
|
|
addon = _addon()
|
|
flow = _Flow(request=_Request(headers={"authorization": "Bearer sk-real-secret"}))
|
|
entry = _log_request(addon, flow)
|
|
self.assertNotIn("authorization", entry["headers"])
|
|
|
|
def test_titlecase_authorization_excluded(self) -> None:
|
|
addon = _addon()
|
|
flow = _Flow(request=_Request(headers={"Authorization": "Bearer sk-real-secret"}))
|
|
entry = _log_request(addon, flow)
|
|
self.assertNotIn("Authorization", entry["headers"])
|
|
self.assertNotIn("authorization", entry["headers"])
|
|
|
|
def test_non_auth_headers_retained(self) -> None:
|
|
addon = _addon()
|
|
flow = _Flow(request=_Request(headers={
|
|
"authorization": "Bearer sk-real-secret",
|
|
"content-type": "application/json",
|
|
}))
|
|
entry = _log_request(addon, flow)
|
|
self.assertIn("content-type", entry["headers"])
|
|
self.assertEqual("application/json", entry["headers"]["content-type"])
|
|
|
|
def test_no_authorization_header_logs_all_others(self) -> None:
|
|
addon = _addon()
|
|
flow = _Flow(request=_Request(headers={"x-request-id": "abc"}))
|
|
entry = _log_request(addon, flow)
|
|
self.assertEqual({"x-request-id": "abc"}, entry["headers"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _log_request — body redaction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
_OPENAI_KEY = "sk-" + "A" * 48
|
|
|
|
|
|
class TestLogRequestBodyRedacted(unittest.TestCase):
|
|
def test_token_pattern_in_body_scrubbed(self) -> None:
|
|
addon = _addon()
|
|
flow = _Flow(request=_Request(body=f"key={_OPENAI_KEY}"))
|
|
entry = _log_request(addon, flow)
|
|
self.assertNotIn(_OPENAI_KEY, entry["body"])
|
|
self.assertIn("********", entry["body"])
|
|
|
|
def test_provisioned_secret_in_body_scrubbed(self) -> None:
|
|
addon = _addon()
|
|
secret = "provisioned-egress-secret-xyz"
|
|
flow = _Flow(request=_Request(body=f"token={secret}"))
|
|
with patch.dict("os.environ", {"EGRESS_TOKEN_0": secret}):
|
|
entry = _log_request(addon, flow)
|
|
self.assertNotIn(secret, entry["body"])
|
|
self.assertIn("********", entry["body"])
|
|
|
|
def test_clean_body_preserved(self) -> None:
|
|
addon = _addon()
|
|
payload = '{"model": "claude-3", "max_tokens": 1024}'
|
|
flow = _Flow(request=_Request(body=payload))
|
|
entry = _log_request(addon, flow)
|
|
self.assertEqual(payload, entry["body"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _log_request — non-authorization header value redaction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestLogRequestHeaderValuesRedacted(unittest.TestCase):
|
|
def test_token_in_custom_header_scrubbed(self) -> None:
|
|
addon = _addon()
|
|
flow = _Flow(request=_Request(headers={"x-api-key": _OPENAI_KEY}))
|
|
entry = _log_request(addon, flow)
|
|
self.assertNotIn(_OPENAI_KEY, entry["headers"].get("x-api-key", ""))
|
|
self.assertIn("********", entry["headers"].get("x-api-key", ""))
|
|
|
|
def test_clean_header_value_preserved(self) -> None:
|
|
addon = _addon()
|
|
flow = _Flow(request=_Request(headers={"accept": "application/json"}))
|
|
entry = _log_request(addon, flow)
|
|
self.assertEqual("application/json", entry["headers"]["accept"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _log_response — body redaction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestLogResponseBodyRedacted(unittest.TestCase):
|
|
def test_token_pattern_in_response_body_scrubbed(self) -> None:
|
|
addon = _addon()
|
|
flow = _Flow(
|
|
request=_Request(),
|
|
response=_Response(body=f'{{"key": "{_OPENAI_KEY}"}}'),
|
|
)
|
|
entry = _log_response(addon, flow)
|
|
self.assertNotIn(_OPENAI_KEY, entry["body"])
|
|
self.assertIn("********", entry["body"])
|
|
|
|
def test_provisioned_secret_in_response_body_scrubbed(self) -> None:
|
|
addon = _addon()
|
|
secret = "provisioned-egress-secret-xyz"
|
|
flow = _Flow(
|
|
request=_Request(),
|
|
response=_Response(body=f'{{"token": "{secret}"}}'),
|
|
)
|
|
with patch.dict("os.environ", {"EGRESS_TOKEN_0": secret}):
|
|
entry = _log_response(addon, flow)
|
|
self.assertNotIn(secret, entry["body"])
|
|
self.assertIn("********", entry["body"])
|
|
|
|
def test_clean_response_body_preserved(self) -> None:
|
|
addon = _addon()
|
|
flow = _Flow(request=_Request(), response=_Response(body='{"result": "ok"}'))
|
|
entry = _log_response(addon, flow)
|
|
self.assertEqual('{"result": "ok"}', entry["body"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _log_response — response header value redaction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestLogResponseHeaderValuesRedacted(unittest.TestCase):
|
|
def test_token_in_response_header_scrubbed(self) -> None:
|
|
addon = _addon()
|
|
flow = _Flow(
|
|
request=_Request(),
|
|
response=_Response(headers={"set-cookie": f"token={_OPENAI_KEY}"}),
|
|
)
|
|
entry = _log_response(addon, flow)
|
|
cookie_val = entry["headers"].get("set-cookie", "")
|
|
self.assertNotIn(_OPENAI_KEY, cookie_val)
|
|
self.assertIn("********", cookie_val)
|
|
|
|
def test_clean_response_header_preserved(self) -> None:
|
|
addon = _addon()
|
|
flow = _Flow(
|
|
request=_Request(),
|
|
response=_Response(headers={"content-type": "application/json"}),
|
|
)
|
|
entry = _log_response(addon, flow)
|
|
self.assertEqual("application/json", entry["headers"]["content-type"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|