photoanalyzer/tests/unit/test_security_policy.py

"""US07-02: the request-admission policy, enumerated.

``evaluate`` decides every refusal the API can make before a route runs, so the
whole local-web threat model is one table here: who may call, from where, with what
proof. The middleware and the endpoints are covered black box in
``tests/e2e/test_security.py``; this file pins the rules themselves, including the
combinations a browser can produce but a test client rarely does.
"""

from __future__ import annotations

import pytest

from photo_pipeline.api.security import (
    CSRF_HEADER,
    PUBLIC_PATHS,
    Session,
    evaluate,
    split_host,
)

SESSION = Session(id="session-id", csrf_token="csrf-token")
HOST = "127.0.0.1:8000"
LIMIT = 1024


def check(method="GET", path="/api/v1/workflow", **headers):
    """Evaluate a request that is authenticated and same-origin unless overridden."""
    sent = {
        "host": HOST,
        "cookie-session": SESSION.id,
        CSRF_HEADER: SESSION.csrf_token,
    }
    sent.update({name.replace("_", "-"): value for name, value in headers.items()})
    sent = {name: value for name, value in sent.items() if value is not None}
    return evaluate(
        method=method,
        path=path,
        headers=sent,
        session=SESSION,
        max_request_bytes=LIMIT,
    )


def test_an_authenticated_same_origin_request_is_admitted():
    assert check() is None
    assert check(method="POST", origin="http://127.0.0.1:8000") is None
    assert check(sec_fetch_site="same-origin") is None


@pytest.mark.parametrize("host", ["evil.example", "evil.example:8000", "192.168.1.10:8000", ""])
def test_a_non_loopback_host_is_refused(host):
    """DNS rebinding: the browser thinks it is talking to the attacker's name, which
    resolves to 127.0.0.1. The name is the evidence, so the name is checked."""
    refusal = check(host=host)
    assert (refusal.status, refusal.code) == (403, "host_not_allowed")


@pytest.mark.parametrize("host", ["127.0.0.1:8000", "localhost:8000", "[::1]:8000", "localhost"])
def test_loopback_hosts_are_accepted(host):
    assert check(host=host) is None


@pytest.mark.parametrize(
    "origin",
    [
        "http://evil.example",
        "https://evil.example:8000",
        "http://127.0.0.1:9999",  # another local app is still another origin
        "http://localhost.evil.example:8000",
        "null",
        "file://",
    ],
)
def test_a_foreign_origin_is_refused(origin):
    refusal = check(method="POST", origin=origin)
    assert (refusal.status, refusal.code) == (403, "origin_not_allowed")


@pytest.mark.parametrize("origin", ["http://127.0.0.1:8000", "http://localhost:8000"])
def test_this_applications_origin_is_accepted(origin):
    assert check(method="POST", origin=origin) is None


@pytest.mark.parametrize("site", ["cross-site", "same-site"])
def test_a_cross_site_fetch_is_refused_even_without_an_origin(site):
    """What ``<img src="http://127.0.0.1:8000/...">`` on another page looks like."""
    refusal = check(path="/api/v1/assets/a1/thumbnail", sec_fetch_site=site)
    assert (refusal.status, refusal.code) == (403, "cross_site_blocked")


def test_a_user_initiated_navigation_is_accepted():
    assert check(sec_fetch_site="none") is None


def test_a_request_without_a_session_is_unauthenticated():
    for method, path in [
        ("GET", "/api/v1/workflow"),
        ("POST", "/api/v1/jobs"),
        ("GET", "/api/v1/assets/a1/thumbnail"),
    ]:
        refusal = check(method=method, path=path, cookie_session=None)
        assert (refusal.status, refusal.code) == (401, "unauthenticated"), path


def test_a_forged_session_is_unauthenticated():
    refusal = check(cookie_session="guessed")
    assert (refusal.status, refusal.code) == (401, "unauthenticated")


@pytest.mark.parametrize("path", sorted(PUBLIC_PATHS))
def test_health_and_the_bootstrap_stay_reachable_without_a_session(path):
    assert check(path=path, cookie_session=None) is None


def test_the_static_shell_needs_no_session():
    """It has to load before any JavaScript can ask for one."""
    assert check(path="/app/index.html", cookie_session=None) is None


@pytest.mark.parametrize("method", ["POST", "PUT", "PATCH", "DELETE"])
def test_a_mutation_without_a_valid_csrf_token_is_refused(method):
    for token in (None, "guessed"):
        refusal = check(method=method, path="/api/v1/jobs", **{CSRF_HEADER: token})
        assert (refusal.status, refusal.code) == (403, "csrf_failed")


@pytest.mark.parametrize("method", ["GET", "HEAD", "OPTIONS"])
def test_reads_need_no_csrf_token(method):
    assert check(method=method, **{CSRF_HEADER: None}) is None


def test_an_oversized_body_is_refused_before_it_is_read():
    refusal = check(method="POST", path="/api/v1/jobs", content_length=str(LIMIT + 1))
    assert (refusal.status, refusal.code) == (413, "payload_too_large")
    assert check(method="POST", path="/api/v1/jobs", content_length=str(LIMIT)) is None


def test_the_host_check_precedes_authentication():
    """A refusal must not tell a foreign caller whether its session guess was right."""
    refusal = check(host="evil.example", cookie_session="guessed")
    assert refusal.code == "host_not_allowed"


def test_refusals_name_no_path_secret_or_internal():
    refusals = [
        check(host="evil.example"),
        check(method="POST", origin="http://evil.example"),
        check(cookie_session=None),
        check(method="POST", **{CSRF_HEADER: None}),
    ]
    for refusal in refusals:
        assert SESSION.id not in refusal.message
        assert SESSION.csrf_token not in refusal.message
        assert "/" not in refusal.message


@pytest.mark.parametrize(
    "value,expected",
    [
        ("127.0.0.1:8000", ("127.0.0.1", "8000")),
        ("localhost", ("localhost", "")),
        ("[::1]:8000", ("[::1]", "8000")),
        ("[::1]", ("[::1]", "")),
        ("", ("", "")),
    ],
)
def test_split_host(value, expected):
    assert split_host(value) == expected