294 lines
10 KiB
Python
294 lines
10 KiB
Python
"""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,
|
|
LOOPBACK_HOSTS,
|
|
PUBLIC_PATHS,
|
|
FailureLimiter,
|
|
Session,
|
|
evaluate,
|
|
exposed_hosts,
|
|
external_view,
|
|
split_host,
|
|
trust_refusal,
|
|
)
|
|
from photo_pipeline.config import Config
|
|
|
|
SESSION = Session(id="session-id", csrf_token="csrf-token")
|
|
HOST = "127.0.0.1:8000"
|
|
LIMIT = 1024
|
|
HOSTNAME = "photos.example.com"
|
|
|
|
|
|
def check(
|
|
method="GET",
|
|
path="/api/v1/workflow",
|
|
allowed_hosts=LOOPBACK_HOSTS,
|
|
scheme="http",
|
|
**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,
|
|
allowed_hosts=allowed_hosts,
|
|
scheme=scheme,
|
|
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
|
|
|
|
|
|
# ── US08-01: the same table with a configured trust boundary ─────────────────
|
|
|
|
CONFIGURED = frozenset(LOOPBACK_HOSTS | {HOSTNAME})
|
|
|
|
|
|
def test_a_configured_host_is_accepted_and_its_neighbours_are_not():
|
|
assert check(host=HOSTNAME, allowed_hosts=CONFIGURED) is None
|
|
for host in ("other.example.com", f"evil-{HOSTNAME}", "192.168.1.10"):
|
|
refusal = check(host=host, allowed_hosts=CONFIGURED)
|
|
assert (refusal.status, refusal.code) == (403, "host_not_allowed"), host
|
|
|
|
|
|
def test_the_loopback_default_refuses_a_host_nobody_configured():
|
|
"""The default set is what the app enforced before there was a setting."""
|
|
refusal = check(host=HOSTNAME)
|
|
assert (refusal.status, refusal.code) == (403, "host_not_allowed")
|
|
|
|
|
|
def test_the_origin_must_match_the_external_scheme():
|
|
for scheme in ("http", "https"):
|
|
assert (
|
|
check(
|
|
method="POST",
|
|
host=HOSTNAME,
|
|
origin=f"{scheme}://{HOSTNAME}",
|
|
allowed_hosts=CONFIGURED,
|
|
scheme=scheme,
|
|
)
|
|
is None
|
|
)
|
|
# An HTTPS deployment whose caller claims plain HTTP is a different origin.
|
|
refusal = check(
|
|
method="POST",
|
|
host=HOSTNAME,
|
|
origin=f"http://{HOSTNAME}",
|
|
allowed_hosts=CONFIGURED,
|
|
scheme="https",
|
|
)
|
|
assert (refusal.status, refusal.code) == (403, "origin_not_allowed")
|
|
|
|
|
|
def view(client, *, trusted=(), **headers):
|
|
sent = {name.replace("_", "-"): value for name, value in headers.items()}
|
|
return external_view(
|
|
client=client,
|
|
headers={"host": HOST, **sent},
|
|
scheme="http",
|
|
trusted_proxies=frozenset(trusted),
|
|
)
|
|
|
|
|
|
def test_forwarded_headers_are_ignored_without_a_trusted_proxy():
|
|
forged = {"x_forwarded_proto": "https", "x_forwarded_host": HOSTNAME}
|
|
assert view("10.0.0.9", **forged) == ("http", HOST)
|
|
assert view(None, **forged) == ("http", HOST)
|
|
# Configuring *a* proxy does not trust a caller that is not it.
|
|
assert view("10.0.0.9", trusted=("10.0.0.1",), **forged) == ("http", HOST)
|
|
|
|
|
|
def test_a_trusted_proxy_defines_the_external_scheme_and_host():
|
|
assert view(
|
|
"10.0.0.1", trusted=("10.0.0.1",), x_forwarded_proto="https", x_forwarded_host=HOSTNAME
|
|
) == ("https", HOSTNAME)
|
|
# A chain: the first entry is what the original client asked for.
|
|
assert view(
|
|
"10.0.0.1",
|
|
trusted=("10.0.0.1",),
|
|
x_forwarded_proto="https, http",
|
|
x_forwarded_host=f"{HOSTNAME}, inner.internal",
|
|
) == ("https", HOSTNAME)
|
|
# Trusted but silent: this hop's own view stands.
|
|
assert view("10.0.0.1", trusted=("10.0.0.1",)) == ("http", HOST)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"settings,exposed",
|
|
[
|
|
({}, []),
|
|
({"host": "127.0.0.1"}, []),
|
|
({"allowed_hosts": ("localhost", "127.0.0.1")}, []),
|
|
({"allowed_hosts": (f"{HOSTNAME}:8443",)}, [HOSTNAME]),
|
|
({"host": "0.0.0.0", "allowed_hosts": (HOSTNAME,)}, ["0.0.0.0", HOSTNAME]),
|
|
],
|
|
)
|
|
def test_exposed_hosts_names_only_what_another_machine_can_reach(settings, exposed):
|
|
assert exposed_hosts(Config(**settings)) == exposed
|
|
|
|
|
|
def test_an_exposed_configuration_without_a_secret_must_not_serve():
|
|
refusal = trust_refusal(Config(allowed_hosts=(HOSTNAME,)))
|
|
assert HOSTNAME in refusal and "PHOTO_PIPELINE_ACCESS_SECRET" in refusal
|
|
assert trust_refusal(Config(allowed_hosts=(HOSTNAME,), access_secret="s")) is None
|
|
# Loopback-only, with and without a secret, is unchanged.
|
|
assert trust_refusal(Config()) is None
|
|
assert trust_refusal(Config(access_secret="s")) is None
|
|
|
|
|
|
def test_failed_attempts_are_bounded_per_window():
|
|
limiter = FailureLimiter(limit=2, window=60.0)
|
|
assert not limiter.blocked()
|
|
limiter.record_failure()
|
|
assert not limiter.blocked()
|
|
limiter.record_failure()
|
|
assert limiter.blocked()
|
|
# Attempts age out, so a locked-out operator is not locked out forever.
|
|
limiter._failures = [-120.0, -120.0]
|
|
assert not limiter.blocked()
|
|
|
|
|
|
@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
|