"""The trusted proxy decides the client address and whether a request is HTTPS. Only the entry TRUSTED_PROXY_HOPS in from the right of a forwarding header was written by a proxy the app trusts; a client can put anything to its left. The session cookie's Secure flag and HSTS follow the trusted X-Forwarded-Proto, and plain HTTP is left untouched. """ import pytest from quart import Quart, Response, session from grimoire import config, proxy from grimoire.config import Config from grimoire.proxy import ( HSTS_VALUE, PerRequestSecureSession, apply_https_headers, client_ip, is_https, trusted_entry, ) CLIENT = "203.0.113.7" FORGED = "198.51.100.1" PEER = "192.0.2.10" @pytest.mark.parametrize("header, hops, expected", [ ("", 1, None), (CLIENT, 0, None), (CLIENT, 1, CLIENT), (f"{FORGED}, {CLIENT}", 1, CLIENT), (f"{FORGED}, {CLIENT}, {PEER}", 2, CLIENT), # Shorter than the hop count: the leftmost entry, written by the first proxy. (CLIENT, 2, CLIENT), (" , ", 1, None), ]) def test_trusted_entry_counts_in_from_the_right(header, hops, expected): assert trusted_entry(header, hops) == expected def _app() -> Quart: app = Quart(__name__) app.secret_key = "test-only-key" app.session_interface = PerRequestSecureSession() @app.route("/") async def index(): session["seen"] = True response = Response("ok") apply_https_headers(response) return response return app def _context(app, headers=None, scheme="http"): return app.test_request_context( "/", headers=headers or {}, scheme=scheme, scope_base={"client": (PEER, 40000)}, ) async def test_a_client_forged_address_left_of_the_proxys_is_ignored(monkeypatch): monkeypatch.setattr(Config, "TRUSTED_PROXY_HOPS", 1) async with _context(_app(), {"X-Forwarded-For": f"{FORGED}, {CLIENT}"}): assert client_ip() == CLIENT async def test_zero_hops_ignores_the_header_and_reads_the_socket_peer(monkeypatch): monkeypatch.setattr(Config, "TRUSTED_PROXY_HOPS", 0) async with _context(_app(), {"X-Forwarded-For": CLIENT}): assert client_ip() == PEER async def test_no_header_reads_the_socket_peer(monkeypatch): monkeypatch.setattr(Config, "TRUSTED_PROXY_HOPS", 1) async with _context(_app()): assert client_ip() == PEER async def test_a_header_sent_twice_is_read_as_one_chain(monkeypatch): monkeypatch.setattr(Config, "TRUSTED_PROXY_HOPS", 1) headers = [("X-Forwarded-For", FORGED), ("X-Forwarded-For", CLIENT)] async with _app().test_request_context( "/", headers=headers, scope_base={"client": (PEER, 40000)}, ): assert client_ip() == CLIENT async def test_a_short_proto_header_is_read_where_the_first_proxy_wrote_it(monkeypatch): monkeypatch.setattr(Config, "TRUSTED_PROXY_HOPS", 2) async with _context(_app(), {"X-Forwarded-Proto": "https"}): assert is_https() async def test_a_forged_https_left_of_the_proxys_http_is_ignored(monkeypatch): monkeypatch.setattr(Config, "TRUSTED_PROXY_HOPS", 1) async with _context(_app(), {"X-Forwarded-Proto": "https, http"}): assert not is_https() async def test_a_direct_https_request_is_https_at_any_hop_count(monkeypatch): monkeypatch.setattr(Config, "TRUSTED_PROXY_HOPS", 0) async with _context(_app(), scheme="https"): assert is_https() def test_outside_a_request_nothing_is_trusted(): assert client_ip() == "" assert not is_https() async def _get(monkeypatch, hops, headers): monkeypatch.setattr(Config, "TRUSTED_PROXY_HOPS", hops) response = await _app().test_client().get("/", headers=headers) cookie = " ".join(response.headers.getlist("Set-Cookie")) assert "session=" in cookie return cookie, response.headers.get("Strict-Transport-Security") async def test_hops_zero_with_https_header_sends_no_secure_flag_and_no_hsts(monkeypatch): cookie, hsts = await _get(monkeypatch, 0, {"X-Forwarded-Proto": "https"}) assert "Secure" not in cookie assert hsts is None async def test_hops_one_with_https_header_sends_secure_flag_and_hsts(monkeypatch): cookie, hsts = await _get(monkeypatch, 1, {"X-Forwarded-Proto": "https"}) assert "Secure" in cookie assert hsts == HSTS_VALUE async def test_a_plain_request_sends_neither(monkeypatch): cookie, hsts = await _get(monkeypatch, 1, {}) assert "Secure" not in cookie assert hsts is None def test_the_app_decides_secure_per_request(): from grimoire.app import create_app assert isinstance(create_app().session_interface, PerRequestSecureSession) def test_the_app_reads_forwarding_headers_only_through_this_module(): """A second reader of X-Forwarded-* or the socket peer would be a second trust rule. Matches the reads themselves, not prose that names a header.""" import pathlib import re read = re.compile( r"""headers\.(?:get|getlist|__getitem__)\(\s*["']X-(?:Forwarded|Real)""" r"""|headers\[\s*["']X-(?:Forwarded|Real)""" r"""|\.remote_addr\b""", re.IGNORECASE, ) src = pathlib.Path(proxy.__file__).parent readers = sorted( str(path.relative_to(src)) for path in src.rglob("*.py") if path.name != "proxy.py" and read.search(path.read_text()) ) assert readers == [] assert read.search((src / "proxy.py").read_text()), "the pattern no longer finds the real reader" @pytest.mark.parametrize("env, expected", [ ({}, 1), ({"TRUSTED_PROXY_HOPS": "2"}, 2), ({"TRUSTED_PROXY_HOPS": "0"}, 0), ({"TRUSTED_PROXY_HOPS": "two"}, -1), ({"TRUST_PROXY_HEADERS": "false"}, 0), ({"TRUST_PROXY_HEADERS": "true"}, 1), ({"TRUST_PROXY_HEADERS": "false", "TRUSTED_PROXY_HOPS": "1"}, 1), ]) def test_the_hop_count_from_the_environment(monkeypatch, env, expected): monkeypatch.delenv("TRUSTED_PROXY_HOPS", raising=False) monkeypatch.delenv("TRUST_PROXY_HEADERS", raising=False) for name, value in env.items(): monkeypatch.setenv(name, value) assert config._trusted_proxy_hops() == expected def test_a_hop_count_that_is_not_a_number_refuses_to_start(monkeypatch): monkeypatch.setattr(Config, "TRUSTED_PROXY_HOPS", -1) with pytest.raises(ValueError, match="TRUSTED_PROXY_HOPS"): Config.validate() @pytest.mark.parametrize("key", sorted(config.PLACEHOLDER_SECRET_KEYS)) def test_a_placeholder_secret_key_refuses_to_start(monkeypatch, key): monkeypatch.setattr(Config, "SECRET_KEY", key) with pytest.raises(ValueError, match="public placeholder"): Config.validate()