kuhyx/signal-bot
1
0
Fork 0
mirror of https://github.com/kuhyx/signal-bot.git synced 2026-11-10 19:07:19 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat: the management backend, absorbing the tagger

Browse Source 
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
This commit is contained in:
Krzysztof Rudnicki 2026-08-31 14:38:24 +02:00
parent bdd20654bf
commit 0b539fc327
13 changed files with 1220 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
signal_bot/manage/app.py Normal file
View File

@ -0,0 +1,46 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The management application, assembled.
A second process rather than more routes on the bot's own app. Two reasons,
and both are about blast radius: 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 is only
listening while somebody has the page open.
Order is load-bearing. The guard goes on first so it sees every request; the
API routers are registered next; the static catch-all is registered last,
because ``/{path:path}`` would otherwise shadow every route after it.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from fastapi import FastAPI
from signal_bot.manage import assets, guard, routes_actions, routes_album, routes_env, routes_status
if TYPE_CHECKING:
from signal_bot.manage.deps import ManageDeps
DEFAULT_CONTAINER = "signal-api"
def build_app(deps: ManageDeps, *, container: str = DEFAULT_CONTAINER) -> FastAPI:
"""The whole page, bound to one bot and one album.
Args:
deps: Everything the app is pointed at.
container: The docker container signal-cli runs in.
Returns:
An application ready for uvicorn, or for a test client.
"""
app = FastAPI(title="signal-bot", docs_url=None, redoc_url=None, openapi_url=None)
guard.install(app, deps.port)
app.include_router(routes_status.router(deps))
app.include_router(routes_env.router(deps))
app.include_router(routes_actions.router(deps, container))
app.include_router(routes_album.router(deps))
assets.install(app, deps.dist_dir)
return app

75
signal_bot/manage/assets.py Normal file
View File

@ -0,0 +1,75 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Serving the built page, and saying so plainly when it is not built.
``StaticFiles`` raises at construction when its directory is missing, so
mounting it unconditionally would make the whole app fail to start on a fresh
clone -- and in CI, where pytest runs without ever invoking ``pnpm build``.
The branch is therefore explicit: built, or a 503 that names the command to
run. "Not built yet" is a normal state with an obvious fix, and it deserves a
sentence rather than a traceback.
No new dependency. Starlette's ``StaticFiles`` is anyio-based and needs no
``aiofiles``, which is verified rather than assumed: this repository's runtime
pins are exactly six packages and adding a seventh would trip the freshness
gate.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Final
from fastapi import FastAPI, Response, status
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
if TYPE_CHECKING:
from pathlib import Path
INDEX: Final = "index.html"
ASSETS: Final = "assets"
BUILD_COMMAND: Final = "pnpm --dir web install && pnpm --dir web build"
NOT_BUILT: Final = (
"<!doctype html><meta charset=utf-8><title>signal-bot</title>"
"<p>The management page has not been built yet. Run:"
f"<pre>{BUILD_COMMAND}</pre>"
)
def is_built(dist: Path) -> bool:
"""Whether there is a built page to serve.
Args:
dist: The directory ``vite build`` writes into.
Returns:
True when both halves of a vite build are there. Both, because
mounting StaticFiles on a missing ``assets/`` raises exactly the way
a missing ``dist/`` does, and a half-built directory should answer
"not built" rather than take the app down with it.
"""
return (dist / INDEX).is_file() and (dist / ASSETS).is_dir()
def install(app: FastAPI, dist: Path) -> None:
"""Serve the built page, or explain its absence, at every unclaimed path.
Registered last, after every API route, so the catch-all cannot shadow
them. The catch-all exists because the page routes client-side: a reload
on a sub-view asks this server for a path only the browser knows about.
Args:
app: The FastAPI application to serve from.
dist: The directory ``vite build`` writes into.
"""
if is_built(dist):
# Hashed filenames under /assets, so they are safe to cache and are
# the only paths that must not fall through to index.html.
app.mount(f"/{ASSETS}", StaticFiles(directory=dist / ASSETS), name=ASSETS)
@app.get("/{_path:path}", response_class=HTMLResponse)
async def page(_path: str, response: Response) -> str:
if not is_built(dist):
response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
return NOT_BUILT
return (dist / INDEX).read_text(encoding="utf-8")

49
signal_bot/manage/deps.py Normal file
View File

