Files
Anthias/tests/test_asset_headers.py
Viktor Petersson eaf6d9fcd5 feat(webview): per-asset custom HTTP headers for web assets (#3162)
* feat(webview): per-asset custom HTTP headers for web assets

Adds a per-asset setting that injects custom request headers (e.g. an
Authorization token for a private Grafana dashboard) into QtWebEngine
when a webpage asset is displayed. Works on both Qt5 and Qt6 via one
non-gated code path. Closes the request in the linked feature issue.

- Store headers in Asset.metadata['headers'] as {name: value}; no
  migration (mirrors the refresh_interval_s feature end-to-end)
- Validate/sanitise header names (RFC 7230 token) and reject CR/LF/NUL
  in values to prevent header splitting; cap count and length
- v2 API: custom_headers read + write fields folded into metadata
- Edit modal: webpage-only "Custom HTTP headers" textarea, one
  Name: Value per line, parsed + clamped in assets_update
- Viewer sends headers as JSON over a new setRequestHeaders D-Bus slot
  (version-skew latched like setReloadInterval); reloads on header change
- C++ RequestHeaderInterceptor on the shared QWebEngineProfile, scoped
  to the asset's own host so tokens never leak to third-party requests
- Fix host-only test isolation: move shared redis/dbus/env mocks to a
  root conftest.py so the api/tests tree inherits them, and reset the
  ViewerPublisher/ReplyCollector singletons per test
- Add unit, v2 API, and viewer D-Bus tests

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(viewer): self-heal a transient setRequestHeaders failure

Hardware testing on x86/Qt6 surfaced a race: the first setRequestHeaders
D-Bus call after a cold webview spawn can lose to the webview's D-Bus
registration and fail transiently. On a single-asset playlist the
unchanged-URL short-circuit then skips loadPage forever, so the C++
interceptor never receives the staged headers and the page loads without
them.

- _apply_request_headers now returns whether the headers are in effect
- view_webpage only caches current_browser_headers on success, so a
  transient failure leaves the mismatch in place and the next asset_loop
  tick re-issues both setRequestHeaders and loadPage
- add a regression test for the transient-retry path

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(tests): make root conftest a no-op when the app stack is absent

The root conftest that carries the Redis/D-Bus/env isolation is also an
ancestor of tools/raspberry_pi_imager/tests, which the run-python-linter
job runs with -p no:django in a minimal venv (no Django, no pytz). Its
import-time connect_to_redis patch pulled in anthias_common.utils ->
pytz and crashed conftest collection for those app-free tests.

Gate every app-dependent hook (connect_to_redis patch, _mock_redis and
_ensure_assetdir autouse fixtures) on an import probe so the file
degrades to a no-op when the app isn't importable, while still providing
full isolation for the app test suites.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(review): origin-scope injected headers, fix interceptor lifetime

Addresses PR review (Copilot + CodeQL):

- Scope custom headers to the asset's full origin (scheme + host + port)
  instead of host only, so an Authorization header can't ride an
  https->http downgrade or reach a different port on the same host.
- Parent the request interceptor to the process-lifetime default profile
  (not the View) and detach it in ~View, removing a potential
  use-after-free where the profile could outlive the interceptor.
- v2 custom_headers validation raises a fixed, self-authored message
  rather than echoing the caught exception text (CodeQL "information
  exposure through an exception").

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(review): cache a copy of applied request headers, not the caller's dict

Copilot re-review: assigning current_browser_headers = headers aliased
the caller's dict, so a later in-place mutation could make the next
value comparison spuriously equal and skip a needed reload. Store
dict(headers) instead.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(viewer): defer navigation when request headers can't be applied

Copilot re-review (security): on a transient setRequestHeaders failure
during an asset transition, view_webpage still issued loadPage. The C++
side then reused the PREVIOUS asset's staged headers and scoped them to
the new asset's origin — leaking e.g. a prior asset's Authorization
header to the next asset's host.

Now loadPage is skipped entirely when the header update didn't apply;
the URL/header cache is left stale so the next asset_loop tick retries
once the D-Bus call succeeds. Supersedes the earlier "cache only on
success" tweak.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(review): reject non-string custom-header values instead of coercing

Copilot re-review: DictField(child=CharField(...)) coerced numeric /
boolean JSON values to strings ({"X": 123} -> "123") before
validate_asset_headers could reject them, so the strict validation was
bypassed. Drop the CharField child (use the unvalidated child) on both
the create and update serializers so values reach validate_asset_headers
verbatim and its isinstance(str) gate 400s non-strings. Also avoids
CharField trimming a value we forward on the wire byte-for-byte.

Adds numeric / bool / null rejection cases to the API test.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* docs(review): say same-origin, not same-host, throughout

Copilot re-review (8 doc nits): the origin-scoping change left comments,
docstrings, and the edit-modal copy still describing "same-host" /
"same-site" scoping. Align all of them with the implementation
(same-origin = scheme + host + port), and correct the root conftest
comment to note ENVIRONMENT/dbus-stub setup still runs while only the
app-dependent hooks no-op when the app stack is absent. No logic change.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(review): narrow conftest probe + validate headers at the D-Bus boundary

Copilot re-review:
- conftest app-availability probe caught bare Exception, which would
  silently disable Redis isolation for the whole app suite on any real
  import error. Narrow to ImportError (the missing-pytz/Django case);
  let genuine app import failures surface.
- C++ setRequestHeaders forwarded arbitrary JSON names/values into
  info.setHttpHeader() unchecked. Add defensive re-validation at this
  trust-no-one D-Bus boundary (RFC 7230 name token, reject CR/LF/NUL
  values, cap count/lengths) mirroring validate_asset_headers, so a
  hostile/buggy caller can't put splitting bytes on the wire even though
  the server side is the primary gate.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(review): byte-length header cap + doc accuracy

Copilot re-review:
- MAX_HEADER_VALUE_LEN now caps the UTF-8 byte length (value.encode),
  not the Python character count, matching the webview's byte-based cap
  (values go on the wire as UTF-8).
- models.py comment updated to same-origin, and to note the server side
  is the primary gate with the webview re-validating defensively.
- conftest docstring: correct the module path to
  src/anthias_viewer/__init__.py.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-09 09:50:33 +01:00

114 lines
3.7 KiB
Python

"""Unit tests for the per-asset custom-header helpers (feature #2215).
These cover the pure ``metadata['headers']`` sanitisers in
``anthias_server.app.models`` in isolation — no DB, no HTTP — so the
name/value rules that keep CR/LF off the wire are pinned independently of
the serializer and form layers that call them.
"""
import pytest
from anthias_server.app.models import (
MAX_ASSET_HEADERS,
normalize_asset_headers,
parse_header_lines,
validate_asset_headers,
)
class TestNormalizeAssetHeaders:
def test_passes_valid_headers_through(self) -> None:
headers = {'Authorization': 'Bearer abc', 'X-Env': 'prod'}
assert normalize_asset_headers(headers) == headers
@pytest.mark.parametrize(
'value',
[None, 'not-a-dict', 42, [], ('a', 'b')],
)
def test_non_dict_becomes_empty(self, value: object) -> None:
assert normalize_asset_headers(value) == {}
def test_drops_crlf_values(self) -> None:
result = normalize_asset_headers(
{'Good': 'ok', 'Bad': 'a\r\nEvil: 1', 'Also': 'b\nc'}
)
assert result == {'Good': 'ok'}
def test_drops_null_byte_values(self) -> None:
assert normalize_asset_headers({'X': 'a\x00b'}) == {}
def test_drops_non_token_names(self) -> None:
result = normalize_asset_headers(
{'sp ace': 'v', 'co:lon': 'v', '': 'v', 'Ok': 'v'}
)
assert result == {'Ok': 'v'}
def test_drops_non_string_entries(self) -> None:
assert normalize_asset_headers({'X': 1, 2: 'v', 'Y': 'ok'}) == {
'Y': 'ok'
}
def test_strips_whitespace_around_name(self) -> None:
assert normalize_asset_headers({' X-Env ': 'prod'}) == {
'X-Env': 'prod'
}
def test_caps_at_max(self) -> None:
result = normalize_asset_headers(
{f'X-H{i}': 'v' for i in range(MAX_ASSET_HEADERS + 5)}
)
assert len(result) == MAX_ASSET_HEADERS
class TestValidateAssetHeaders:
def test_returns_clean_dict(self) -> None:
headers = {'Authorization': 'Bearer x'}
assert validate_asset_headers(headers) == headers
@pytest.mark.parametrize(
'value',
[
{'Bad Name': 'v'},
{'X': 'a\r\nb'},
{'': 'v'},
{'X:': 'v'},
{'X': 1},
'not-a-dict',
],
)
def test_raises_on_malformed(self, value: object) -> None:
with pytest.raises(ValueError):
validate_asset_headers(value)
def test_raises_over_count_cap(self) -> None:
with pytest.raises(ValueError):
validate_asset_headers(
{f'X-H{i}': 'v' for i in range(MAX_ASSET_HEADERS + 1)}
)
class TestParseHeaderLines:
def test_parses_name_value_lines(self) -> None:
text = 'Authorization: Bearer abc\nX-Env: prod'
assert parse_header_lines(text) == {
'Authorization': 'Bearer abc',
'X-Env': 'prod',
}
def test_ignores_blank_and_colonless_lines(self) -> None:
text = '\n\nAuthorization: x\njust-a-comment\n\n'
assert parse_header_lines(text) == {'Authorization': 'x'}
def test_value_keeps_later_colons(self) -> None:
assert parse_header_lines('X-Range: bytes=0-1:2') == {
'X-Range': 'bytes=0-1:2'
}
def test_drops_malformed_lines(self) -> None:
# A CRLF can't survive splitlines(), but a bad name still drops.
assert parse_header_lines('bad name: v\nOk: v') == {'Ok': 'v'}
@pytest.mark.parametrize('value', [None, 42, []])
def test_non_string_becomes_empty(self, value: object) -> None:
assert parse_header_lines(value) == {}