feat(egress): replace log bool with integer log levels (0/1/2)

Level 0 (off, default): no stderr output beyond boot line.
Level 1 (blocks): each block/warn emitted as JSON with reason and
request context (host, method, path, response_status for inbound).
Level 2 (full): level-1 events + egress_request and egress_response
JSON lines for every forwarded connection.

Block logging at level 1+ replaces the previous plain-text stderr write.
DLP warn logging is also gated on level 1+. All block call sites now pass
_req_ctx(flow) so the blocked request is visible in the log entry.
Boot message shows log level label (off/blocks/full).

Adds PRD 0053 documenting wire format, manifest format, and all log event
shapes.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

bot_bottle/egress.py

@@ -62,7 +62,7 @@ class EgressPlan:
     egress_network: str = ""
     mitmproxy_ca_host_path: Path = Path()
     mitmproxy_ca_cert_only_host_path: Path = Path()
-    log: bool = False
+    log: int = 0
 
 
 def egress_manifest_routes(
@@ -192,11 +192,11 @@ def _route_to_yaml_fields(r: Route) -> dict[str, object]:
 def egress_render_routes(
     routes: tuple[EgressRoute, ...],
     *,
-    log: bool = False,
+    log: int = 0,
 ) -> str:
     lines: list[str] = []
     if log:
-        lines.append("log: true")
+        lines.append(f"log: {log}")
     lines.append("routes:")
     if not routes:
         lines[-1] = "routes: []"

bot_bottle/egress_addon.py

@@ -15,6 +15,8 @@ from pathlib import Path
 from mitmproxy import http  # type: ignore[import-not-found]
 
 from egress_addon_core import (  # type: ignore[import-not-found]
+    LOG_BLOCKS,
+    LOG_FULL,
     Config,
     Route,
     decide,
@@ -51,10 +53,11 @@ class EgressAddon:
                 self.config = Config(routes=())
             return
         self.config = new_config
+        log_label = ("off", "blocks", "full")[self.config.log]
         sys.stderr.write(
             f"egress: loaded {len(self.config.routes)} route(s): "
             f"{', '.join(r.host for r in self.config.routes)}"
-            f"{' [log=on]' if self.config.log else ''}\n"
+            f" [log={log_label}]\n"
         )
 
     def _install_sighup(self) -> None:
@@ -84,8 +87,24 @@ class EgressAddon:
             {"Content-Type": "text/plain; charset=utf-8"},
         )
 
-    def _block(self, flow: http.HTTPFlow, reason: str) -> None:
-        sys.stderr.write(f"{reason}\n")
+    def _req_ctx(self, flow: http.HTTPFlow) -> dict[str, object]:
+        return {
+            "host": flow.request.pretty_host,
+            "method": flow.request.method,
+            "path": flow.request.path,
+        }
+
+    def _block(
+        self,
+        flow: http.HTTPFlow,
+        reason: str,
+        ctx: dict[str, object] | None = None,
+    ) -> None:
+        if self.config.log >= LOG_BLOCKS:
+            entry: dict[str, object] = {"event": "egress_block", "reason": reason}
+            if ctx:
+                entry.update(ctx)
+            sys.stderr.write(json.dumps(entry) + "\n")
         flow.response = http.Response.make(
             403,
             reason.encode("utf-8"),
@@ -135,7 +154,11 @@ class EgressAddon:
                 scan_text = auth_header + "\n" + body
             dlp_result = scan_outbound(route, scan_text, os.environ)
             if dlp_result is not None and dlp_result.severity == "block":
-                self._block(flow, f"egress DLP: {dlp_result.reason}")
+                self._block(
+                    flow,
+                    f"egress DLP: {dlp_result.reason}",
+                    ctx=self._req_ctx(flow),
+                )
                 return
 
         # Strip inbound Authorization — agent cannot smuggle tokens.
@@ -147,6 +170,7 @@ class EgressAddon:
                 "egress: git push over HTTPS is not supported; "
                 "use the bottle.git SSH path (gitleaks-scanned by "
                 "git-gate's pre-receive hook).",
+                ctx=self._req_ctx(flow),
             )
             return
 
@@ -163,13 +187,13 @@ class EgressAddon:
         )
 
         if decision.action == "block":
-            self._block(flow, decision.reason)
+            self._block(flow, decision.reason, ctx=self._req_ctx(flow))
             return
 
         if decision.inject_authorization is not None:
             flow.request.headers["authorization"] = decision.inject_authorization
 