@ -0,0 +1,49 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Everything the management app is pointed at, as one value.
Every path is a field rather than a module constant. That is what lets the
tests point the whole app at ``tmp_path`` and never go near the real ``.env``
(mode 0600, holding the live account) or the real album (mode 0700, holding
photographs of people who did not agree to be in a test fixture).
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from pathlib import Path
import httpx
from signal_bot.config import Config
from signal_bot.pack_album import Album
@dataclass(frozen=True, slots=True)
class ManageDeps:
"""What one running management page is bound to."""
config: Config
"""The bot's own configuration, already validated by load_config."""
env_path: Path
"""The .env the page edits. Not derived: the tests point it elsewhere."""
dist_dir: Path
"""Where `pnpm build` puts the page. Absent until it has been run."""
album: Album
"""The one pack this session tags, chosen by --dir."""
client: httpx.AsyncClient
"""Shared, so the proxied /healthz and the group listing pool sockets."""
port: int
"""The port this app is served on, so the Host guard knows what to allow."""
@property
def bot_url(self) -> str:
"""The bot's own loopback base URL, which this app proxies to."""
return f"http://{self.config.bind_host}:{self.config.bind_port}"

84
signal_bot/manage/guard.py Normal file
View File

@ -0,0 +1,84 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The two checks that make a loopback bind actually mean loopback.
Binding to 127.0.0.1 stops the network. It does not stop the browser already
running on this machine, and this app writes ``.env`` -- the file that decides
which groups the bot answers in and who may add it to one. Two attacks reach a
loopback port from an ordinary web page, and both are cheap to close:
- **Form CSRF.** A page on any site can POST a cross-origin ``<form>`` to
127.0.0.1 without reading the reply. A form can only send
``application/x-www-form-urlencoded``, ``multipart/form-data`` or
``text/plain``, so requiring ``application/json`` on every mutating request
refuses it outright, with no token and no session to keep.
- **DNS rebinding.** A hostname the attacker controls is re-pointed at
127.0.0.1 after the page loads, which makes the browser treat this app as
same-origin. The rebound request still carries the attacker's hostname in
``Host``, so requiring a loopback ``Host`` refuses that too.
Roughly forty lines instead of authentication, which is the right trade for
something started by hand, used for a few minutes and stopped again.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Final
from fastapi import FastAPI, Request, Response, status
from fastapi.responses import JSONResponse
if TYPE_CHECKING:
from collections.abc import Awaitable, Callable
# GET and HEAD do not change anything, and requiring a content type on them
# would break every ordinary navigation to the page itself.
MUTATING: Final = frozenset({"POST", "PUT", "PATCH", "DELETE"})
JSON_TYPE: Final = "application/json"
LOOPBACK_HOSTS: Final = ("127.0.0.1", "localhost", "[::1]")
def allowed_host(host: str, port: int) -> bool:
"""Whether a Host header names this app on this machine.
Args:
host: The request's Host header, possibly with a port.
port: The port this app is serving on.
Returns:
True for a loopback name, with either the right port or none.
"""
if not host:
# HTTP/1.1 requires it. Something that omits it is not a browser
# doing anything ordinary, so it does not get the benefit of the doubt.
return False
name, _, given = host.rpartition(":")
if not name:
name, given = host, ""
return name in LOOPBACK_HOSTS and given in {"", str(port)}
def install(app: FastAPI, port: int) -> None:
"""Add both checks to one app, in front of every route.
Args:
app: The FastAPI application to guard.
port: The port it will be served on.
"""
@app.middleware("http")
async def check(
request: Request, call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
if not allowed_host(request.headers.get("host", ""), port):
return JSONResponse(
{"status": "refused", "reason": "this page is only served on loopback"},
status_code=status.HTTP_421_MISDIRECTED_REQUEST,
)
if request.method in MUTATING:
declared = request.headers.get("content-type", "").split(";")[0].strip()
if declared != JSON_TYPE:
return JSONResponse(
{"status": "refused", "reason": f"{JSON_TYPE} is required"},
status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
)
return await call_next(request)

86
signal_bot/manage/routes_actions.py Normal file
View File

@ -0,0 +1,86 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The two things this page does rather than reads.
Both are deliberately thin. ``/api/selftest`` proxies the bot's own endpoint
instead of reimplementing it, so ``SELFTEST_GROUPS`` stays fail-closed in
exactly one place and this page cannot become a second way to post into a
group. The container restart goes through ``signal_bot.external``, the one
module in this repository allowed to start a process -- a direct
``subprocess`` call here would fail both ``ruff S603`` and the test that
asserts by name which files may import it.
"""
from __future__ import annotations
import asyncio
from typing import TYPE_CHECKING, Any
import httpx
from fastapi import APIRouter, HTTPException, status
from pydantic import BaseModel
from signal_bot import external
if TYPE_CHECKING:
from signal_bot.manage.deps import ManageDeps
# external.run is synchronous and defaults to a 60s timeout. A container
# restart takes seconds, and blocking the event loop for them would freeze
# the page that is showing you the restart.
RESTART_TIMEOUT = 90.0
class SelfTest(BaseModel):
"""One command to type at the bot, and where."""
group: str
text: str
attachments: list[str] = []
def router(deps: ManageDeps, container: str) -> APIRouter:
"""The selftest proxy and the container restart.
Args:
deps: The running app's configuration and shared HTTP client.
container: The docker container signal-cli runs in.
Returns:
A router to include in the management app.
"""
api = APIRouter(prefix="/api")
@api.post("/selftest")
async def selftest(request: SelfTest) -> dict[str, Any]:
"""Ask the bot to type a command at itself, and pass on what it says."""
try:
response = await deps.client.post(
f"{deps.bot_url}/_selftest",
json=request.model_dump(),
timeout=30.0,
)
except httpx.HTTPError as error:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"the bot did not answer: {error}",
) from error
# Verbatim, including 403 and 503: the refusal belongs to the bot,
# and rewording it here would make two different sentences for one
# rule.
body: dict[str, Any] = response.json()
if response.status_code != status.HTTP_200_OK:
raise HTTPException(status_code=response.status_code, detail=body)
return body
@api.post("/restart-container")
async def restart() -> dict[str, Any]:
"""Restart signal-cli's container, and report what actually happened."""
result = await asyncio.to_thread(
external.run, ["docker", "restart", container], timeout=RESTART_TIMEOUT
)
# external.run never raises: a non-zero exit, a missing binary and a
# timeout all arrive as ok=False, which is the one thing this must
# not be able to mistake for success.
return {"ok": result.ok, "stdout": result.stdout, "stderr": result.stderr}
return api

114
signal_bot/manage/routes_album.py Normal file
View File

@ -0,0 +1,114 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The photo tagger, absorbed.
These are ``tool/tagger.py``'s seven routes, moved here so they are inside
the coverage gate rather than beside it -- which is the reason an
unconditional ImportError once sat in that file through two releases. The
behaviour is unchanged, deliberately: ``Album`` already deals in ``KeyError``
and ``ConfigError`` so routes can translate them, and the media route is
still indexed by integer.
That last one is not a style choice. A filename supplied by a client is a
path, and this directory holds private photographs.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from fastapi import APIRouter, HTTPException, status
from fastapi.responses import FileResponse
from signal_bot.groups import ConfigError
from signal_bot.manage.routes_status import live_groups
from signal_bot.pack import MEDIA_DIR
if TYPE_CHECKING:
from signal_bot.manage.deps import ManageDeps
def roster(api: APIRouter, deps: ManageDeps) -> None:
"""The two endpoints that write people.json.
Split from :func:`router` along the seam that matters: everything here
can refuse, and a refusal has to reach the page rather than the next
restart.
Args:
api: The router to register on.
deps: The running app's album.
"""
@api.post("/setup")
async def setup(payload: dict[str, Any]) -> dict[str, str]:
"""Write the roster, then prove the bot could load it."""
try:
deps.album.setup(
list(payload.get("groups", [])),
list(payload.get("people", [])),
list(payload.get("combos", [])),
)
except ConfigError as error:
# Straight into the page: a manifest the bot would refuse should
# be fixed here, in the form, not found at the next restart.
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail=str(error)
) from error
return {"status": "ok"}
@api.post("/person")
async def person(payload: dict[str, Any]) -> dict[str, str]:
"""Add one person mid-session, and return the key they were given."""
# Mid-session, not only at setup: the reason to add somebody is
# usually a photo of them already on the screen.
try:
key = deps.album.add_person([str(t) for t in payload.get("triggers", [])])
except ConfigError as error:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail=str(error)
) from error
return {"status": "ok", "key": key}
def router(deps: ManageDeps) -> APIRouter:
"""Every album endpoint the page needs.
Args:
deps: The running app's album and HTTP client.
Returns:
A router to include in the management app.
"""
api = APIRouter(prefix="/api/album")
roster(api, deps)
@api.get("/state")
async def state() -> dict[str, Any]:
"""Everything the tagging view draws itself from."""
return {**deps.album.state(), "known_groups": await live_groups(deps)}
@api.post("/tag")
async def tag(payload: dict[str, Any]) -> dict[str, str]:
"""Record who is in one file."""
try:
deps.album.tag(str(payload["file"]), [str(key) for key in payload.get("keys", [])])
except KeyError as error:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(error)) from error
return {"status": "ok"}
@api.get("/summary")
async def summary() -> dict[str, list[str]]:
"""What each command would answer with, and what is still untagged."""
return {"lines": deps.album.summary()}
@api.get("/media/{index}")
async def media(index: int) -> FileResponse:
"""One item, by position in the album."""
# By index, never by a name the page sends back: a filename from a
# client is a path, and portraits.path_for exists for the same reason.
files = deps.album.files
if not 0 <= index < len(files):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="no such item")
return FileResponse(deps.album.directory / MEDIA_DIR / files[index])
return api

