Files
bvandeusenandClaude Opus 5 c851b901df
CI & Build / Build now, or wait for Android? (push) Successful in 3s
CI & Build / Python lint (push) Successful in 3s
CI & Build / TypeScript typecheck (push) Successful in 7s
CI & Build / Python tests (push) Successful in 8s
CI & Build / integration (push) Successful in 14s
CI & Build / Build & push image (push) Successful in 15s
The proxy-hops test still read the value from Config
09b5f87 moved trusted_proxy_hops out of the environment and into the
settings registry, but tests/test_proxy.py kept asserting against
Config.trusted_proxy_hops() — which no longer exists. The unit lane has
been red since that commit.

Assert through live() instead. That is what proxy.py actually calls, and
it is seeded from the defaults at import time, so the test covers the
case that matters: a boot that has not reached the database yet still
counts one hop rather than zero.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-23 16:31:21 -04:00

74 lines
3.1 KiB
Python

"""The proxy trust boundary.
The whole security property is "a caller cannot forge their own address", and it rests
on counting in from the RIGHT of the header rather than the left. These are the cases
that tell the two apart — pure functions, no request context, no database.
"""
from thoughtsync.proxy import forwarded_for, trusted_entry
from thoughtsync.settings import live
PEER = "10.0.0.1" # the socket address: our own proxy, or the caller when unproxied
def test_default_is_one_hop():
# One reverse proxy terminating TLS — this deployment, and the only shape that is
# safe to assume. A wrong default here is a silent security bug, not a preference.
#
# Read through live() rather than the registry: live() is what proxy.py actually
# calls, and it is seeded from the defaults at import time so the value is right
# before the first database read. A boot that never reached the DB must still
# count one hop, not zero.
assert live("trusted_proxy_hops") == 1
def test_no_proxy_ignores_the_header_entirely():
# hops=0 says nothing in front of us appends anything, so the header can only be
# something a caller invented.
assert forwarded_for("1.2.3.4", PEER, 0) == PEER
def test_one_hop_reads_what_our_proxy_wrote():
assert forwarded_for("203.0.113.7", PEER, 1) == "203.0.113.7"
def test_a_forged_prefix_is_never_selected():
# THE test. A caller sends `X-Forwarded-For: 1.2.3.4`; our proxy appends the
# address it actually saw. Reading from the left would hand the caller a fresh
# rate-limit bucket for every value they invent.
assert forwarded_for("1.2.3.4, 203.0.113.7", PEER, 1) == "203.0.113.7"
# …and padding it doesn't help either.
assert forwarded_for("a, b, c, d, 203.0.113.7", PEER, 1) == "203.0.113.7"
def test_two_hops_sees_past_a_cdn():
# Cloudflare appended the real client; our proxy appended Cloudflare.
assert forwarded_for("203.0.113.7, 172.16.0.5", PEER, 2) == "203.0.113.7"
assert forwarded_for("1.2.3.4, 203.0.113.7, 172.16.0.5", PEER, 2) == "203.0.113.7"
def test_a_short_header_falls_back_rather_than_reaching_left():
# Fewer proxies than configured. Reaching further left would start believing
# entries no proxy of ours wrote, so the safe direction is the socket address —
# at worst several callers share one bucket.
assert forwarded_for("203.0.113.7", PEER, 2) == PEER
assert forwarded_for("", PEER, 1) == PEER
def test_malformed_headers_do_not_crash_or_leak_empties():
assert forwarded_for(",,,", PEER, 1) == PEER
assert forwarded_for(" , 203.0.113.7 , ", PEER, 1) == "203.0.113.7"
def test_the_key_is_length_bounded():
# It becomes a dict key in the limiter; an unbounded header must not become an
# unbounded allocation.
assert len(forwarded_for("x" * 5000, PEER, 1)) <= 64
def test_trusted_entry_reports_absence_rather_than_guessing():
# `is_https` needs to tell "no trusted entry" apart from "an entry saying http",
# which is why this returns None rather than a default.
assert trusted_entry("", 1) is None
assert trusted_entry("https", 0) is None
assert trusted_entry("http, https", 1) == "https"