mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-10 14:27:00 -04:00
cb38f79377
## Description Extract the trusted forwarded-header trust policy into `headroom.proxy.forwarded_policy`, leaving `forwarded_headers` as the FastAPI/request-state adapter. This makes CIDR parsing, peer trust, leftmost forwarded-for handling, and rejection decisions deterministic and directly testable without request/logging side effects. 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 `ForwardedHeaderInputs` and `ForwardedHeaderResolution` as pure policy value objects. - Moved CIDR parsing, IP normalization, trust membership, header splitting, and forwarded-header resolution into `headroom.proxy.forwarded_policy`. - Kept `headroom.proxy.forwarded_headers` as the request adapter with the same public API and compatibility helper names. - Added direct tests for trusted, rejected, direct-client, IPv4-mapped IPv6, and leftmost `X-Forwarded-For` policy behavior. - Included the LiteLLM callback hook compatibility shim needed for repo-wide mypy on branches based on `main`. ## 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_forwarded_policy.py tests/test_forwarded_headers.py tests/test_litellm_callback.py tests/test_compress_api.py::TestLiteLLMCallback -q 51 passed in 6.36s python -m ruff check . All checks passed! python -m ruff format --check . 1095 files already formatted python -m mypy headroom --ignore-missing-imports Success: no issues found in 409 source files ``` ## Real Behavior Proof - Environment: Windows, Python 3.13.13, local worktree `C:\git\headroom-pr-slice8`. - Exact command / steps: Ran the focused pytest suite plus repo-wide Ruff, format check, and mypy commands listed above. - Observed result: The existing request-facing forwarded-header behavior remains covered by `tests/test_forwarded_headers.py`, while the extracted pure policy is covered by `tests/test_forwarded_policy.py`. - Not tested: Full test suite locally; CI will run the full 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 LiteLLM shim is repeated here because this branch is intentionally independent from the other open architecture slices and must stay green against current `main`.
70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
"""Tests for pure trusted-forwarded-header policy."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from headroom.proxy.forwarded_policy import (
|
|
ForwardedHeaderInputs,
|
|
header_first,
|
|
parse_cidr_list,
|
|
peer_is_trusted_gateway,
|
|
resolve_forwarded_headers,
|
|
)
|
|
|
|
|
|
def test_resolve_trusted_peer_honors_sanitized_forwarded_values() -> None:
|
|
cidrs = parse_cidr_list("10.0.0.0/8")
|
|
result = resolve_forwarded_headers(
|
|
ForwardedHeaderInputs(
|
|
peer_host="10.0.0.5",
|
|
forwarded_for="203.0.113.7, 10.0.0.99",
|
|
forwarded_proto=" https ",
|
|
forwarded_host=" api.example.com ",
|
|
),
|
|
cidrs,
|
|
)
|
|
|
|
assert result.trusted is True
|
|
assert result.rejected is False
|
|
assert result.client_ip == "203.0.113.7"
|
|
assert result.forwarded == {
|
|
"for": "203.0.113.7",
|
|
"proto": "https",
|
|
"host": "api.example.com",
|
|
}
|
|
|
|
|
|
def test_resolve_untrusted_peer_rejects_forwarded_values() -> None:
|
|
cidrs = parse_cidr_list("10.0.0.0/8")
|
|
result = resolve_forwarded_headers(
|
|
ForwardedHeaderInputs(
|
|
peer_host="8.8.8.8",
|
|
forwarded_for="203.0.113.7",
|
|
forwarded_proto="https",
|
|
forwarded_host="api.example.com",
|
|
),
|
|
cidrs,
|
|
)
|
|
|
|
assert result.trusted is False
|
|
assert result.rejected is True
|
|
assert result.client_ip == "8.8.8.8"
|
|
assert result.forwarded == {"for": "", "proto": "", "host": ""}
|
|
|
|
|
|
def test_resolve_direct_untrusted_peer_without_forwarded_values_is_not_rejection() -> None:
|
|
result = resolve_forwarded_headers(ForwardedHeaderInputs(peer_host="8.8.8.8"), ())
|
|
|
|
assert result.trusted is False
|
|
assert result.rejected is False
|
|
assert result.client_ip == "8.8.8.8"
|
|
assert result.forwarded == {"for": "", "proto": "", "host": ""}
|
|
|
|
|
|
def test_peer_trust_handles_ipv4_mapped_ipv6() -> None:
|
|
cidrs = parse_cidr_list("10.0.0.0/8")
|
|
|
|
assert peer_is_trusted_gateway("::ffff:10.0.0.1", cidrs) is True
|
|
|
|
|
|
def test_header_first_uses_leftmost_forwarded_for_hop() -> None:
|
|
assert header_first("203.0.113.7, 10.0.0.99, 10.0.0.5") == "203.0.113.7"
|