refactor(proxy): extract internal header policy (#1990)

## Description

Extracts the internal x-headroom request-header stripping policy from
`headroom.proxy.helpers` into a focused policy module. This keeps the
security-sensitive upstream filtering rule independently testable while
preserving the existing helper API used by provider handlers.

Closes #

## Type of Change

- [ ] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [x] Code refactoring (no functional changes)

## Changes Made

- Added `headroom.proxy.internal_header_policy` for strip mode
resolution and x-headroom header filtering.
- Kept `get_strip_internal_headers_mode()` and
`_strip_internal_headers()` as compatibility wrappers in `helpers.py`.
- Added direct unit tests for default/disabled/invalid modes,
case-insensitive filtering, and copy semantics.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [x] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [ ] Manual testing performed

### Test Output

```text
python -m pytest tests/test_internal_header_policy.py tests/test_header_isolation.py
29 passed in 4.89s

python -m ruff check .
All checks passed!

python -m ruff format --check .
1069 files already formatted

python -m mypy headroom --ignore-missing-imports
Success: no issues found in 410 source files

gitleaks protect --staged --no-banner --redact
no leaks found
```

## Real Behavior Proof

- Environment: Windows, Python 3.13.13
- Exact command / steps: Ran focused policy/header isolation pytest
coverage plus full ruff, ruff format check, mypy, and staged gitleaks
scan.
- Observed result: Header stripping behavior remains green end-to-end,
direct policy tests cover security-sensitive parsing/filtering rules,
and local quality/security gates pass.
- Not tested: Full repository pytest suite locally; CI covers the
broader matrix.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [ ] I have updated the CHANGELOG.md if applicable

## Screenshots (if applicable)

N/A

## Additional Notes

Documentation and changelog updates are not applicable for this internal
refactor. The default-branch Dependabot alerts reported during push are
pre-existing and unrelated to this PR.

headroom/proxy/helpers.py

@@ -39,6 +39,14 @@ from headroom.proxy.body_forwarding import (
 )
 from headroom.proxy.body_forwarding import serialize_body_canonical
 from headroom.proxy.ccr_session_tracker import SessionCcrTracker as _SessionCcrTracker
+from headroom.proxy.internal_header_policy import (
+    INTERNAL_HEADER_PREFIX,
+    STRIP_INTERNAL_HEADERS_DEFAULT,
+    STRIP_INTERNAL_HEADERS_ENV,
+    StripInternalHeadersMode,
+    resolve_strip_internal_headers_mode,
+    strip_internal_headers,
+)
 from headroom.proxy.tool_injection_config import (
     ToolInjectionStickyMode,
 )
@@ -1474,16 +1482,9 @@ def is_anthropic_auth(headers: dict[str, str]) -> bool:
 # tell its client about its own work. This helper only filters
 # request-side headers.
 
-_INTERNAL_HEADER_PREFIX = "x-headroom-"
-
-# Operator opt-in env var. ``enabled`` (default) strips internal
-# ``x-headroom-*`` headers from every upstream-bound forwarder.
-# ``disabled`` is an explicit operator opt-in for diagnostic shadow
-# tracing — NOT a fallback. Per realignment build constraint #4 the
-# behaviour is loud, configurable, and never silent.
-_STRIP_INTERNAL_HEADERS_ENV = "HEADROOM_STRIP_INTERNAL_HEADERS"
-StripInternalHeadersMode = Literal["enabled", "disabled"]
-_STRIP_INTERNAL_HEADERS_DEFAULT: StripInternalHeadersMode = "enabled"
+_INTERNAL_HEADER_PREFIX = INTERNAL_HEADER_PREFIX
+_STRIP_INTERNAL_HEADERS_ENV = STRIP_INTERNAL_HEADERS_ENV
+_STRIP_INTERNAL_HEADERS_DEFAULT = STRIP_INTERNAL_HEADERS_DEFAULT
 
 
 def get_strip_internal_headers_mode() -> StripInternalHeadersMode:
@@ -1493,14 +1494,7 @@ def get_strip_internal_headers_mode() -> StripInternalHeadersMode:
     restart. Unknown values raise loudly per the no-silent-fallback
     build constraint.
     """
-    raw = os.environ.get(_STRIP_INTERNAL_HEADERS_ENV, "").strip().lower()
-    if not raw:
-        return _STRIP_INTERNAL_HEADERS_DEFAULT
-    if raw in ("enabled", "disabled"):
-        return cast(StripInternalHeadersMode, raw)
-    raise ValueError(
-        f"Invalid {_STRIP_INTERNAL_HEADERS_ENV}={raw!r}; expected 'enabled' or 'disabled'"
-    )
+    return resolve_strip_internal_headers_mode(os.environ.get(_STRIP_INTERNAL_HEADERS_ENV))
 
 
 def _strip_internal_headers(headers: dict[str, str]) -> dict[str, str]:
@@ -1516,11 +1510,7 @@ def _strip_internal_headers(headers: dict[str, str]) -> dict[str, str]:
     is set, returns a shallow copy unchanged. That mode is for diagnostic
     shadow tracing only and is documented as a per-deploy choice.
     """
-    mode = get_strip_internal_headers_mode()
-    if mode == "disabled":
-        # Always return a copy so callers can mutate without surprise.
-        return dict(headers)
-    return {k: v for k, v in headers.items() if not k.lower().startswith(_INTERNAL_HEADER_PREFIX)}
+    return strip_internal_headers(headers, mode=get_strip_internal_headers_mode())
 
 
 def log_outbound_headers(

headroom/proxy/internal_header_policy.py

@@ -0,0 +1,40 @@
+"""Policy for stripping proxy-internal request headers before upstream calls."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Literal, cast
+
+INTERNAL_HEADER_PREFIX = "x-headroom-"
+STRIP_INTERNAL_HEADERS_ENV = "HEADROOM_STRIP_INTERNAL_HEADERS"
+StripInternalHeadersMode = Literal["enabled", "disabled"]
+STRIP_INTERNAL_HEADERS_DEFAULT: StripInternalHeadersMode = "enabled"
+
+
+def resolve_strip_internal_headers_mode(raw: str | None) -> StripInternalHeadersMode:
+    """Resolve the configured internal-header strip mode."""
+
+    normalized = (raw or "").strip().lower()
+    if not normalized:
+        return STRIP_INTERNAL_HEADERS_DEFAULT
+    if normalized in ("enabled", "disabled"):
+        return cast(StripInternalHeadersMode, normalized)
+    raise ValueError(
+        f"Invalid {STRIP_INTERNAL_HEADERS_ENV}={normalized!r}; expected 'enabled' or 'disabled'"
+    )
+
+
+def strip_internal_headers(
+    headers: Mapping[str, str],
+    *,
+    mode: StripInternalHeadersMode,
+) -> dict[str, str]:
+    """Return a copy of headers with internal x-headroom-* request headers removed."""
+
+    if mode == "disabled":
+        return dict(headers)
+    return {
+        key: value
+        for key, value in headers.items()
+        if not key.lower().startswith(INTERNAL_HEADER_PREFIX)
+    }

tests/test_internal_header_policy.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+import pytest
+
+from headroom.proxy.internal_header_policy import (
+    STRIP_INTERNAL_HEADERS_ENV,
+    resolve_strip_internal_headers_mode,
+    strip_internal_headers,
+)
+
+
+def test_resolve_strip_internal_headers_mode_defaults_to_enabled() -> None:
+    assert resolve_strip_internal_headers_mode(None) == "enabled"
+    assert resolve_strip_internal_headers_mode("  ") == "enabled"
+
+
+def test_resolve_strip_internal_headers_mode_accepts_known_values() -> None:
+    assert resolve_strip_internal_headers_mode("ENABLED") == "enabled"
+    assert resolve_strip_internal_headers_mode(" disabled ") == "disabled"
+
+
+def test_resolve_strip_internal_headers_mode_rejects_unknown_values() -> None:
+    with pytest.raises(ValueError, match=STRIP_INTERNAL_HEADERS_ENV):
+        resolve_strip_internal_headers_mode("maybe")
+
+
+def test_strip_internal_headers_removes_headroom_headers_case_insensitively() -> None:
+    headers = {
+        "Authorization": "Bearer token",
+        "x-headroom-bypass": "true",
+        "X-Headroom-User-Id": "user-1",
+        "content-type": "application/json",
+    }
+
+    stripped = strip_internal_headers(headers, mode="enabled")
+
+    assert stripped == {
+        "Authorization": "Bearer token",
+        "content-type": "application/json",
+    }
+    assert "x-headroom-bypass" in headers
+
+
+def test_strip_internal_headers_disabled_returns_copy_unchanged() -> None:
+    headers = {"x-headroom-mode": "passthrough", "content-type": "application/json"}
+
+    copied = strip_internal_headers(headers, mode="disabled")
+
+    assert copied == headers
+    assert copied is not headers