[BUG] security-guidance: model reviews 401 behind an LLM gateway because ANTHROPIC_CUSTOM_HEADERS is never sent

One more reference, for the setup this breaks. Other LLM gateways — Subscriptions and gateways documents it as supported:

Setting only that variable, without a gateway credential, doesn't replace the subscription. Requests still route through the gateway, but a saved claude.ai login remains the active credential, so its usage limits and billing apply.

So ANTHROPIC_BASE_URL plus a subscription sign-in is a documented configuration, and in it the gateway can only identify the caller through ANTHROPIC_CUSTOM_HEADERS. Setting a gateway credential variable instead would, per the same section, bill the traffic per token.

Added by Claude Code (agent) at the reporter's request.

Thanks @jrdeck for the report and proposed fix. I reproduced the same missing-header behavior in this repository's public source at 79babc372d64101f981bd2b52c3dbe588596dc56 and prepared an additive patch plus regression tests below.

Scope: this checkout's security-guidance manifest says 2.0.0; your report describes marketplace 2.0.10. The shared problem is _build_auth_headers() in the raw HTTP review path. This checkout does not have the marketplace /v1/models lookup, so this patch/test does not cover that lookup or claim to validate the entire marketplace version.

Validation: Python 3.9.6, real _call_claude() → urllib → a loopback HTTP gateway. With a required x-litellm-api-key, the original code returns None and records HTTP 401 for both API-key and token authentication. With this eight-line production change, both receive and parse the gateway's synthetic success response. Of seven test methods, six fail against the original (seven assertion failures including subtests); all seven pass after the change. The unset/empty-header control passes in both versions. I also applied the complete patch to a clean copy of the pinned source: git apply --check and all seven tests pass.

Coverage also includes API-key-first behavior when both credentials exist; 401 → token fallback and its sticky preference; custom Authorization alongside API-key auth; case-insensitive collisions with already generated auth/protocol headers; newline/CRLF separation, whitespace, empty values and values containing colons; duplicate custom names differing in case; and header construction with no base-URL override. Each client subprocess receives only fake credentials and isolated environment/HOME. No SDK was installed and no model or external gateway was called.

Merge semantics: following your additive suggestion, headers already generated for the current request win case-insensitively, and the first custom occurrence wins. Other header names remain allowed, including Authorization when the selected auth is x-api-key; the fallback still regenerates and preserves its chosen auth. This collision policy is a proposed plugin design choice, not a claim about the CLI's collision precedence. The environment-variable reference and gateway guide describe newline-separated additional headers without a custom-endpoint-only restriction, so the merge applies whenever the variable is set. The patch adds no header-value logging. Invalid HTTP characters are left to urllib validation; it does not reproduce the CLI's configuration diagnostics.

Run from the repository root:

python3 -m unittest discover -s plugins/security-guidance/tests -v

External PR creation is currently restricted. Would a maintainer be willing to take this patch, or indicate the preferred submission route? Credit for identifying the issue and the original fix belongs to @jrdeck; this follow-up contributes the public-repository reproduction and tests. I have not opened a PR or pushed a branch.

<details>
<summary>Complete patch against 79babc372d64101f981bd2b52c3dbe588596dc56</summary>

--- a/plugins/security-guidance/hooks/llm.py
+++ b/plugins/security-guidance/hooks/llm.py
@@ -200,6 +200,14 @@
     else:
         headers["x-api-key"] = ANTHROPIC_API_KEY
     headers["anthropic-beta"] = ",".join(betas)
+    # Add gateway/routing headers while keeping this request's auth and protocol headers.
+    header_names = {name.lower() for name in headers}
+    for line in os.environ.get("ANTHROPIC_CUSTOM_HEADERS", "").splitlines():
+        name, sep, value = line.partition(":")
+        name = name.strip()
+        if sep and name and name.lower() not in header_names:
+            headers[name] = value.strip()
+            header_names.add(name.lower())
     return headers
 
 
