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,
         )
 
 

tests/unit/test_egress_addon_request_flow.py

@@ -413,6 +413,28 @@ class TestAuthInjection(unittest.TestCase):
         assert flow.response is not None
         self.assertEqual(403, flow.response.status_code)
 
+    def test_preserve_auth_passes_agent_token_through(self) -> None:
+        route = Route(host="registry-1.docker.io", preserve_auth=True)
+        addon = _addon(Config(routes=(route,)))
+        flow = _Flow(_Request(
+            host="registry-1.docker.io",
+            headers={"authorization": "Bearer agent-registry-token"},
+        ))
+        _run_request(addon, flow)
+        self.assertEqual("Bearer agent-registry-token", flow.request.headers.get("authorization"))
+        self.assertIsNone(flow.response)
+
+    def test_default_route_strips_agent_auth(self) -> None:
+        route = Route(host="registry-1.docker.io")
+        addon = _addon(Config(routes=(route,)))
+        flow = _Flow(_Request(
+            host="registry-1.docker.io",
+            headers={"authorization": "Bearer agent-registry-token"},
+        ))
+        _run_request(addon, flow)
+        self.assertIsNone(flow.request.headers.get("authorization"))
+        self.assertIsNone(flow.response)
+
 
 # ---------------------------------------------------------------------------
 # git push / fetch over HTTPS

tests/unit/test_manifest_egress.py

@@ -458,6 +458,24 @@ class TestRole(unittest.TestCase):
             _bottle([{"host": "x.example", "role": ["x", 42]}])
 
 
+class TestPreserveAuth(unittest.TestCase):
+    def test_omitted_defaults_false(self):
+        b = _bottle([{"host": "registry-1.docker.io"}])
+        self.assertFalse(b.egress.routes[0].PreserveAuth)
+
+    def test_true_accepted(self):
+        b = _bottle([{"host": "registry-1.docker.io", "preserve_auth": True}])
+        self.assertTrue(b.egress.routes[0].PreserveAuth)
+
+    def test_false_accepted(self):
+        b = _bottle([{"host": "registry-1.docker.io", "preserve_auth": False}])
+        self.assertFalse(b.egress.routes[0].PreserveAuth)
+
+    def test_non_bool_rejected(self):
+        with self.assertRaises(ManifestError):
+            _bottle([{"host": "registry-1.docker.io", "preserve_auth": "yes"}])
+
+
 class TestPipelockKeyRejected(unittest.TestCase):
     def test_pipelock_key_rejected_as_unknown(self):
         with self.assertRaises(ManifestError):