81
signal_bot/manage/routes_env.py Normal file
View File

@ -0,0 +1,81 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Reading and writing ``.env`` from the page.
Three refusals, in order, and the order matters. A key that may not be
written is refused before anything is parsed, so a read-only key cannot be
smuggled past by making the rest of the file invalid. Then the whole
candidate file is validated by the function startup uses. Only then is
anything written, so a rejected edit leaves the file byte-identical.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from fastapi import APIRouter, HTTPException, status
from pydantic import BaseModel
from signal_bot.groups import ConfigError
from signal_bot.manage import envfile, envkeys
from signal_bot.manage.routes_status import live_groups
if TYPE_CHECKING:
from signal_bot.manage.deps import ManageDeps
class EnvUpdate(BaseModel):
"""The keys the form changed, and what to set them to."""
values: dict[str, str]
def router(deps: ManageDeps) -> APIRouter:
"""The two ``.env`` endpoints.
Args:
deps: The running app's configuration and the path it may edit.
Returns:
A router to include in the management app.
"""
api = APIRouter(prefix="/api")
@api.get("/env")
async def read() -> dict[str, Any]:
"""Every key the page shows, and whether groups can be edited now."""
text = deps.env_path.read_text(encoding="utf-8")
groups = await live_groups(deps)
return {
"keys": envkeys.settings(text),
# False turns the two group fields read-only in the form rather
# than letting them be written blind. See envkeys.NEEDS_GROUPS.
"groups_known": bool(groups),
"needs_groups": list(envkeys.NEEDS_GROUPS),
}
@api.put("/env")
async def write(update: EnvUpdate) -> dict[str, Any]:
"""Validate an edit against startup's own rules, then write it."""
groups = await live_groups(deps)
refused = envkeys.refuse(update.values, groups_known=bool(groups))
if refused is not None:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"{refused} is not editable from this page",
)
text = deps.env_path.read_text(encoding="utf-8")
try:
envkeys.validate(text, update.values)
except ConfigError as error:
# load_config's own message: the page shows what the journal
# would have shown at the next restart, before that restart.
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail=str(error)
) from error
envfile.write(deps.env_path, update.values)
# The running process loaded the old values and will keep using them.
# Saying so is the whole of the contract; restarting is the
# operator's to do, from the terminal they started this page in.
return {"status": "written", "restart_required": True}
return api

