US07-02: Harden API Authorization and Path Boundaries (#84)

tests/unit/test_path_policy.py

@@ -81,3 +81,42 @@ def test_symlink_within_root_is_allowed(tmp_path):
         pytest.skip("cannot create symlink on this platform")
     found = path_policy.discover([root])
     assert target in found and link in found
+
+
+def test_resolve_in_roots_returns_the_path_the_caller_must_use(tmp_path):
+    """The resolved path is the answer, not a yes/no: a caller that revalidates one
+    name and then opens another has an open symlink race (US07-02)."""
+    root = tmp_path / "lib"
+    (root / "sub").mkdir(parents=True)
+    target = root / "sub" / "real.jpg"
+    target.write_bytes(b"x")
+    link = root / "alias.jpg"
+    os.symlink(target, link)
+
+    assert path_policy.resolve_in_roots([root], link) == target.resolve()
+    assert path_policy.resolve_in_roots([root], target) == target.resolve()
+
+
+def test_resolve_in_roots_refuses_escapes_without_naming_them(tmp_path):
+    root = tmp_path / "lib"
+    root.mkdir()
+    outside = tmp_path / "outside.jpg"
+    outside.write_bytes(b"x")
+    link = root / "alias.jpg"
+    os.symlink(outside, link)
+
+    for candidate in (link, outside, root / "_IGNORE" / "a.jpg"):
+        with pytest.raises(path_policy.PathPolicyError) as raised:
+            path_policy.resolve_in_roots([root], candidate)
+        assert str(outside) not in str(raised.value)
+
+
+def test_resolve_in_roots_checks_every_configured_root(tmp_path):
+    first, second = tmp_path / "one", tmp_path / "two"
+    first.mkdir()
+    second.mkdir()
+    photo = second / "b.jpg"
+    photo.write_bytes(b"x")
+    assert path_policy.resolve_in_roots([first, second], photo) == photo.resolve()
+    # No configured boundary means nothing to check against.
+    assert path_policy.resolve_in_roots([], photo) == photo

tests/unit/test_security_policy.py

@@ -0,0 +1,169 @@
+"""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