phase: 82_security_headers
All completion criteria verified green — no defects found, nothing to fix. Final report: **Phase 82 (security headers) — final verification pass: all green** - Verified prior-run implementation: `app/core/security_headers.py` (pure-ASGI, header-only, exact A1 CSP), registration in `app/main.py` after `configure_caching` (outermost), unit/integration/E2E suites. - Deviation confirmed sound: `data:`-URI favicon (blocked by locked CSP) → static `frontend/assets/favicon.svg` in 5 templates + Containerfile `cp`; SVG element byte-identical to the old data-URI (verified programmatically); serves 200 with all three headers. - Curl check (server booted like e2e conftest, log: `/tmp/curl_security_headers_final.log`): `/`, `/api/health`, `/assets/styles.css`, `/nope` (404) → all three headers, CSP exactly `default-src 'self'; base-uri 'none'; frame-ancestors 'none'`. - `uv run pytest tests/unit/test_security_headers.py tests/integration/test_security_headers.py -v --no-cov` → 13 passed (incl. SSE byte-identity pin). - `uv run pytest tests/e2e/test_security_headers.py -v --no-cov` (isolated) → 2 passed (headers + zero CSP violations + painted page). - SSE tripwire `uv run pytest tests/e2e/test_chat_rag.py -v --no-cov` → 3 passed. - `uv run pytest --cov=app --cov-report=term-missing` → 1665 passed, app/ 99% (>90%); `uv run ruff check . && uv run pyright` → clean (0 errors). - `git diff --stat` limited to phase-82 files + the two documented deviations (favicon set, `tests/unit/__init__.py`); no `pyproject.toml`/`uv.lock`/JS diffs. - Commit + phase-dir move left to the harness per pipeline rules (not executed by me). Next pending phase: `83_chat_save_payload_limits`.
This commit is contained in:
@@ -0,0 +1,128 @@
|
||||
"""Phase 82 E2E (Playwright): the security headers reach a real browser and
|
||||
the strict CSP does not break the page.
|
||||
|
||||
Story: n/a — security hardening, audit SEC-04 (``.agents/remediation_plan.md``).
|
||||
Run in isolation (DB must be up: ``podman compose up -d db``):
|
||||
|
||||
uv run pytest tests/e2e/test_security_headers.py -v --no-cov
|
||||
|
||||
No LLM interaction: the page load IS the contract. The suite never touches
|
||||
the chat stream and never depends on the aipi endpoint — the shared
|
||||
``app_server`` fixture's deterministic LOCAL mock LLM is simply unused.
|
||||
|
||||
Test → task mapping (phase 82, task 03):
|
||||
1. ``test_chat_page_navigation_headers_and_csp_clean`` — a real ``page.goto``
|
||||
of the chat shell (``/``): the navigation response carries all three
|
||||
headers (CSP exactly the A1 string), the page painted under the CSP
|
||||
(the dark-tech canvas applied → the stylesheet loaded), the shell's
|
||||
``#main`` booted, and zero console CSP violations.
|
||||
2. ``test_sources_page_navigation_headers_and_csp_clean`` — the same header
|
||||
+ no-violation contract for the second page, ``/sources.html``.
|
||||
|
||||
Paint-check note (vs. the task text): the task suggested asserting the
|
||||
``body`` background-color is not transparent, but the pinned design
|
||||
contract (phase 08, ``test_dark_tech_theme.py``) puts the visible canvas on
|
||||
``<html>`` (``html { background: var(--bg) }``) and keeps ``<body>``
|
||||
transparent on purpose (the z-index:-1 grid layer must show). The
|
||||
stylesheet-loaded-under-the-CSP contract is therefore pinned on the
|
||||
``<html>`` canvas color — the same value phase 08 pins.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
from playwright.sync_api import ConsoleMessage, Page, expect
|
||||
|
||||
#: The exact owner-approved policy (phase 82, decision A1).
|
||||
CSP = "default-src 'self'; base-uri 'none'; frame-ancestors 'none'"
|
||||
|
||||
#: Chromium reports CSP denials to the console with this phrasing
|
||||
#: ("Refused to … because it violates the following Content Security
|
||||
#: Policy directive …" / "The Content Security Policy directive …
|
||||
#: prevents execution …").
|
||||
_CSP_CONSOLE = re.compile(r"Content Security Policy", re.IGNORECASE)
|
||||
|
||||
#: The dark-tech canvas — ``frontend/assets/styles.css`` ``--bg: #0f0a0a``,
|
||||
#: pinned by phase 08 (``test_dark_tech_theme.py``).
|
||||
CANVAS_BG = "rgb(15, 10, 10)"
|
||||
|
||||
|
||||
def _attach_csp_watch(page: Page) -> list[str]:
|
||||
"""Record every console message that mentions a CSP denial.
|
||||
|
||||
Attached BEFORE navigation (the contract — denials of the main
|
||||
document's resources are logged while it loads). Returns the live
|
||||
list; assert it is empty after the load.
|
||||
"""
|
||||
violations: list[str] = []
|
||||
|
||||
def _on_console(msg: ConsoleMessage) -> None:
|
||||
if _CSP_CONSOLE.search(msg.text):
|
||||
violations.append(msg.text)
|
||||
|
||||
page.on("console", _on_console)
|
||||
return violations
|
||||
|
||||
|
||||
def _assert_navigation_headers(page: Page, url: str) -> None:
|
||||
"""A real navigation: the response carries all three headers, CSP exact."""
|
||||
response = page.goto(url)
|
||||
assert response is not None, f"no navigation response for {url}"
|
||||
assert response.status == 200, f"{url} → {response.status}"
|
||||
headers = response.headers # lowercase keys
|
||||
assert headers.get("content-security-policy") == CSP, (
|
||||
f"{url}: CSP is {headers.get('content-security-policy')!r}, "
|
||||
f"expected the exact A1 string {CSP!r}"
|
||||
)
|
||||
assert headers.get("x-frame-options") == "DENY", (
|
||||
f"{url}: X-Frame-Options is {headers.get('x-frame-options')!r}"
|
||||
)
|
||||
assert headers.get("x-content-type-options") == "nosniff", (
|
||||
f"{url}: X-Content-Type-Options is {headers.get('x-content-type-options')!r}"
|
||||
)
|
||||
|
||||
|
||||
def _assert_chat_page_painted(page: Page) -> None:
|
||||
"""The page painted under the CSP — styles loaded, shell booted.
|
||||
|
||||
The visible canvas is ``<html>`` (see the module docstring): its
|
||||
computed color can only be the dark-tech palette if ``/assets/*.css``
|
||||
was fetched and applied under ``default-src 'self'`` — a CSP-broken
|
||||
document would be a white/default canvas without ``#main`` styled.
|
||||
"""
|
||||
bg = page.evaluate(
|
||||
"() => getComputedStyle(document.documentElement).backgroundColor"
|
||||
)
|
||||
assert bg == CANVAS_BG, (
|
||||
f"dark-tech canvas not applied under the CSP (stylesheet not loaded?): {bg}"
|
||||
)
|
||||
expect(page.locator("#main")).to_be_attached()
|
||||
box = page.locator("#main").bounding_box()
|
||||
assert box is not None and box["width"] > 0 and box["height"] > 0, (
|
||||
"shell <main> is present but invisible/empty — the page did not boot"
|
||||
)
|
||||
|
||||
|
||||
def test_chat_page_navigation_headers_and_csp_clean(
|
||||
page: Page, app_url: str, db_ready: None
|
||||
) -> None:
|
||||
"""AC1: real navigation of the chat shell → three headers (CSP exact),
|
||||
painted page, zero CSP violations on the console."""
|
||||
violations = _attach_csp_watch(page)
|
||||
_assert_navigation_headers(page, app_url + "/")
|
||||
_assert_chat_page_painted(page)
|
||||
# Console delivery is async — let pending messages settle before the
|
||||
# zero-violation gate (the load event already fired).
|
||||
page.wait_for_timeout(250)
|
||||
assert violations == [], f"CSP violations on the chat page: {violations}"
|
||||
|
||||
|
||||
def test_sources_page_navigation_headers_and_csp_clean(
|
||||
page: Page, app_url: str, db_ready: None
|
||||
) -> None:
|
||||
"""AC2: the second page (Sources) — same three headers, no CSP
|
||||
violations."""
|
||||
violations = _attach_csp_watch(page)
|
||||
_assert_navigation_headers(page, app_url + "/sources.html")
|
||||
page.wait_for_timeout(250)
|
||||
assert violations == [], f"CSP violations on the sources page: {violations}"
|
||||
@@ -0,0 +1,104 @@
|
||||
"""Integration: the phase-82 security-header contract against the REAL app.
|
||||
|
||||
Phase 82 (audit SEC-04) adds a header-only, pure-ASGI middleware
|
||||
(``app/core/security_headers.py``) registered OUTERMOST in
|
||||
``app/main.py`` (last ``add_middleware`` — after
|
||||
``configure_caching``), so every response this process serves — pages,
|
||||
API JSON, static assets, even the static catch-all's 404s — carries:
|
||||
|
||||
* ``Content-Security-Policy`` — the exact A1 string (``default-src 'self';
|
||||
base-uri 'none'; frame-ancestors 'none'`` → clickjacking closed, no
|
||||
inline anything because the No-CDN frontend has none);
|
||||
* ``X-Frame-Options: DENY`` — the legacy no-framing fallback;
|
||||
* ``X-Content-Type-Options: nosniff`` — the MIME-confusion belt.
|
||||
|
||||
This suite pins those three on the four response shapes the browser
|
||||
actually sees — a page (``/``), an API response (``/api/health``), a
|
||||
static asset (``/assets/styles.css``), and a 404
|
||||
(``/definitely-not-a-page`` — the outermost-position proof: the 404 is
|
||||
built deep inside the static catch-all, yet the headers are still on
|
||||
it) — against the real ``app.main:app`` with the real static mount on
|
||||
the real ``frontend/`` tree (no mocks, same client as
|
||||
``tests/integration/test_caching_revalidation.py`` via the shared
|
||||
``client`` fixture). It also pins that the two middlewares coexist: the
|
||||
CSP middleware is header-only, so the phase-33/54 ``?v=<token>``
|
||||
rewrite from ``CachingMiddleware`` still runs under it (the last
|
||||
regression below). The SSE chat stream's byte-identity is pinned by the
|
||||
unit streaming test and the chat E2E suites (the tripwires).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import httpx
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.core.security_headers import CSP
|
||||
|
||||
#: The exact owner-approved A1 policy string (phase 82). The constant is
|
||||
#: the single source of truth; the unit suite additionally pins that the
|
||||
#: constant itself equals this literal.
|
||||
_EXPECTED = {
|
||||
"content-security-policy": CSP,
|
||||
"x-frame-options": "DENY",
|
||||
"x-content-type-options": "nosniff",
|
||||
}
|
||||
|
||||
|
||||
def _assert_security_headers(response: httpx.Response) -> None:
|
||||
"""All three phase-82 headers present with the exact values (the
|
||||
httpx header lookup is case-insensitive)."""
|
||||
path = response.request.url.path
|
||||
for name, value in _EXPECTED.items():
|
||||
assert response.headers.get(name) == value, (
|
||||
f"{path}: {name} must be {value!r}, "
|
||||
f"got {response.headers.get(name)!r}"
|
||||
)
|
||||
|
||||
|
||||
def test_page_carries_all_three_headers(client: TestClient) -> None:
|
||||
"""``GET /`` (the shell page) — 200 + all three headers, CSP exactly
|
||||
the A1 string."""
|
||||
response = client.get("/")
|
||||
assert response.status_code == 200
|
||||
_assert_security_headers(response)
|
||||
assert response.headers["content-security-policy"] == (
|
||||
"default-src 'self'; base-uri 'none'; frame-ancestors 'none'"
|
||||
)
|
||||
|
||||
|
||||
def test_api_health_carries_all_three_headers(client: TestClient) -> None:
|
||||
"""``GET /api/health`` (an API JSON response) — 200 + all three
|
||||
headers."""
|
||||
response = client.get("/api/health")
|
||||
assert response.status_code == 200
|
||||
_assert_security_headers(response)
|
||||
|
||||
|
||||
def test_static_asset_carries_all_three_headers(client: TestClient) -> None:
|
||||
"""``GET /assets/styles.css`` (the static mount) — 200 + all three
|
||||
headers (the nosniff belt guards exactly this surface)."""
|
||||
response = client.get("/assets/styles.css")
|
||||
assert response.status_code == 200
|
||||
_assert_security_headers(response)
|
||||
|
||||
|
||||
def test_static_catchall_404_carries_all_three_headers(client: TestClient) -> None:
|
||||
"""``GET /definitely-not-a-page`` — 404 from the static catch-all,
|
||||
built deep inside the app, yet all three headers ride on it: the
|
||||
outermost-position proof (a response the middleware cannot see
|
||||
unless it wraps everything)."""
|
||||
response = client.get("/definitely-not-a-page")
|
||||
assert response.status_code == 404
|
||||
_assert_security_headers(response)
|
||||
|
||||
|
||||
def test_caching_rewrite_still_runs_under_headers_middleware(client: TestClient) -> None:
|
||||
"""Coexistence pin: the header middleware is header-only (it never
|
||||
drains or rewrites a body), so the phase-33/54 ``CachingMiddleware``
|
||||
``?v=<token>`` asset rewrite on the same page must still run — the
|
||||
served HTML still carries at least one ``?v=`` reference."""
|
||||
response = client.get("/")
|
||||
assert response.status_code == 200
|
||||
assert response.text.count("?v=") >= 1, (
|
||||
"the ?v=<token> asset rewrite no longer runs — the outermost "
|
||||
"security-header middleware altered or swallowed the body"
|
||||
)
|
||||
@@ -0,0 +1,11 @@
|
||||
"""Unit test package.
|
||||
|
||||
Like ``tests/e2e/``, this directory is a package so its test modules
|
||||
import as ``unit.<name>`` instead of a top-level ``<name>``: a
|
||||
same-basename file in two collected directories (phase 82 ships
|
||||
``test_security_headers.py`` in BOTH this directory and
|
||||
``tests/integration/``) makes pytest's default import mode raise an
|
||||
"import file mismatch" on the full suite. ``tests/integration/`` stays a
|
||||
plain directory on purpose — several of its files use sibling imports
|
||||
(``from test_chat_api import …``), which require top-level module names.
|
||||
"""
|
||||
@@ -0,0 +1,294 @@
|
||||
"""Unit tests: the pure-ASGI security-headers middleware (phase 82, task 01).
|
||||
|
||||
Pins the middleware against the phase design:
|
||||
* the ``CSP`` constant is the exact owner-approved A1 string;
|
||||
* a plain 200 response carries all three headers with the exact values;
|
||||
* a 404-shaped response carries them too (the static catch-all's 404s
|
||||
get the headers as soon as the middleware is registered in
|
||||
``app/main.py``, task 02);
|
||||
* a **streaming** SSE response passes through byte-identical — both
|
||||
chunks, in order, chunk boundaries preserved (no buffering, no
|
||||
coalescing) — AND carries the headers (the SSE passthrough pin);
|
||||
* the HTTP path wraps ``send`` (and leaves ``receive`` alone);
|
||||
* a non-HTTP scope (websocket / lifespan) passes through untouched —
|
||||
the app is called with the ORIGINAL ``send``/``receive`` (no
|
||||
wrapping, no header injection attempt, no crash).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from collections.abc import AsyncIterator
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from starlette.responses import StreamingResponse
|
||||
from starlette.types import ASGIApp, Message, Scope
|
||||
|
||||
from app.core.security_headers import CSP, SecurityHeadersMiddleware
|
||||
|
||||
#: The exact expected header set (decision A1 for the CSP, A4 for the
|
||||
#: other two).
|
||||
EXPECTED_HEADERS = {
|
||||
"content-security-policy": "default-src 'self'; base-uri 'none'; frame-ancestors 'none'",
|
||||
"x-frame-options": "DENY",
|
||||
"x-content-type-options": "nosniff",
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ASGI drivers / test apps (the repo's unit suite has no pytest-asyncio —
|
||||
# ``asyncio.run`` matches the existing style, see test_caching.py)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _drive(
|
||||
app: ASGIApp, scope: Scope, receive_messages: list[Message] | None = None
|
||||
) -> list[Message]:
|
||||
"""Run ``app`` for one connection; return every ``send`` message in order."""
|
||||
queued: list[Message] = list(
|
||||
receive_messages
|
||||
if receive_messages is not None
|
||||
else [{"type": "http.request", "body": b"", "more_body": False}]
|
||||
)
|
||||
sent: list[Message] = []
|
||||
|
||||
async def receive() -> Message:
|
||||
return queued.pop(0) if queued else {"type": "http.disconnect"}
|
||||
|
||||
async def send(message: Message) -> None:
|
||||
sent.append(message)
|
||||
|
||||
async def _run() -> None:
|
||||
await app(scope, receive, send)
|
||||
|
||||
asyncio.run(_run())
|
||||
return sent
|
||||
|
||||
|
||||
def _http_scope() -> Scope:
|
||||
return {
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": "/",
|
||||
"query_string": b"",
|
||||
"headers": [],
|
||||
"raw_path": b"/",
|
||||
"root_path": "",
|
||||
"scheme": "http",
|
||||
"server": ("testserver", 80),
|
||||
}
|
||||
|
||||
|
||||
def _header(message: Message, name: str) -> str | None:
|
||||
for key, value in message.get("headers", []):
|
||||
if key.lower() == name.lower().encode("ascii"):
|
||||
return value.decode("latin-1")
|
||||
return None
|
||||
|
||||
|
||||
def _plain_app(status: int, body: bytes, headers: list | None = None) -> ASGIApp:
|
||||
"""A raw ASGI app answering with exactly one start + one body message."""
|
||||
|
||||
async def app(scope: Scope, receive: Any, send: Any) -> None:
|
||||
await send(
|
||||
{"type": "http.response.start", "status": status, "headers": list(headers or [])}
|
||||
)
|
||||
await send({"type": "http.response.body", "body": body, "more_body": False})
|
||||
|
||||
return app
|
||||
|
||||
|
||||
def _assert_security_headers(start: Message, expected_extra: dict[str, str] | None = None) -> None:
|
||||
for name, value in EXPECTED_HEADERS.items():
|
||||
assert _header(start, name) == value, f"expected header {name!r} == {value!r}"
|
||||
for name, value in (expected_extra or {}).items():
|
||||
assert _header(start, name) == value, f"expected pre-existing header {name!r} kept"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# The policy constant
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_csp_constant_is_the_exact_a1_policy() -> None:
|
||||
"""The owner-approved A1 string, verbatim: same-origin default, no
|
||||
base-tag hijack, no framing — no 'unsafe-inline', no report sink."""
|
||||
assert CSP == "default-src 'self'; base-uri 'none'; frame-ancestors 'none'"
|
||||
assert "unsafe-inline" not in CSP
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Plain responses
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_plain_200_response_carries_all_three_headers_with_exact_values() -> None:
|
||||
wrapped = SecurityHeadersMiddleware(_plain_app(200, b"hello",
|
||||
headers=[[b"content-type", b"text/plain"]]))
|
||||
sent = _drive(wrapped, _http_scope())
|
||||
|
||||
assert [m["type"] for m in sent] == ["http.response.start", "http.response.body"]
|
||||
start, body_msg = sent
|
||||
assert start["status"] == 200
|
||||
_assert_security_headers(start, expected_extra={"content-type": "text/plain"})
|
||||
# The body passes through byte-identical.
|
||||
assert body_msg["body"] == b"hello"
|
||||
assert body_msg["more_body"] is False
|
||||
|
||||
|
||||
def test_404_shaped_response_carries_all_three_headers() -> None:
|
||||
"""A 404 (e.g. the static catch-all's "file not found" JSON) gets the
|
||||
headers too — with an EMPTY downstream header list (no pre-existing
|
||||
headers to preserve)."""
|
||||
wrapped = SecurityHeadersMiddleware(_plain_app(404, b"not found"))
|
||||
sent = _drive(wrapped, _http_scope())
|
||||
|
||||
start, body_msg = sent
|
||||
assert start["status"] == 404
|
||||
_assert_security_headers(start)
|
||||
assert body_msg["body"] == b"not found"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# The SSE streaming passthrough pin
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_sse_stream_passes_through_byte_identical_with_headers() -> None:
|
||||
"""A ``text/event-stream`` response driven through a REAL starlette
|
||||
``StreamingResponse`` (async generator yielding two SSE frames):
|
||||
the body arrives as exactly the two chunks, in order, boundaries
|
||||
intact — no buffering, no coalescing — and the three headers are on
|
||||
the start message (the SSE passthrough pin; a body-touching
|
||||
regression would break this and the chat E2E suites)."""
|
||||
|
||||
async def frames() -> AsyncIterator[bytes]:
|
||||
yield b"data: a\n\n"
|
||||
yield b"data: b\n\n"
|
||||
|
||||
response = StreamingResponse(frames(), media_type="text/event-stream")
|
||||
sent = _drive(SecurityHeadersMiddleware(response), _http_scope())
|
||||
|
||||
start = sent[0]
|
||||
assert start["type"] == "http.response.start"
|
||||
assert start["status"] == 200
|
||||
_assert_security_headers(start)
|
||||
# starlette's own StreamingResponse appends ``; charset=utf-8`` — the
|
||||
# media type itself must be SSE (the middleware never rewrites it).
|
||||
content_type = _header(start, "content-type")
|
||||
assert content_type is not None
|
||||
assert content_type.split(";", 1)[0].strip() == "text/event-stream"
|
||||
|
||||
body_msgs = [m for m in sent[1:] if m["type"] == "http.response.body"]
|
||||
# starlette's StreamingResponse closes the generator with a final
|
||||
# EMPTY trailer frame (no bytes) after the last chunk — the two data
|
||||
# frames must still arrive untouched, in order, boundaries intact.
|
||||
assert [m["body"] for m in body_msgs] == [b"data: a\n\n", b"data: b\n\n", b""]
|
||||
assert b"".join(m["body"] for m in body_msgs) == b"data: a\n\ndata: b\n\n"
|
||||
assert [m["more_body"] for m in body_msgs] == [True, True, False]
|
||||
|
||||
|
||||
def test_sse_frames_are_forwarded_in_order_without_delayed_buffering() -> None:
|
||||
"""Message-level pin: every ``http.response.body`` message the app
|
||||
sends is forwarded as-is (same bytes, same ``more_body``) — the
|
||||
wrapper mutates ONLY the ``http.response.start`` message."""
|
||||
original_frames: list[Message] = []
|
||||
|
||||
async def recording_app(scope: Scope, receive: Any, send: Any) -> None:
|
||||
# Same response shape as the SSE route: start, then two frames.
|
||||
await send(
|
||||
{
|
||||
"type": "http.response.start",
|
||||
"status": 200,
|
||||
"headers": [[b"content-type", b"text/event-stream"]],
|
||||
}
|
||||
)
|
||||
for frame in (b"data: a\n\n", b"data: b\n\n"):
|
||||
original_frames.append(
|
||||
{"type": "http.response.body", "body": frame, "more_body": frame != b"data: b\n\n"}
|
||||
)
|
||||
await send(original_frames[-1])
|
||||
|
||||
sent = _drive(SecurityHeadersMiddleware(recording_app), _http_scope())
|
||||
|
||||
forwarded = [m for m in sent[1:] if m["type"] == "http.response.body"]
|
||||
# Identical objects: forwarded byte-for-byte, in order, untouched.
|
||||
assert forwarded == original_frames
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Scope handling: http wrapping vs non-http passthrough
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_http_scope_wraps_send_but_not_receive() -> None:
|
||||
seen: dict[str, Any] = {}
|
||||
|
||||
async def app(scope: Scope, receive: Any, send: Any) -> None:
|
||||
seen["receive"] = receive
|
||||
seen["send"] = send
|
||||
await send({"type": "http.response.start", "status": 200, "headers": []})
|
||||
await send({"type": "http.response.body", "body": b"", "more_body": False})
|
||||
|
||||
queued: list[Message] = [{"type": "http.request", "body": b"", "more_body": False}]
|
||||
sent: list[Message] = []
|
||||
|
||||
async def receive() -> Message:
|
||||
return queued.pop(0) if queued else {"type": "http.disconnect"}
|
||||
|
||||
async def send(message: Message) -> None:
|
||||
sent.append(message)
|
||||
|
||||
asyncio.run(SecurityHeadersMiddleware(app)(_http_scope(), receive, send))
|
||||
|
||||
assert seen["send"] is not send # the wrapped send did the header writes
|
||||
assert seen["receive"] is receive # receive is never touched
|
||||
assert sent[0]["type"] == "http.response.start"
|
||||
_assert_security_headers(sent[0])
|
||||
|
||||
|
||||
@pytest.mark.parametrize("scope_type", ["websocket", "lifespan"], ids=["websocket", "lifespan"])
|
||||
def test_non_http_scope_passes_through_untouched(scope_type: str) -> None:
|
||||
"""Non-HTTP connections (websocket, lifespan) reach the app with the
|
||||
ORIGINAL ``send``/``receive`` — no wrapping, no header injection
|
||||
attempt, no crash."""
|
||||
seen: dict[str, Any] = {}
|
||||
|
||||
first, second = {
|
||||
"websocket": (
|
||||
{"type": "websocket.accept"},
|
||||
{"type": "websocket.send", "text": "hi"},
|
||||
),
|
||||
"lifespan": (
|
||||
{"type": "lifespan.startup.complete"},
|
||||
{"type": "lifespan.shutdown.complete"},
|
||||
),
|
||||
}[scope_type]
|
||||
|
||||
async def app(scope: Scope, receive: Any, send: Any) -> None:
|
||||
seen["scope"] = scope
|
||||
seen["receive"] = receive
|
||||
seen["send"] = send
|
||||
await send(first)
|
||||
await send(second)
|
||||
|
||||
queued: list[Message] = [
|
||||
{"type": f"{scope_type}.connect"} if scope_type == "websocket"
|
||||
else {"type": "lifespan.startup"}
|
||||
]
|
||||
sent: list[Message] = []
|
||||
|
||||
async def receive() -> Message:
|
||||
return queued.pop(0) if queued else {"type": "lifespan.shutdown"}
|
||||
|
||||
async def send(message: Message) -> None:
|
||||
sent.append(message)
|
||||
|
||||
scope = {"type": scope_type, "path": "/", "raw_path": b"/"}
|
||||
asyncio.run(SecurityHeadersMiddleware(app)(scope, receive, send))
|
||||
|
||||
assert seen["scope"] is scope
|
||||
assert seen["send"] is send # the ORIGINAL send — the fast path
|
||||
assert seen["receive"] is receive
|
||||
# Both messages pass through untouched — no headers injected anywhere.
|
||||
assert sent == [first, second]
|
||||
Reference in New Issue
Block a user