102
signal_bot/manage/routes_status.py Normal file
View File

@ -0,0 +1,102 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""What the bot currently is: alive or not, in which groups, configured how.
All three reads degrade rather than fail. The management page is most useful
exactly when something is down, so a page that cannot render because the
container is unreachable would be missing at the only moment it was wanted.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
import httpx
from fastapi import APIRouter
if TYPE_CHECKING:
from signal_bot.manage.deps import ManageDeps
# Enough to recognise the number, not enough to be the number. The page runs
# on the operator's own machine, but this repository has leaked a phone number
# into a public place once already and the fix was to delete the repository.
VISIBLE_DIGITS = 4
async def live_groups(deps: ManageDeps) -> list[dict[str, str]]:
"""The groups this account is in, so an id is picked and never typed.
Lifted from ``tool/tagger.py``, where it existed for the same reason: a
group id is 44 characters of base64 and retyping one is how you end up
scoping the bot to a group that does not exist.
Args:
deps: The running app's configuration and shared HTTP client.
Returns:
Every group with an id, or empty when the container cannot be
reached. Empty is a real answer -- see ``envkeys.refuse``, which
turns it into "group settings are read-only right now".
"""
try:
response = await deps.client.get(deps.config.groups_url, timeout=5.0)
response.raise_for_status()
listing = response.json()
except httpx.HTTPError, ValueError:
# No container, no network, or something that is not JSON at the
# other end. None of them stop the rest of the page working.
return []
return [
{"id": str(entry.get("id", "")), "name": str(entry.get("name", ""))}
for entry in listing
if entry.get("id")
]
def router(deps: ManageDeps) -> APIRouter:
"""The three read-only status endpoints.
Args:
deps: The running app's configuration, client and album.
Returns:
A router to include in the management app.
"""
api = APIRouter(prefix="/api")
@api.get("/status")
async def status() -> dict[str, Any]:
"""Whether the bot answers, and what it says about itself."""
try:
response = await deps.client.get(f"{deps.bot_url}/healthz", timeout=5.0)
body = response.json()
except httpx.HTTPError, ValueError:
# The bot being down is the single most important thing this
# page can tell you, so it is a state and never an error.
return {"reachable": False, "bot": {}}
return {"reachable": True, "bot": body}
@api.get("/groups")
async def groups() -> dict[str, list[dict[str, str]]]:
"""The live group listing, or empty when the container is unreachable."""
return {"groups": await live_groups(deps)}
@api.get("/config")
async def config() -> dict[str, Any]:
"""The resolved configuration, as the running bot sees it."""
settings = deps.config
return {
"phone_number": f"…{settings.phone_number[-VISIBLE_DIGITS:]}",
"is_pinned": settings.is_pinned,
"group_ids": list(settings.group_ids),
"trusted_inviters_count": len(settings.trusted_inviters),
"api_base_url": settings.api_base_url,
"database_path": settings.database_path,
"photo_pack_dirs": [str(path) for path in settings.photo_pack_dirs],
"report_at": f"{settings.report_hour:02d}:{settings.report_minute:02d}",
"cooldown_seconds": settings.cooldown_seconds,
"silence_seconds": settings.silence_seconds,
"log_level": settings.log_level,
"bind": f"{settings.bind_host}:{settings.bind_port}",
}
return api

