FabledGrimoire/tests/test_proxy.py

"""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"):
    # Quart's ASGI layer writes the socket peer into a Remote-Addr header,
    # replacing any the client sent; a test request context takes it as given.
    return app.test_request_context(
        "/", headers={"Remote-Addr": PEER, **(headers or {})}, scheme=scheme,
    )


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 = [("Remote-Addr", PEER), ("X-Forwarded-For", FORGED), ("X-Forwarded-For", CLIENT)]
    async with _app().test_request_context("/", headers=headers):
        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()