--- /dev/null
+++ b/plugins/security-guidance/tests/test_custom_headers.py
@@ -0,0 +1,148 @@
+"""Exercise the raw HTTP review path against a loopback gateway (no API calls)."""
+import json
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import threading
+import unittest
+
+HOOKS = Path(__file__).resolve().parents[1] / "hooks"
+CALL = """
+import json, llm
+results = [llm._call_claude('test-only prompt', {'type': 'object'}, thinking_budget=0)
+           for _ in range(int(__import__('os').environ.get('TEST_CALLS', '1'))) ]
+print(json.dumps({'results': results, 'error': llm._last_call_claude_http_error}))
+"""
+
+
+class CustomHeadersTest(unittest.TestCase):
+    def setUp(self):
+        self.scratch = tempfile.TemporaryDirectory()
+        self.addCleanup(self.scratch.cleanup)
+        self.requests = []
+        self.require_gateway = False
+        self.reject_key = False
+        test = self
+
+        class Gateway(BaseHTTPRequestHandler):
+            def do_POST(self):
+                headers = {key.lower(): value for key, value in self.headers.items()}
+                payload = json.loads(self.rfile.read(int(self.headers['Content-Length'])))
+                test.requests.append((self.path, headers, payload))
+                authorized = not test.require_gateway or headers.get('x-litellm-api-key') == 'Bearer gateway-test-key'
+                if test.reject_key and 'x-api-key' in headers:
+                    authorized = False
+                self.send_response(200 if authorized else 401)
+                self.send_header('Content-Type', 'application/json')
+                self.end_headers()
+                body = {'content': [{'type': 'text', 'text': '{"reviewed": true}'}]} if authorized else {'error': 'test gateway rejected request'}
+                self.wfile.write(json.dumps(body).encode())
+
+            def log_message(self, *args):
+                pass
+
+        self.server = ThreadingHTTPServer(('127.0.0.1', 0), Gateway)
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+        self.thread.start()
+        self.addCleanup(self.server.server_close)
+        self.addCleanup(self.thread.join)
+        self.addCleanup(self.server.shutdown)
+
+    def run_client(self, custom=None, key='api-test-key', token='', calls=1, script=CALL):
+        # Isolate credentials, proxies, provider flags, SDK settings, and user HOME.
+        env = {
+            'HOME': self.scratch.name,
+            'PYTHONPATH': str(HOOKS),
+            'PYTHONDONTWRITEBYTECODE': '1',
+            'ANTHROPIC_API_KEY': key,
+            'ANTHROPIC_AUTH_TOKEN': token,
+            'ANTHROPIC_BASE_URL': f'http://127.0.0.1:{self.server.server_port}',
+            'TEST_CALLS': str(calls),
+        }
+        if custom is not None:
+            env['ANTHROPIC_CUSTOM_HEADERS'] = custom
+        result = subprocess.run([sys.executable, '-c', script], env=env,
+                                cwd=self.scratch.name, capture_output=True, text=True, timeout=15)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(result.stderr, '')
+        return json.loads(result.stdout)
+
+    def assert_reviewed(self, result, calls=1):
+        self.assertEqual(result, {'results': [{'reviewed': True}] * calls, 'error': None})
+        for path, _, payload in self.requests:
+            self.assertEqual(path, '/v1/messages')
+            self.assertEqual(payload['messages'], [{'role': 'user', 'content': 'test-only prompt'}])
+
+    def test_api_key_review_authenticates_to_gateway(self):
+        self.require_gateway = True
+        self.assert_reviewed(self.run_client('x-litellm-api-key: Bearer gateway-test-key'))
+        self.assertEqual(self.requests[0][1]['x-api-key'], 'api-test-key')
+        self.assertNotIn('authorization', self.requests[0][1])
+
+    def test_token_review_authenticates_to_gateway(self):
+        self.require_gateway = True
+        self.assert_reviewed(self.run_client('x-litellm-api-key: Bearer gateway-test-key', key='', token='token-test-key'))
+        self.assertEqual(self.requests[0][1]['authorization'], 'Bearer token-test-key')
+        self.assertNotIn('x-api-key', self.requests[0][1])
+
+    def test_gateway_header_survives_401_fallback_and_sticky_token(self):
+        self.require_gateway = True
+        self.reject_key = True
+        self.assert_reviewed(self.run_client('x-litellm-api-key: Bearer gateway-test-key\nAuthorization: Bearer gateway-auth', token='token-test-key', calls=2), calls=2)
+        self.assertEqual(len(self.requests), 3)
+        self.assertEqual(self.requests[0][1]['x-api-key'], 'api-test-key')
+        self.assertEqual(self.requests[0][1]['authorization'], 'Bearer gateway-auth')
+        for _, headers, _ in self.requests[1:]:
+            self.assertEqual(headers['authorization'], 'Bearer token-test-key')
+            self.assertNotIn('x-api-key', headers)
+        for _, headers, _ in self.requests:
+            self.assertEqual(headers['x-litellm-api-key'], 'Bearer gateway-test-key')
+
+    def test_multiline_headers_preserve_colons_and_empty_values(self):
+        self.assert_reviewed(self.run_client('\n X-Tenant : example \r\nX-Route: http://gateway:8080/a\nX-Empty:\nx-tenant: ignored-duplicate\nmissing-colon\n: missing-name\n'))
+        headers = self.requests[0][1]
+        self.assertEqual(headers.get('x-tenant'), 'example')
+        self.assertEqual(headers.get('x-route'), 'http://gateway:8080/a')
+        self.assertEqual(headers.get('x-empty'), '')
+        self.assertNotIn('missing-colon', headers)
+
+    def test_custom_headers_keep_selected_auth_and_protocol_headers(self):
+        custom = '\n'.join(['aUtHoRiZaTiOn: Bearer override', 'X-API-KEY: override',
+                            'CONTENT-TYPE: text/plain', 'Anthropic-Version: invalid',
+                            'ANTHROPIC-BETA: invalid', 'X-Tenant: example'])
+        for key, token in [('api-test-key', ''), ('', 'token-test-key')]:
+            with self.subTest(key=bool(key)):
+                self.assert_reviewed(self.run_client(custom, key=key, token=token))
+                headers = self.requests[-1][1]
+                self.assertEqual(headers.get('x-tenant'), 'example')
+                self.assertEqual(headers['content-type'], 'application/json')
+                self.assertEqual(headers['anthropic-version'], '2023-06-01')
+                self.assertIn('structured-outputs-2025-11-13', headers['anthropic-beta'])
+                if key:
+                    self.assertEqual(headers['x-api-key'], key)
+                    self.assertEqual(headers['authorization'], 'Bearer override')
+                else:
+                    self.assertEqual(headers['authorization'], 'Bearer ' + token)
+                    self.assertEqual(headers['x-api-key'], 'override')
+                    self.assertIn('oauth-2025-04-20', headers['anthropic-beta'])
+
+    def test_unset_and_empty_headers_preserve_existing_auth(self):
+        for custom in [None, '', '\n \n']:
+            with self.subTest(custom=custom):
+                self.assert_reviewed(self.run_client(custom, token='token-test-key'))
+                headers = self.requests[-1][1]
+                self.assertEqual(headers['x-api-key'], 'api-test-key')
+                self.assertNotIn('authorization', headers)
+                self.assertNotIn('x-litellm-api-key', headers)
+
+    def test_custom_headers_are_not_gated_on_base_url(self):
+        # Only build headers for the default URL; never send a request to it.
+        script = "import os; os.environ.pop('ANTHROPIC_BASE_URL'); import json, llm; print(json.dumps(llm._build_auth_headers(False)))"
+        headers = self.run_client('X-Tenant: example', script=script)
+        self.assertEqual(headers.get('X-Tenant'), 'example')
+
+
+if __name__ == '__main__':
+    unittest.main()

</details>

Hi, I'm picking this up. I'll make _build_auth_headers() merge ANTHROPIC_CUSTOM_HEADERS (one Name: Value per line, never overriding the auth/protocol headers the plugin sets) so gateway calls carry the same headers Claude Code sends. — Sahil Jagtap

Fix is up: https://github.com/sahiljagtap08/claude-code/pull/2 (branch sahiljagtap/security-guidance-custom-headers, one commit on current main). PRs from forks are currently rejected on this repo, so it's on my fork for a maintainer to cherry-pick:

git fetch https://github.com/sahiljagtap08/claude-code.git sahiljagtap/security-guidance-custom-headers
git cherry-pick FETCH_HEAD

_build_auth_headers() now merges ANTHROPIC_CUSTOM_HEADERS (first colon splits name/value, \n or \r\n separated). Headers the plugin already set for the request win case-insensitively, so a custom Authorization: line can't replace the credential the 401 fallback chose. Verified against a loopback gateway that 401s without x-litellm-api-key: request succeeds with the variable set, still 401s without it. — Sahil Jagtap