116
tests/manage_fixture.py Normal file
View File

@ -0,0 +1,116 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""A whole management app, pointed entirely at a temp directory.
Every path the app touches is a field on ManageDeps, so nothing here can
reach the real .env or the real album. The bot at the other end is an
httpx.MockTransport, so no test needs a container, a network or a phone.
Deliberately a builder rather than a fixture: the tests want apps whose bot
is up, down, or answering nonsense, and saying which belongs in the test.
"""
from __future__ import annotations
import json
from typing import TYPE_CHECKING
import httpx
from fastapi.testclient import TestClient
from signal_bot.config import Config
from signal_bot.manage.app import build_app
from signal_bot.manage.deps import ManageDeps
from signal_bot.pack import MEDIA_DIR, PEOPLE_FILE, TAGS_FILE
from signal_bot.pack_album import Album
from tests.conftest import WATCHED_GROUP
from tests.pack_fixture import PIXEL
if TYPE_CHECKING:
from collections.abc import Callable
from pathlib import Path
PORT = 8140
# The guard refuses anything that is not loopback, so every client in the
# suite has to say so -- which is also the only way the refusal gets tested.
BASE_URL = f"http://127.0.0.1:{PORT}"
# Invented, and short on purpose: the personal-data gate blocks anything
# matching group.<16+ base64 chars>, so a realistic-looking fixture id would
# make this repository refuse every future commit.
FIXTURE_GROUP = "group.FIXTURE-ALPHA"
ENV_TEXT = """\
# The account, in +E.164 form.
PHONE_NUMBER=+15550000000
# Empty means every group this account is in.
GROUP_IDS=
COOLDOWN_SECONDS=10
"""
HEALTHY = {"status": "ok", "version": "0.0.0", "pope": "Petrus", "papacy": "nr 1"}
def album_at(root: Path) -> Album:
"""A two-item album with one person, on disk under ``root``."""
media = root / MEDIA_DIR
media.mkdir(parents=True)
(media / "one.png").write_bytes(PIXEL)
(media / "two.png").write_bytes(PIXEL)
(root / PEOPLE_FILE).write_text(
json.dumps({"groups": [WATCHED_GROUP], "people": [{"key": "p1", "triggers": ["!alfa"]}]}),
encoding="utf-8",
)
(root / TAGS_FILE).write_text(json.dumps({"one.png": ["p1"]}), encoding="utf-8")
return Album(root)
def bot_is_up(request: httpx.Request) -> httpx.Response:
"""A container and a bot that both answer normally."""
if request.url.path == "/healthz":
return httpx.Response(200, json=HEALTHY)
if request.url.path.endswith("/_selftest"):
return httpx.Response(200, json={"status": "delivered", "group": FIXTURE_GROUP})
return httpx.Response(200, json=[{"id": FIXTURE_GROUP, "name": "alfa"}, {"name": "no id here"}])
def bot_is_down(_request: httpx.Request) -> httpx.Response:
"""Nothing at the other end at all."""
msg = "connection refused"
raise httpx.ConnectError(msg)
def answers_junk(_request: httpx.Request) -> httpx.Response:
"""Something is listening, but it is not speaking JSON."""
return httpx.Response(200, text="<html>not json</html>")
def deps_at(
tmp_path: Path,
handler: Callable[[httpx.Request], httpx.Response] = bot_is_up,
*,
built: bool = False,
) -> ManageDeps:
"""Everything a management app needs, all of it under ``tmp_path``."""
env_path = tmp_path / ".env"
env_path.write_text(ENV_TEXT, encoding="utf-8")
env_path.chmod(0o600)
dist = tmp_path / "dist"
if built:
(dist / "assets").mkdir(parents=True)
(dist / "assets" / "app.js").write_text("// built", encoding="utf-8")
(dist / "index.html").write_text("<title>built</title>", encoding="utf-8")
return ManageDeps(
config=Config(phone_number="+15550000000"),
env_path=env_path,
dist_dir=dist,
album=album_at(tmp_path / "pack"),
client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
port=PORT,
)
def client_for(deps: ManageDeps) -> TestClient:
"""A test client that talks to the app the way a browser on this box does."""
return TestClient(build_app(deps), base_url=BASE_URL)

118
tests/test_manage_album.py Normal file
View File

@ -0,0 +1,118 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The tagger's seven routes, absorbed and still behaving the same."""
from __future__ import annotations
import json
from typing import TYPE_CHECKING
from signal_bot.pack import PEOPLE_FILE, TAGS_FILE
from tests.manage_fixture import FIXTURE_GROUP, bot_is_down, client_for, deps_at
if TYPE_CHECKING:
from pathlib import Path
def test_the_state_carries_the_album_and_the_groups_to_pick_from(
tmp_path: Path,
) -> None:
with client_for(deps_at(tmp_path)) as client:
body = client.get("/api/album/state").json()
assert [entry["name"] for entry in body["files"]] == ["one.png", "two.png"]
assert body["known_groups"] == [{"id": FIXTURE_GROUP, "name": "alfa"}]
def test_tagging_is_offline_work_and_survives_an_unreachable_container(
tmp_path: Path,
) -> None:
with client_for(deps_at(tmp_path, bot_is_down)) as client:
assert client.get("/api/album/state").json()["known_groups"] == []
def test_tagging_one_file_records_who_is_in_it(tmp_path: Path) -> None:
deps = deps_at(tmp_path)
with client_for(deps) as client:
answer = client.post("/api/album/tag", json={"file": "two.png", "keys": ["p1"]})
assert answer.json() == {"status": "ok"}
tags = json.loads((deps.album.directory / TAGS_FILE).read_text(encoding="utf-8"))
assert tags["two.png"] == ["p1"]
def test_tagging_a_file_the_album_does_not_have_is_a_404(tmp_path: Path) -> None:
with client_for(deps_at(tmp_path)) as client:
answer = client.post("/api/album/tag", json={"file": "nope.png", "keys": []})
assert answer.status_code == 404
def test_the_summary_says_what_each_command_would_answer_with(
tmp_path: Path,
) -> None:
with client_for(deps_at(tmp_path)) as client:
lines = client.get("/api/album/summary").json()["lines"]
assert lines[0].startswith("== 2 items")
def test_media_is_served_by_position_in_the_album(tmp_path: Path) -> None:
with client_for(deps_at(tmp_path)) as client:
answer = client.get("/api/album/media/0")
assert answer.status_code == 200
assert answer.content.startswith(b"\x89PNG")
def test_an_index_outside_the_album_is_a_404_at_either_end(tmp_path: Path) -> None:
with client_for(deps_at(tmp_path)) as client:
assert client.get("/api/album/media/-1").status_code == 404
assert client.get("/api/album/media/2").status_code == 404
def test_a_filename_cannot_be_asked_for_at_all(tmp_path: Path) -> None:
# The whole defence: the route takes an int, so FastAPI refuses a name
# before the handler runs. This directory holds private photographs, and
# a filename from a client is a path.
with client_for(deps_at(tmp_path)) as client:
# 422 from FastAPI's own coercion, before any handler code runs.
assert client.get("/api/album/media/one.png").status_code == 422
assert client.get("/api/album/media/..%2F..%2Fetc%2Fpasswd").status_code != 200
def test_the_roster_can_be_written_from_the_page(tmp_path: Path) -> None:
deps = deps_at(tmp_path)
with client_for(deps) as client:
answer = client.post(
"/api/album/setup",
json={
"groups": [FIXTURE_GROUP],
"people": [{"key": "p1", "triggers": ["!alfa"]}],
"combos": [],
},
)
assert answer.json() == {"status": "ok"}
people = json.loads((deps.album.directory / PEOPLE_FILE).read_text(encoding="utf-8"))
assert people["groups"] == [FIXTURE_GROUP]
def test_a_roster_the_bot_could_not_load_is_refused_at_the_form(
tmp_path: Path,
) -> None:
# Not found at the next restart: a manifest the bot would refuse should
# be fixed here, while the person who wrote it is still looking at it.
with client_for(deps_at(tmp_path)) as client:
answer = client.post(
"/api/album/setup",
json={"groups": [], "people": [{"key": "p1", "triggers": ["!a"]}], "combos": []},
)
assert answer.status_code == 400
def test_a_person_can_be_added_mid_session(tmp_path: Path) -> None:
# The reason to add somebody is usually a photo of them already on screen.
with client_for(deps_at(tmp_path)) as client:
answer = client.post("/api/album/person", json={"triggers": ["!delta"]})
assert answer.json()["status"] == "ok"
assert answer.json()["key"].startswith("p")
def test_adding_a_person_with_no_triggers_is_refused(tmp_path: Path) -> None:
with client_for(deps_at(tmp_path)) as client:
answer = client.post("/api/album/person", json={"triggers": []})
assert answer.status_code == 400

