feat(egress): add preserve_auth flag to pass agent Authorization through

Adds a per-route boolean field preserve_auth (default false) that skips
the gateway's Authorization header stripping for that host. Intended for
registry endpoints like Docker Hub (registry-1.docker.io) and GHCR
(ghcr.io) where the agent must supply its own per-scope bearer token.

Threaded through ManifestEgressRoute → EgressRoute → Route, serialized
in route_to_yaml_dict, and parsed in parse_routes. The strip at
egress_addon.py now checks route.preserve_auth before popping the header.

Closes #392

bot_bottle/egress.py

@@ -146,6 +146,7 @@ def egress_manifest_routes(
             outbound_detectors=r.OutboundDetectors,
             inbound_detectors=r.InboundDetectors,
             outbound_on_match=r.OutboundOnMatch,
+            preserve_auth=r.PreserveAuth,
         ))
     return tuple(out)
 

bot_bottle/egress_addon.py

@@ -367,7 +367,10 @@ class EgressAddon:
 
         # Strip agent-set Authorization after DLP scan so smuggled tokens
         # are caught above; the route may inject gateway-owned auth below.
-        flow.request.headers.pop("authorization", None)
+        # Routes with preserve_auth=True pass the header through as-is so the
+        # agent's own credentials (e.g. registry bearer tokens) reach the upstream.
+        if route is None or not route.preserve_auth:
+            flow.request.headers.pop("authorization", None)
 
         # Build headers mapping for match evaluation
         req_headers = {k.lower(): v for k, v in flow.request.headers.items()}

bot_bottle/egress_addon_core.py

@@ -78,6 +78,7 @@ class Route:
     inbound_detectors: tuple[str, ...] | None = None
     # "" means unset → DEFAULT_OUTBOUND_ON_MATCH. See OUTBOUND_ON_MATCH_VALUES.
     outbound_on_match: str = ""
+    preserve_auth: bool = False
 
 
 LOG_OFF = 0    # no logging
@@ -308,11 +309,18 @@ def _parse_one(idx: int, raw: object) -> Route:
         idx, host, raw_dict,
     )
 
+    preserve_auth_raw = raw_dict.get("preserve_auth", False)
+    if preserve_auth_raw is not True and preserve_auth_raw is not False:
+        raise ValueError(
+            f"{label} ({host}): 'preserve_auth' must be a boolean"
+        )
+    preserve_auth: bool = preserve_auth_raw
+
     for k in raw_dict:
-        if k not in ("host", "matches", "auth_scheme", "token_env", "dlp", "git"):
+        if k not in ("host", "matches", "auth_scheme", "token_env", "dlp", "git", "preserve_auth"):
             raise ValueError(
                 f"{label} ({host}): unknown key {k!r}; accepted keys "
-                f"are 'host', 'matches', 'auth_scheme', 'token_env', 'dlp', 'git'"
+                f"are 'host', 'matches', 'auth_scheme', 'token_env', 'dlp', 'git', 'preserve_auth'"
             )
 
     return Route(
@@ -324,6 +332,7 @@ def _parse_one(idx: int, raw: object) -> Route:
         outbound_detectors=outbound_detectors,
         inbound_detectors=inbound_detectors,
         outbound_on_match=outbound_on_match,
+        preserve_auth=preserve_auth,
     )
 
 
@@ -376,6 +385,8 @@ def route_to_yaml_dict(r: Route) -> dict[str, object]:
         dlp["outbound_on_match"] = r.outbound_on_match
     if dlp:
         d["dlp"] = dlp
+    if r.preserve_auth:
+        d["preserve_auth"] = True
     return d
 
 

bot_bottle/manifest_egress.py

@@ -71,6 +71,7 @@ class ManifestEgressRoute:
     OutboundDetectors: tuple[str, ...] | None = None
     InboundDetectors: tuple[str, ...] | None = None
     OutboundOnMatch: str = ""
+    PreserveAuth: bool = False
 
     @classmethod
     def from_dict(cls, bottle_name: str, idx: int, raw: object) -> "ManifestEgressRoute":
@@ -190,11 +191,22 @@ class ManifestEgressRoute:
                         f"only 'fetch' is accepted"
                     )
 
+        # --- preserve_auth ---
+        preserve_auth = False
+        if "preserve_auth" in d:
+            raw_preserve_auth = d.get("preserve_auth")
+            if not isinstance(raw_preserve_auth, bool):
+                raise ManifestError(
+                    f"{label} preserve_auth must be a boolean "
+                    f"(was {type(raw_preserve_auth).__name__})"
+                )
+            preserve_auth = raw_preserve_auth
+
         for k in d:
-            if k not in ("host", "matches", "auth", "role", "dlp", "git"):
+            if k not in ("host", "matches", "auth", "role", "dlp", "git", "preserve_auth"):
                 raise ManifestError(
                     f"{label} has unknown key {k!r}; accepted keys are "
-                    f"'host', 'matches', 'auth', 'role', 'dlp', 'git'"
+                    f"'host', 'matches', 'auth', 'role', 'dlp', 'git', 'preserve_auth'"
                 )
 
         return cls(
@@ -207,6 +219,7 @@ class ManifestEgressRoute:
             OutboundDetectors=outbound_detectors,
             InboundDetectors=inbound_detectors,
             OutboundOnMatch=outbound_on_match,
+            PreserveAuth=preserve_auth,
         )