signal-bot/tests/test_manage_guard.py
Krzysztof kuhy Rudnicki 0b539fc327 feat: the management backend, absorbing the tagger
A second process rather than more routes on the bot's own app. This one can
rewrite .env, and the bot's port already serves /_selftest, which posts for
real into a real group; keeping them apart means the surface that can
reconfigure the bot only listens while somebody has the page open.

The tagger's seven routes move here unchanged, out of tool/ and into the
coverage gate -- which is the reason an unconditional ImportError once sat
in tagger.py through two releases. The media route is still indexed by
integer, and that is not a style choice: a filename from a client is a path,
and the directory holds private photographs. FastAPI refuses a name with 422
before the handler runs, which the tests assert.

guard.py is the part that was not asked for. Binding to loopback stops the
network, not the browser already running on this machine, and two attacks
reach a loopback port from an ordinary web page: a cross-origin form, which
can only send urlencoded, multipart or plain text, so requiring
application/json refuses it with no token to keep; and DNS rebinding, whose
request still carries the attacker's hostname in Host. Forty lines instead
of authentication, which is the right trade for something started by hand
and stopped again a few minutes later.

assets.py branches on whether the page has been built, because StaticFiles
raises at construction on a missing directory -- mounting it unconditionally
would take the app down on a fresh clone, and in CI, where pytest never runs
pnpm build. Both branches are tested; "not built" answers with the command.

The container restart goes through signal_bot.external, wrapped in
asyncio.to_thread: it is synchronous with a 60s default, and blocking the
loop would freeze the page showing the restart.

100% branch coverage across all eleven modules, no omits, no pragmas.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01416TegJ3sa8xL5qGGgrygP
2026-08-31 14:38:24 +02:00

86 lines
2.9 KiB
Python

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The two checks that make a loopback bind actually mean loopback."""
from __future__ import annotations
from typing import TYPE_CHECKING
import pytest
from signal_bot.manage import guard
from tests.manage_fixture import BASE_URL, PORT, client_for, deps_at
if TYPE_CHECKING:
from pathlib import Path
@pytest.mark.parametrize(
"host",
["127.0.0.1:8140", "localhost:8140", "127.0.0.1", "localhost", "[::1]:8140"],
)
def test_a_loopback_host_is_this_machine_asking(host: str) -> None:
assert guard.allowed_host(host, PORT) is True
@pytest.mark.parametrize(
"host",
[
# DNS rebinding: the name resolves to 127.0.0.1 by the time the
# request is made, but the browser still sends the attacker's host.
"evil.example.com",
"evil.example.com:8140",
# A different port on this machine is a different application.
"127.0.0.1:9999",
# HTTP/1.1 requires it; something omitting it is not an ordinary
# browser and does not get the benefit of the doubt.
"",
],
)
def test_anything_else_is_refused(host: str) -> None:
assert guard.allowed_host(host, PORT) is False
def test_a_rebound_request_never_reaches_a_route(tmp_path: Path) -> None:
with client_for(deps_at(tmp_path)) as client:
answer = client.get("/api/status", headers={"host": "evil.example.com"})
assert answer.status_code == 421
assert "loopback" in answer.json()["reason"]
def test_a_cross_origin_form_post_cannot_write_anything(tmp_path: Path) -> None:
# A cross-origin <form> can only send urlencoded, multipart or plain
# text. Requiring application/json refuses all three without a token.
with client_for(deps_at(tmp_path)) as client:
answer = client.post(
"/api/restart-container",
content="values=whatever",
headers={"content-type": "application/x-www-form-urlencoded"},
)
assert answer.status_code == 415
assert "application/json" in answer.json()["reason"]
def test_a_content_type_with_a_charset_is_still_json(tmp_path: Path) -> None:
with client_for(deps_at(tmp_path)) as client:
answer = client.put(
"/api/env",
content='{"values": {}}',
headers={"content-type": "application/json; charset=utf-8"},
)
assert answer.status_code == 200
def test_reading_needs_no_content_type_at_all(tmp_path: Path) -> None:
# Requiring one on GET would refuse every ordinary navigation to the page.
with client_for(deps_at(tmp_path)) as client:
assert client.get("/api/status").status_code == 200
def test_the_guard_is_in_front_of_every_route_including_the_page(
tmp_path: Path,
) -> None:
with client_for(deps_at(tmp_path, built=True)) as client:
answer = client.get("/", headers={"host": "evil.example.com"})
assert answer.status_code == 421
assert BASE_URL.endswith(str(PORT))