-        if self.config.log:
+        if self.config.log >= LOG_FULL:
             self._log_request(flow)
 
     def response(self, flow: http.HTTPFlow) -> None:
@@ -179,7 +203,7 @@ class EgressAddon:
             return
         if flow.response is None:
             return
-        if self.config.log:
+        if self.config.log >= LOG_FULL:
             self._log_response(flow)
         body = flow.response.get_text(strict=False) or ""
         if not body:
@@ -187,10 +211,18 @@ class EgressAddon:
         result = scan_inbound(route, body)
         if result is None:
             return
+        resp_ctx = {**self._req_ctx(flow), "response_status": flow.response.status_code}
         if result.severity == "block":
-            self._block(flow, f"egress DLP: {result.reason}")
-        elif result.severity == "warn":
-            sys.stderr.write(f"egress DLP warn: {result.reason}\n")
+            self._block(flow, f"egress DLP: {result.reason}", ctx=resp_ctx)
+        elif result.severity == "warn" and self.config.log >= LOG_BLOCKS:
+            sys.stderr.write(
+                json.dumps({
+                    "event": "egress_warn",
+                    "reason": f"egress DLP: {result.reason}",
+                    **resp_ctx,
+                })
+                + "\n"
+            )
 
 
 addons = [EgressAddon()]

bot_bottle/egress_addon_core.py

@@ -70,10 +70,15 @@ class Route:
     inbound_detectors: tuple[str, ...] | None = None
 
 
+LOG_OFF = 0    # no logging
+LOG_BLOCKS = 1  # log block/warn events with request context
+LOG_FULL = 2    # log block/warn events + full request and response bodies
+
+
 @dataclass(frozen=True)
 class Config:
     routes: tuple[Route, ...]
-    log: bool = False
+    log: int = LOG_OFF
 
 
 @dataclass(frozen=True)
@@ -341,14 +346,17 @@ def load_routes(text: str) -> tuple[Route, ...]:
 
 
 def parse_config(payload: object) -> "Config":
-    """Parse a full egress config payload (top-level log flag + routes)."""
+    """Parse a full egress config payload (top-level log level + routes)."""
     if not isinstance(payload, dict):
         raise ValueError("routes payload: top-level must be an object")
     payload_dict: dict[str, object] = typing.cast(dict[str, object], payload)
 
-    log_raw: object = payload_dict.get("log", False)
-    if not isinstance(log_raw, bool):
-        raise ValueError("routes payload: 'log' must be true or false")
+    log_raw: object = payload_dict.get("log", LOG_OFF)
+    if log_raw is True or log_raw is False or not isinstance(log_raw, int) \
+            or log_raw not in (LOG_OFF, LOG_BLOCKS, LOG_FULL):
+        raise ValueError(
+            f"routes payload: 'log' must be {LOG_OFF}, {LOG_BLOCKS}, or {LOG_FULL}"
+        )
 
     routes = parse_routes(payload)
     return Config(routes=routes, log=log_raw)
@@ -564,6 +572,9 @@ def scan_inbound(
 
 
 __all__ = [
+    "LOG_BLOCKS",
+    "LOG_FULL",
+    "LOG_OFF",
     "Config",
     "Decision",
     "HeaderMatch",

bot_bottle/manifest_egress.py

@@ -346,10 +346,13 @@ def _parse_dlp_block(
     return outbound, inbound
 
 
+LOG_LEVELS = frozenset({0, 1, 2})
+
+
 @dataclass(frozen=True)
 class EgressConfig:
     routes: tuple[EgressRoute, ...] = ()
-    Log: bool = False
+    Log: int = 0
 
     @classmethod
     def from_dict(cls, bottle_name: str, raw: object) -> "EgressConfig":
@@ -368,10 +371,11 @@ class EgressConfig:
                 for i, entry in enumerate(routes_list)
             )
             validate_egress_routes(bottle_name, routes)
-        log_raw = d.get("log", False)
-        if not isinstance(log_raw, bool):
+        log_raw = d.get("log", 0)
+        if isinstance(log_raw, bool) or not isinstance(log_raw, int) \
+                or log_raw not in LOG_LEVELS:
             raise ManifestError(
-                f"bottle '{bottle_name}' egress.log must be true or false"
+                f"bottle '{bottle_name}' egress.log must be 0, 1, or 2"
             )
         for k in d:
             if k not in ("routes", "log"):