61
tests/test_manage_assets.py Normal file
View File

@ -0,0 +1,61 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Serving the built page, and saying so plainly when it is not built."""
from __future__ import annotations
from typing import TYPE_CHECKING
from signal_bot.manage import assets
from tests.manage_fixture import client_for, deps_at
if TYPE_CHECKING:
from pathlib import Path
def test_the_page_is_served_once_it_has_been_built(tmp_path: Path) -> None:
with client_for(deps_at(tmp_path, built=True)) as client:
answer = client.get("/")
assert answer.status_code == 200
assert "built" in answer.text
def test_hashed_assets_are_served_from_their_own_mount(tmp_path: Path) -> None:
with client_for(deps_at(tmp_path, built=True)) as client:
answer = client.get("/assets/app.js")
assert answer.status_code == 200
assert "// built" in answer.text
def test_a_client_routed_path_falls_through_to_the_page(tmp_path: Path) -> None:
# A reload on a sub-view asks this server for a path only the browser
# knows about, so the catch-all has to answer with the page itself.
with client_for(deps_at(tmp_path, built=True)) as client:
assert "built" in client.get("/config").text
def test_an_unbuilt_page_names_the_command_instead_of_crashing(
tmp_path: Path,
) -> None:
# StaticFiles raises at construction on a missing directory, so mounting
# it unconditionally would take the whole app down on a fresh clone --
# and in CI, where pytest runs without ever calling pnpm build.
with client_for(deps_at(tmp_path)) as client:
answer = client.get("/")
assert answer.status_code == 503
assert assets.BUILD_COMMAND in answer.text
def test_a_half_built_directory_counts_as_not_built(tmp_path: Path) -> None:
# index.html without assets/ would raise from the mount exactly the way
# a missing dist/ does.
dist = tmp_path / "dist"
dist.mkdir()
(dist / "index.html").write_text("<title>half</title>", encoding="utf-8")
assert assets.is_built(dist) is False
def test_the_api_is_not_shadowed_by_the_catch_all(tmp_path: Path) -> None:
# /{path:path} would swallow every route registered after it, which is
# why the assets go on last.
with client_for(deps_at(tmp_path, built=True)) as client:
assert client.get("/api/status").json()["reachable"] is True

85
tests/test_manage_guard.py Normal file
View File

@ -0,0 +1,85 @@
# 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))

203
tests/test_manage_routes.py Normal file
View File

@ -0,0 +1,203 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Status, config, .env and the two actions."""
from __future__ import annotations
from typing import TYPE_CHECKING
import httpx
from signal_bot import external
from tests.manage_fixture import (
ENV_TEXT,
FIXTURE_GROUP,
answers_junk,
bot_is_down,
client_for,
deps_at,
)
if TYPE_CHECKING:
from pathlib import Path
import pytest
def test_a_healthy_bot_is_reported_with_what_it_says_about_itself(
tmp_path: Path,
) -> None:
with client_for(deps_at(tmp_path)) as client:
body = client.get("/api/status").json()
assert body["reachable"] is True
assert body["bot"]["status"] == "ok"
def test_the_bot_being_down_is_a_state_and_never_an_error(tmp_path: Path) -> None:
# The single most important thing this page can tell you, so it must not
# be the thing that stops the page rendering.
with client_for(deps_at(tmp_path, bot_is_down)) as client:
answer = client.get("/api/status")
assert answer.status_code == 200
assert answer.json() == {"reachable": False, "bot": {}}
def test_something_listening_that_is_not_the_bot_is_also_just_down(
tmp_path: Path,
) -> None:
with client_for(deps_at(tmp_path, answers_junk)) as client:
assert client.get("/api/status").json()["reachable"] is False
def test_groups_are_listed_so_an_id_is_picked_and_never_typed(
tmp_path: Path,
) -> None:
with client_for(deps_at(tmp_path)) as client:
groups = client.get("/api/groups").json()["groups"]
# The entry with no id is dropped: it cannot be picked, so offering it
# would only produce a group setting that fails at the next startup.
assert groups == [{"id": FIXTURE_GROUP, "name": "alfa"}]
def test_an_unreachable_container_lists_no_groups_rather_than_failing(
tmp_path: Path,
) -> None:
with client_for(deps_at(tmp_path, bot_is_down)) as client:
assert client.get("/api/groups").json()["groups"] == []
def test_a_group_listing_that_is_not_json_lists_no_groups(tmp_path: Path) -> None:
with client_for(deps_at(tmp_path, answers_junk)) as client:
assert client.get("/api/groups").json()["groups"] == []
def test_the_configuration_is_shown_with_the_number_cut_down(
tmp_path: Path,
) -> None:
with client_for(deps_at(tmp_path)) as client:
body = client.get("/api/config").json()
# Enough to recognise, not enough to be the number. This repository
# leaked one into a public place once and the fix was to delete it.
assert body["phone_number"] == "…0000"
assert "+1555" not in body["phone_number"]
assert body["bind"] == "127.0.0.1:8137"
assert body["report_at"] == "21:37"
def test_the_env_view_says_whether_group_settings_can_be_edited_now(
tmp_path: Path,
) -> None:
with client_for(deps_at(tmp_path)) as client:
body = client.get("/api/env").json()
assert body["groups_known"] is True
assert "GROUP_IDS" in body["needs_groups"]
assert {"key": "COOLDOWN_SECONDS", "value": "10", "editable": True} in body["keys"]
def test_group_settings_go_read_only_when_the_container_is_unreachable(
tmp_path: Path,
) -> None:
with client_for(deps_at(tmp_path, bot_is_down)) as client:
assert client.get("/api/env").json()["groups_known"] is False
def test_an_ordinary_edit_is_written_and_asks_for_a_restart(tmp_path: Path) -> None:
deps = deps_at(tmp_path)
with client_for(deps) as client:
answer = client.put("/api/env", json={"values": {"COOLDOWN_SECONDS": "30"}})
assert answer.json() == {"status": "written", "restart_required": True}
written = deps.env_path.read_text(encoding="utf-8")
assert "COOLDOWN_SECONDS=30" in written
assert "# Empty means every group this account is in." in written
def test_a_read_only_key_is_refused_by_name(tmp_path: Path) -> None:
deps = deps_at(tmp_path)
with client_for(deps) as client:
# The value never matters: a read-only key is refused before
# anything is parsed, so it cannot be smuggled past by making the
# rest of the file invalid.
answer = client.put("/api/env", json={"values": {"PHONE_NUMBER": "anything"}})
assert answer.status_code == 403
assert "PHONE_NUMBER" in answer.json()["detail"]
assert deps.env_path.read_text(encoding="utf-8") == ENV_TEXT
def test_a_group_edit_is_refused_while_no_listing_has_been_read(
tmp_path: Path,
) -> None:
deps = deps_at(tmp_path, bot_is_down)
with client_for(deps) as client:
answer = client.put("/api/env", json={"values": {"GROUP_IDS": "group.x"}})
assert answer.status_code == 403
assert deps.env_path.read_text(encoding="utf-8") == ENV_TEXT
def test_an_edit_the_bot_would_refuse_at_boot_leaves_the_file_alone(
tmp_path: Path,
) -> None:
deps = deps_at(tmp_path)
with client_for(deps) as client:
answer = client.put("/api/env", json={"values": {"REPORT_HOUR": "99"}})
assert answer.status_code == 400
assert "REPORT_HOUR" in answer.json()["detail"]
# Byte-identical: validation happens before anything is written.
assert deps.env_path.read_text(encoding="utf-8") == ENV_TEXT
def test_a_selftest_is_proxied_to_the_bot_rather_than_reimplemented(
tmp_path: Path,
) -> None:
with client_for(deps_at(tmp_path)) as client:
answer = client.post("/api/selftest", json={"group": "alfa", "text": "!kot"})
assert answer.json()["status"] == "delivered"
def test_the_bots_own_refusal_is_passed_through_unchanged(tmp_path: Path) -> None:
def refuses(request: httpx.Request) -> httpx.Response:
if request.url.path.endswith("/_selftest"):
return httpx.Response(403, json={"status": "refused", "reason": "not listed"})
return httpx.Response(200, json=[])
# SELFTEST_GROUPS stays fail-closed in exactly one place, and rewording
# its refusal here would make two sentences for one rule.
with client_for(deps_at(tmp_path, refuses)) as client:
answer = client.post("/api/selftest", json={"group": "other", "text": "!kot"})
assert answer.status_code == 403
assert answer.json()["detail"]["reason"] == "not listed"
def test_a_bot_that_does_not_answer_is_a_bad_gateway(tmp_path: Path) -> None:
with client_for(deps_at(tmp_path, bot_is_down)) as client:
answer = client.post("/api/selftest", json={"group": "alfa", "text": "!kot"})
assert answer.status_code == 502
def test_restarting_the_container_reports_what_actually_happened(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
seen: list[list[str]] = []
def fake_run(command: list[str], *, timeout: float = 0.0) -> external.Result:
seen.append(list(command))
assert timeout > 0
return external.Result(ok=True, stdout="signal-api", stderr="")
monkeypatch.setattr(external, "run", fake_run)
with client_for(deps_at(tmp_path)) as client:
answer = client.post("/api/restart-container", json={})
assert answer.json() == {"ok": True, "stdout": "signal-api", "stderr": ""}
assert seen == [["docker", "restart", "signal-api"]]
def test_a_restart_that_failed_is_not_allowed_to_look_like_success(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(
external,
"run",
lambda *_a, **_k: external.Result(ok=False, stdout="", stderr="no such container"),
)
with client_for(deps_at(tmp_path)) as client:
body = client.post("/api/restart-container", json={}).json()
assert body["ok"] is False
assert body["stderr"] == "no such container"