phase: 15_agent_mtls
All green. Report:
**Verified / implemented (final pass of `15_agent_mtls`)**
- Wrote `data/logs/verify-15-tls-e2e.py`: a real `host_agent.py --cert-bundle` + a real dashboard process over a scratch DB (`HOST_TARGETS=loop=https://…`) — **27/27 PASS**, transcript `verify-15-tls-e2e.out`, notes `data/logs/NOTES-tls-rollout.md`.
- Read-only checks against the running dev store: 5 `certs.*` rows byte-identical across a restart; served bundle == the `meta` row (1430 B); 0 events containing `-----BEGIN`; unsigned `/certs`→`303 ?next=`, unsigned `/api/certs`→`401`, signed-in `200`; screenshots `data/logs/15-final-{certs,system}.png`.
- Docs (step 6, full review): README — added *Protecting an agent, host by host* (the rollout task-07 left out), a `tests/test_certs.py`/TLS-matrix/sandbox-installer paragraph, and fixed two stale “the one write endpoint in the app” claims (`NUT_ALLOW_ACTIONS`, Battery-test buttons). AGENTS.md — test count 1424→1427, bundle-fetch command in *Commands*, a Testing-policy bullet on structurally pinning a key-holding module, fleet transport bullet (token mode is now the compatibility path), gap 11 final state.
**Tests / lint (full suite 100% green)**
- `uv run pytest` → **1427 passed, 15 skipped**, exit 0 · `bash .agents/validate.sh` → **exit 0** (coverage **98.94%** vs 94% floor, `certs.py` 100%; ruff “All checks passed”; pyright 0 errors; `node --check` clean; hygiene clean)
- `uv run pytest -m live --live` → **11 passed / 0 failed** · `./scripts/dev-server.sh status` → all 9 sources `ok`
- `uv pip list --outdated` → empty after bumping `fastapi 0.152.0 → 0.153.0` (`uv.lock` only); full suite re-run green.
**Completion criteria**
- Fresh-DB root CA, idempotent: ✅ (tests + live store hashes) · agent TLS matrix incl. wrong-CA/certless 403/`/health` CA-only/revoked 403 + live denylist re-read: ✅
- `https` `host:`/`disks:` `ok` + revoked says “agent certificate revoked” with no dial: ✅ on loopback (real agent + real dashboard) and spy-pinned hermetically; **not** on the fleet (below)
- Signed-in `/certs` + rotate/revoke/root-rotate + byte-identical bundle download: ✅ · installer bundle/unit/`--status` over TLS: ✅ (0640, see deviation)
- `pytest`/`validate.sh`/`--outdated`: ✅ · token-mode byte-identical, engine/UPS untouched, public surface unchanged (route walk): ✅
- **Live fleet TLS (deskwork+gateway flip): NOT done** — no live changes. It needs root on hosts without the installer grant (`deskwork`), has no both-ways-working window (`http://` dashboard vs https-only agent), and would deploy the agent from this not-yet-committed tree. Exact procedure recorded in README + `data/logs/NOTES-tls-rollout.md`.
**Deviations**: bundle mode `0640 root:dashboards-agent` instead of the task file’s `0600` (the agent re-reads its own bundle for the denylist; 0600 breaks the unit — already recorded in AGENTS.md); the loopback dashboard substitutes for the fleet flip. Incidental: a stray `pkill` of mine killed the dev server mid-pass; restarted and verified healthy.
**Next pending phase**: `.agents/phases/todo/16_agent_smart_pass_budget` (then `17_installer_device_filter`).
This commit is contained in:
@@ -156,6 +156,18 @@ SMART_MISSING_BINARY = ("smart: /usr/sbin/smartctl not found — install smartmo
|
||||
"give this user device read access (disk group) for SMART data")
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def host_agent_module() -> Any:
|
||||
"""The stdlib agent as a module, for the tests that run it outside `tests/test_agent.py`.
|
||||
|
||||
Loaded by path because it is one file that ships to every host and must not become a package
|
||||
(see `_load_host_agent`). The phase-15 collector tests need one module object for both the
|
||||
bundle they load and the handler they serve, so the TLS dial runs the artifact the fleet
|
||||
really runs rather than a look-alike.
|
||||
"""
|
||||
return _load_host_agent()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def smart_shape() -> Any:
|
||||
"""`smart_shape('nvme')` / `smart_shape('ata_hdd')` -> the capture, parsed by the agent.
|
||||
|
||||
+642
-7
@@ -1,26 +1,49 @@
|
||||
"""The host agent: stdlib-only, read-only by construction, and wrong on no box.
|
||||
|
||||
Three things are asserted here:
|
||||
Four things are asserted here:
|
||||
* parsing of /proc and /sys files exactly as Linux writes them (fake roots, so the
|
||||
weird cases are testable on any machine);
|
||||
* the collector produces a complete, non-crashing snapshot against the *real* /proc of
|
||||
whatever runs the tests, because that is the code that ships;
|
||||
* the HTTP surface answers GETs, needs a token when it is exposed, and has no write path.
|
||||
* the HTTP surface answers GETs, needs a token when it is exposed, and has no write path;
|
||||
* the TLS mode (phase 15) answers a real handshake only to a client certificate the bundle's
|
||||
root signed and the bundle's denylist does not name, with certificates minted by the
|
||||
dashboard's own `certs.py` — there is no second generator to disagree with.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import http.client
|
||||
import importlib.util
|
||||
import json
|
||||
import re
|
||||
import socket
|
||||
import ssl
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from http.server import ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import Any, cast
|
||||
|
||||
import pytest
|
||||
from cryptography import x509
|
||||
from cryptography.hazmat.primitives.asymmetric import ec
|
||||
from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePrivateKey
|
||||
from cryptography.hazmat.primitives.hashes import SHA256
|
||||
from cryptography.hazmat.primitives.serialization import (
|
||||
Encoding,
|
||||
NoEncryption,
|
||||
PrivateFormat,
|
||||
load_pem_private_key,
|
||||
)
|
||||
from cryptography.x509.oid import NameOID
|
||||
|
||||
from dashboards import certs
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
AGENT_PATH = ROOT / "agent" / "host_agent.py"
|
||||
@@ -414,6 +437,141 @@ def test_the_pass_budget_skips_the_rest_with_one_error(tmp_path, monkeypatch) ->
|
||||
assert any("budget" in e and "3 disk(s) skipped" in e for e in snap["errors"])
|
||||
|
||||
|
||||
class _TimeProxy:
|
||||
"""The agent's `time` module with one instrumented function: `monotonic`.
|
||||
|
||||
Everything else (`time()`, `sleep()`) stays the stdlib one, so patching the clock cannot
|
||||
leak into the rest of the snapshot or into pytest.
|
||||
"""
|
||||
|
||||
def __init__(self, clock: FakeSmartPassClock) -> None:
|
||||
self._clock = clock
|
||||
|
||||
def monotonic(self) -> float:
|
||||
return self._clock.now
|
||||
|
||||
def __getattr__(self, name: str) -> Any:
|
||||
"""Everything the agent's `time` module offers besides `monotonic`, unchanged."""
|
||||
return getattr(time, name)
|
||||
|
||||
|
||||
class FakeSmartPassClock:
|
||||
"""A clock plus a `smartctl` spawn the test drives, so a **deadline** is assertable.
|
||||
|
||||
The budget arithmetic in `_smart_pass` is invisible to a fake binary: `subprocess.run`'s
|
||||
`timeout` never reaches the child, and burning 80 real seconds to watch a 16-disk pass
|
||||
truncate is not a measurement anybody keeps. So this drives the two things the pass
|
||||
actually reads — `time.monotonic()` and `subprocess.run()` — and records the timeout each
|
||||
device was given, which is `min(deadline - now, 5 s)`: the pass's own view of how much
|
||||
budget was left when it asked. Both patches go on `host_agent`'s own module globals
|
||||
through proxies, never onto the stdlib modules themselves.
|
||||
"""
|
||||
|
||||
def __init__(self, *, seconds_per_device: float, start: float = 10_000.0) -> None:
|
||||
self.seconds_per_device = seconds_per_device
|
||||
self.now = start
|
||||
self.calls: list[tuple[str, float]] = []
|
||||
self.output = read_fixture(NVME_CAPTURE)
|
||||
|
||||
def run(self, argv: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]:
|
||||
device = str(argv[-1]).rsplit("/", 1)[-1]
|
||||
timeout = float(kwargs["timeout"])
|
||||
self.calls.append((device, timeout))
|
||||
self.now += self.seconds_per_device
|
||||
return subprocess.CompletedProcess(argv, 0, self.output, "")
|
||||
|
||||
def install(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(host_agent, "time", _TimeProxy(self))
|
||||
monkeypatch.setattr(host_agent, "subprocess", SimpleNamespace(
|
||||
run=self.run,
|
||||
TimeoutExpired=subprocess.TimeoutExpired,
|
||||
CompletedProcess=subprocess.CompletedProcess))
|
||||
|
||||
|
||||
# driveripper's real inventory: 8 NVMe + 8 spinning, which is the shape the budget scales for
|
||||
SHELF_DISKS = [f"nvme{i}n1" for i in range(8)] + [f"sd{c}" for c in "abcdefgh"]
|
||||
|
||||
|
||||
def write_disk_shelf(root: Path, names: list[str]) -> None:
|
||||
for name in names:
|
||||
write_fake_disk(root, name, model=f"SYSFS {name}")
|
||||
|
||||
|
||||
def test_the_pass_budget_scales_with_the_disk_count(tmp_path, monkeypatch) -> None:
|
||||
"""A 16-disk pass gets 5 s x 16, not the 20 s floor — because a truncated pass nulls the
|
||||
tail drives on every single poll (the bug that put this on the operator's screen on
|
||||
2026-10-19). The clock here burns 1.5 s per device, so the pass costs 24 s of budget:
|
||||
over the floor, far under the scale. With the flat budget this test truncates.
|
||||
"""
|
||||
root = tmp_path / "root"
|
||||
write_disk_shelf(root, SHELF_DISKS)
|
||||
clock = FakeSmartPassClock(seconds_per_device=1.5)
|
||||
clock.install(monkeypatch)
|
||||
collector = host_agent.Collector(str(root))
|
||||
snap = collector.snapshot()
|
||||
|
||||
asked = [name for name, _ in clock.calls]
|
||||
assert asked == snap["disk_io"]["device_names"], "every disk got its smartctl run"
|
||||
assert len(asked) == 16
|
||||
assert clock.seconds_per_device * len(asked) > host_agent.SMARTCTL_BUDGET_SECONDS, (
|
||||
"this pass costs more than the 20 s floor — that is what makes it a scaling test")
|
||||
assert [e for e in snap["errors"] if "budget" in e] == [], snap["errors"]
|
||||
assert all(device["smart"] is not None for device in snap["disk_io"]["devices"])
|
||||
# the *late* devices still hold the whole per-device bound, which is the point of scaling
|
||||
assert clock.calls[-1][1] == host_agent.SMARTCTL_TIMEOUT_SECONDS
|
||||
assert clock.calls[0][1] == host_agent.SMARTCTL_TIMEOUT_SECONDS
|
||||
|
||||
|
||||
def test_the_pass_budget_floor_is_twenty_seconds(tmp_path, monkeypatch) -> None:
|
||||
"""Scaling is `max(20 s, 5 s x disks)`, so one or two disks still get the 20 s floor and
|
||||
not a 5 s or 10 s window. The second device's timeout is the visible half: with a 20 s
|
||||
floor it is `min(20 - 18, 5)` = 2 s, where a budget of 5 s x 2 would have skipped it and a
|
||||
30 s floor would still have handed out the full 5 s.
|
||||
"""
|
||||
root = tmp_path / "root"
|
||||
write_disk_shelf(root, ["nvme0n1", "sda"])
|
||||
clock = FakeSmartPassClock(seconds_per_device=18.0)
|
||||
clock.install(monkeypatch)
|
||||
collector = host_agent.Collector(str(root))
|
||||
snap = collector.snapshot()
|
||||
|
||||
assert [name for name, _ in clock.calls] == ["nvme0n1", "sda"]
|
||||
assert clock.calls[0][1] == host_agent.SMARTCTL_TIMEOUT_SECONDS
|
||||
assert clock.calls[1][1] == pytest.approx(
|
||||
host_agent.SMARTCTL_BUDGET_SECONDS - clock.seconds_per_device)
|
||||
assert [e for e in snap["errors"] if "budget" in e] == []
|
||||
assert all(device["smart"] is not None for device in snap["disk_io"]["devices"])
|
||||
|
||||
|
||||
def test_the_pass_budget_error_line_names_the_scaled_figure(tmp_path, monkeypatch) -> None:
|
||||
"""When a scaled pass does run out, the operator's one line says *which* budget ran out:
|
||||
80 s on a 16-disk shelf, never the 20 s floor. A test that asserted only "a budget error
|
||||
appears" would pass on the pre-scaling code; this one cannot.
|
||||
"""
|
||||
root = tmp_path / "root"
|
||||
write_disk_shelf(root, SHELF_DISKS)
|
||||
clock = FakeSmartPassClock(seconds_per_device=6.0)
|
||||
clock.install(monkeypatch)
|
||||
collector = host_agent.Collector(str(root))
|
||||
snap = collector.snapshot()
|
||||
|
||||
scaled = max(host_agent.SMARTCTL_BUDGET_SECONDS,
|
||||
host_agent.SMARTCTL_TIMEOUT_SECONDS * len(SHELF_DISKS))
|
||||
assert scaled == 80.0
|
||||
# 6 s per device: devices 1-14 fit inside 80 s (78 s spent), the 15th finds the deadline
|
||||
# gone, and its sibling is skipped with it.
|
||||
assert len(clock.calls) == 14, [name for name, _ in clock.calls]
|
||||
assert clock.calls[-1][1] == pytest.approx(scaled - 6.0 * 13)
|
||||
budget_lines = [e for e in snap["errors"] if "budget" in e]
|
||||
assert len(budget_lines) == 1, snap["errors"]
|
||||
assert budget_lines[0] == (f"smart: pass budget of {scaled:.0f}s exhausted, "
|
||||
f"2 disk(s) skipped")
|
||||
assert f"{host_agent.SMARTCTL_BUDGET_SECONDS:.0f}s" not in budget_lines[0]
|
||||
by_name = {device["name"]: device for device in snap["disk_io"]["devices"]}
|
||||
assert [by_name[name]["smart"] is None for name in SHELF_DISKS[:14]] == [False] * 14
|
||||
assert [by_name[name]["smart"] is None for name in SHELF_DISKS[14:]] == [True, True]
|
||||
|
||||
|
||||
def test_unexecutable_smartctl_degrades_like_a_missing_one(tmp_path, monkeypatch) -> None:
|
||||
root = tmp_path / "root"
|
||||
write_fake_disk(root, "nvme0n1")
|
||||
@@ -606,6 +764,9 @@ def agent_server():
|
||||
def start(token: str | None = None) -> tuple[str, int]:
|
||||
host_agent.Handler.collector = host_agent.Collector("/")
|
||||
host_agent.Handler.token = token
|
||||
# Cleared on purpose: `Handler` is process-wide state, so a token-mode server started
|
||||
# after a TLS test in this file must not inherit that test's bundle.
|
||||
host_agent.Handler.bundle = None
|
||||
httpd = ThreadingHTTPServer(("127.0.0.1", 0), host_agent.Handler)
|
||||
servers.append(httpd)
|
||||
threading.Thread(target=httpd.serve_forever, daemon=True).start()
|
||||
@@ -687,8 +848,10 @@ def test_agent_has_no_write_path(agent_server) -> None:
|
||||
assert excinfo.value.code in (405, 501)
|
||||
|
||||
|
||||
def test_refuses_to_bind_off_loopback_without_a_token() -> None:
|
||||
def test_refuses_to_bind_off_loopback_without_a_token(monkeypatch) -> None:
|
||||
"""0.0.0.0 with no token would hand host inventory to the whole LAN."""
|
||||
monkeypatch.delenv("AGENT_TOKEN", raising=False)
|
||||
monkeypatch.delenv("AGENT_CERT_BUNDLE", raising=False)
|
||||
assert host_agent.main(["--host", "0.0.0.0", "--port", "0"]) == 2
|
||||
|
||||
|
||||
@@ -712,9 +875,10 @@ def test_agent_is_dependency_free() -> None:
|
||||
top = {name.split(".")[0] for name in imports}
|
||||
third_party = top - set(sys.stdlib_module_names) - {"host_agent"}
|
||||
assert not third_party, f"agent grew a non-stdlib import: {third_party}"
|
||||
assert top <= {"__future__", "argparse", "collections", "dataclasses", "http", "json",
|
||||
"os", "pathlib", "re", "socket", "subprocess", "threading", "time",
|
||||
"typing"}, f"agent import footprint grew: {sorted(top)}"
|
||||
assert top <= {"__future__", "argparse", "collections", "contextlib", "dataclasses", "http",
|
||||
"json", "os", "pathlib", "re", "shutil", "socket", "ssl", "subprocess",
|
||||
"tempfile", "threading", "time", "typing"}, (
|
||||
f"agent import footprint grew: {sorted(top)}")
|
||||
|
||||
|
||||
# ---- how the token is supplied ------------------------------------------- #
|
||||
@@ -746,6 +910,477 @@ def test_no_token_at_all_is_allowed_on_loopback(tmp_path) -> None:
|
||||
assert (token, error) == (None, None)
|
||||
|
||||
|
||||
def test_main_refuses_to_bind_off_loopback_without_a_token(tmp_path) -> None:
|
||||
def test_main_refuses_to_bind_off_loopback_without_a_token(tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.delenv("AGENT_TOKEN", raising=False)
|
||||
monkeypatch.delenv("AGENT_CERT_BUNDLE", raising=False)
|
||||
code = host_agent.main(["--host", "0.0.0.0", "--token-file", str(tmp_path / "missing")])
|
||||
assert code == 2, "the agent must not start exposed and unauthenticated"
|
||||
|
||||
|
||||
# ---- TLS mode: one installed bundle, and its client certificate is the auth --- #
|
||||
#
|
||||
# Everything below runs a real `ssl` handshake against the real handler, because the claims
|
||||
# here are handshake claims: "optional at the handshake, enforced by the handler" is exactly
|
||||
# the kind of thing that reads fine and then makes `/health` unprobeable. The certificates come
|
||||
# from `dashboards.certs` over a dict store, so the agent is tested against the artifact the
|
||||
# dashboard actually signs rather than against a second generator that could drift from it.
|
||||
|
||||
|
||||
#: The agent module is loaded by `importlib` (it is one file with no importable package), so
|
||||
#: pyright sees `host_agent` as an untyped module and rejects `host_agent.TlsBundle` in a type
|
||||
#: position. This alias is the honest spelling: the object is that class, and the tests below
|
||||
#: assert `isinstance` where the claim actually matters.
|
||||
Bundle = Any
|
||||
|
||||
|
||||
class DictMeta:
|
||||
"""The three-method `certs.MetaStore` slice, backed by a dict (the `test_certs.py` idiom)."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.rows: dict[str, str] = {}
|
||||
|
||||
def get_meta(self, key: str) -> str | None:
|
||||
return self.rows.get(key)
|
||||
|
||||
def set_meta(self, key: str, value: str) -> None:
|
||||
self.rows[key] = value
|
||||
|
||||
def delete_meta(self, key: str) -> int:
|
||||
return 1 if self.rows.pop(key, None) is not None else 0
|
||||
|
||||
|
||||
def sign_client_leaf(state: certs.CertsState, cn: str, where: Path) -> SimpleNamespace:
|
||||
"""A second client identity under the dashboard's own root, so it verifies and can be revoked.
|
||||
|
||||
`certs.py` mints exactly one client certificate on purpose (the dashboard has one identity),
|
||||
so the revoked-peer and wrong-authority cases need a leaf signed by hand. The serial comes
|
||||
back because the denylist names a serial, not a name.
|
||||
"""
|
||||
root_key = cast(EllipticCurvePrivateKey,
|
||||
load_pem_private_key(state.root_key.encode("utf-8"), password=None))
|
||||
root_cert = x509.load_pem_x509_certificate(state.root_cert.encode("utf-8"))
|
||||
key = ec.generate_private_key(ec.SECP256R1())
|
||||
moment = datetime.now(UTC)
|
||||
cert = (
|
||||
x509.CertificateBuilder()
|
||||
.subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)]))
|
||||
.issuer_name(root_cert.subject)
|
||||
.public_key(key.public_key())
|
||||
.serial_number(x509.random_serial_number())
|
||||
.not_valid_before(moment - timedelta(days=1))
|
||||
.not_valid_after(moment + timedelta(days=365))
|
||||
.add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
|
||||
.sign(root_key, SHA256())
|
||||
)
|
||||
cert_path = where / f"{cn}-cert.pem"
|
||||
key_path = where / f"{cn}-key.pem"
|
||||
cert_path.write_bytes(cert.public_bytes(Encoding.PEM))
|
||||
key_path.write_bytes(key.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption()))
|
||||
return SimpleNamespace(cert=cert_path, key=key_path, serial=cert.serial_number)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def tls_fleet(tmp_path_factory: pytest.TempPathFactory) -> SimpleNamespace:
|
||||
"""A dashboard CA, its client pair, one issued agent bundle, and one foreign authority."""
|
||||
base = tmp_path_factory.mktemp("agent-tls")
|
||||
store = DictMeta()
|
||||
state = certs.ensure_certs(store)
|
||||
state = certs.issue_agent_cert(store, state, "localhost", ["127.0.0.1", "localhost"])
|
||||
entry = state.agent("localhost")
|
||||
assert entry is not None
|
||||
for name in ("dashboard", "foreign"):
|
||||
(base / name).mkdir()
|
||||
dashboard = certs.write_material(state, base / "dashboard")
|
||||
foreign = certs.write_material(certs.ensure_certs(DictMeta()), base / "foreign")
|
||||
return SimpleNamespace(
|
||||
root_cert=state.root_cert,
|
||||
bundle=entry.bundle,
|
||||
bundle_rev=entry.bundle_rev,
|
||||
client_cert=str(dashboard.client_cert),
|
||||
client_key=str(dashboard.client_key),
|
||||
foreign_cert=str(foreign.client_cert),
|
||||
foreign_key=str(foreign.client_key),
|
||||
stolen=sign_client_leaf(state, "stolen", base),
|
||||
)
|
||||
|
||||
|
||||
def write_bundle(tmp_path: Path, text: str) -> Path:
|
||||
"""An installed bundle: one PEM file, exactly as `install-agent.sh` will place it."""
|
||||
path = tmp_path / "dashboards-agent.certs"
|
||||
path.write_text(text, encoding="ascii")
|
||||
return path
|
||||
|
||||
|
||||
def load_bundle(tmp_path: Path, text: str) -> Bundle:
|
||||
path = write_bundle(tmp_path, text)
|
||||
bundle, error = host_agent.TlsBundle.load(str(path))
|
||||
assert error is None and bundle is not None
|
||||
return bundle
|
||||
|
||||
|
||||
def client_context(ca: str, cert: str | None = None, key: str | None = None) -> ssl.SSLContext:
|
||||
"""A client that trusts `ca`, and presents a client certificate only when given one."""
|
||||
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
||||
context.load_verify_locations(cadata=ca)
|
||||
if cert and key:
|
||||
context.load_cert_chain(cert, key)
|
||||
return context
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tls_agent_server():
|
||||
"""The real handler on a TLS-wrapped loopback port; yields the base URL."""
|
||||
servers: list[ThreadingHTTPServer] = []
|
||||
|
||||
def start(bundle: Bundle, token: str | None = None) -> str:
|
||||
host_agent.Handler.collector = host_agent.Collector("/", smart=False)
|
||||
host_agent.Handler.token = token
|
||||
host_agent.Handler.bundle = bundle
|
||||
httpd = ThreadingHTTPServer(("127.0.0.1", 0), host_agent.Handler)
|
||||
httpd.socket = bundle.context.wrap_socket(httpd.socket, server_side=True)
|
||||
servers.append(httpd)
|
||||
threading.Thread(target=httpd.serve_forever, daemon=True).start()
|
||||
return f"https://127.0.0.1:{httpd.server_address[1]}"
|
||||
|
||||
yield start
|
||||
for httpd in servers:
|
||||
httpd.shutdown()
|
||||
httpd.server_close()
|
||||
host_agent.Handler.bundle = None
|
||||
|
||||
|
||||
def _json_or_raw(body: bytes) -> dict:
|
||||
"""JSON when the handler answered with JSON; the raw text for the status page."""
|
||||
try:
|
||||
return json.loads(body)
|
||||
except json.JSONDecodeError:
|
||||
return {"raw": body.decode(errors="replace")}
|
||||
|
||||
|
||||
def _https_get(url: str, context: ssl.SSLContext) -> tuple[int, dict]:
|
||||
request = urllib.request.Request(url)
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=15, context=context) as response:
|
||||
return response.status, _json_or_raw(response.read())
|
||||
except urllib.error.HTTPError as error:
|
||||
return error.code, _json_or_raw(error.read())
|
||||
|
||||
|
||||
def test_the_agent_and_the_dashboard_agree_on_the_bundle_lines() -> None:
|
||||
"""The agent parses a PEM by text with stdlib only, so the two literals are the contract."""
|
||||
assert host_agent.BUNDLE_REV_LINE == certs.BUNDLE_REV_LINE
|
||||
assert host_agent.REVOKED_LINE == certs.REVOKED_LINE
|
||||
|
||||
|
||||
def test_the_bundle_lines_are_read_as_written() -> None:
|
||||
text = "# BUNDLE-REV 12\n# REVOKED 5\n# REVOKED 77\n"
|
||||
assert host_agent.bundle_revision(text) == 12
|
||||
assert host_agent.bundle_revoked(text) == frozenset({"5", "77"})
|
||||
assert host_agent.bundle_revision("nothing at all\n") is None
|
||||
assert host_agent.bundle_revision("# BUNDLE-REV not-a-number\n") is None
|
||||
assert host_agent.bundle_revoked("# REVOKED\n") == frozenset()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("revoked, peer_hex, expected", [
|
||||
# the shape a real `x509.random_serial_number()` serial arrives in, both spellings
|
||||
("57989923439447977599989997426575336710275559875",
|
||||
"A285BAC84EB9C5FC282EF7CDB8D962F38BF71C3", True),
|
||||
("12648430", "C0FFEE", True),
|
||||
("14", "0E", True),
|
||||
("0E", "0E", True),
|
||||
("15", "0E", False),
|
||||
("0", "0E", False),
|
||||
])
|
||||
def test_the_denylist_matches_a_serial_whichever_way_it_is_written(revoked: str,
|
||||
peer_hex: str,
|
||||
expected: bool) -> None:
|
||||
"""`getpeercert()` says hex, `certs.py` writes decimal, and the agent must meet in the middle.
|
||||
|
||||
An agent comparing one spelling to the other accepts a revoked client forever: the operator
|
||||
clicked revoke, `/certs` says revoked, and the fleet keeps trusting the stolen identity.
|
||||
"""
|
||||
cert = {"serialNumber": peer_hex} if peer_hex else {}
|
||||
assert host_agent.serial_is_revoked(cert, frozenset({revoked})) is expected
|
||||
|
||||
|
||||
def test_a_certificate_with_no_serial_is_never_treated_as_revoked() -> None:
|
||||
assert host_agent.serial_is_revoked({}, frozenset({"1", "0"})) is False
|
||||
|
||||
|
||||
def test_resolve_bundle_reports_the_revision_and_the_denylist_of_a_real_bundle(
|
||||
tls_fleet: SimpleNamespace, tmp_path: Path) -> None:
|
||||
text = tls_fleet.bundle + f"{host_agent.REVOKED_LINE}{tls_fleet.stolen.serial}\n"
|
||||
context, revision, revoked, error = host_agent.resolve_bundle(str(write_bundle(tmp_path, text)))
|
||||
assert error is None
|
||||
assert context is not None and isinstance(context, ssl.SSLContext)
|
||||
assert revision == tls_fleet.bundle_rev
|
||||
assert revoked == frozenset({str(tls_fleet.stolen.serial)})
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shape", ["missing", "empty", "no-key", "no-revision"])
|
||||
def test_a_bundle_that_cannot_be_loaded_is_an_error_naming_the_file(
|
||||
tls_fleet: SimpleNamespace, tmp_path: Path, shape: str) -> None:
|
||||
"""Every failure is a message the operator can act on, and none of them is "no TLS today"."""
|
||||
if shape == "missing":
|
||||
path = tmp_path / "absent.certs"
|
||||
elif shape == "empty":
|
||||
path = write_bundle(tmp_path, " \n")
|
||||
elif shape == "no-key":
|
||||
body = [block for block in re.findall(
|
||||
r"-----BEGIN [A-Z0-9 ]+-----.*?-----END [A-Z0-9 ]+-----\n", tls_fleet.bundle, re.S)
|
||||
if block.startswith("-----BEGIN CERTIFICATE")]
|
||||
path = write_bundle(tmp_path, "# BUNDLE-REV 1\n" + "".join(body))
|
||||
else:
|
||||
path = write_bundle(tmp_path, tls_fleet.bundle.replace("# BUNDLE-REV 1\n", ""))
|
||||
context, revision, revoked, error = host_agent.resolve_bundle(str(path))
|
||||
assert context is None and revision == 0 and revoked == frozenset()
|
||||
assert error is not None and ("absent" in error if shape == "missing" else str(path) in error)
|
||||
if shape == "no-revision":
|
||||
assert "BUNDLE-REV" in error
|
||||
|
||||
|
||||
def test_the_unpacked_pems_are_gone_once_the_context_has_them(
|
||||
tls_fleet: SimpleNamespace, tmp_path: Path, monkeypatch) -> None:
|
||||
"""The unit's RuntimeDirectory is used when it exists, and holds nothing afterwards.
|
||||
|
||||
`ssl` loads a chain from paths, so the key is written; leaving it there means the private key
|
||||
outlives the process on a disk an operator did not choose, and `kill -9` has no cleanup path.
|
||||
"""
|
||||
runtime = tmp_path / "run"
|
||||
runtime.mkdir()
|
||||
monkeypatch.setattr(host_agent, "BUNDLE_RUNTIME_DIR", runtime)
|
||||
context, _revision, _revoked, error = host_agent.resolve_bundle(str(write_bundle(tmp_path,
|
||||
tls_fleet.bundle)))
|
||||
assert error is None and context is not None
|
||||
assert list(runtime.iterdir()) == [], "the private key must not stay on disk once loaded"
|
||||
|
||||
|
||||
def test_a_missing_runtime_directory_falls_back_to_a_temporary_one(
|
||||
tls_fleet: SimpleNamespace, tmp_path: Path, monkeypatch) -> None:
|
||||
monkeypatch.setattr(host_agent, "BUNDLE_RUNTIME_DIR", tmp_path / "not-mounted")
|
||||
context, revision, _revoked, error = host_agent.resolve_bundle(str(write_bundle(tmp_path,
|
||||
tls_fleet.bundle)))
|
||||
assert error is None and context is not None
|
||||
assert revision == tls_fleet.bundle_rev
|
||||
|
||||
|
||||
def test_a_client_certificate_signed_by_the_root_reads_the_snapshot(
|
||||
tls_fleet: SimpleNamespace, tls_agent_server, tmp_path: Path) -> None:
|
||||
"""The whole point of the phase, plus the two fields task 03 reads.
|
||||
|
||||
`bundle_rev` is how `/certs` tells "this host runs an old bundle" from "this host is current",
|
||||
so it is the number the bundle carried, not a value the agent made up.
|
||||
"""
|
||||
base = tls_agent_server(load_bundle(tmp_path, tls_fleet.bundle))
|
||||
status, body = _https_get(base + "/snapshot",
|
||||
client_context(tls_fleet.root_cert, tls_fleet.client_cert,
|
||||
tls_fleet.client_key))
|
||||
assert status == 200
|
||||
assert body["ok"] is True and body["host"]["hostname"]
|
||||
assert body["tls"] is True
|
||||
assert body["bundle_rev"] == tls_fleet.bundle_rev
|
||||
for path in ("/info", "/schema", "/"):
|
||||
status, _body = _https_get(base + path, client_context(tls_fleet.root_cert,
|
||||
tls_fleet.client_cert,
|
||||
tls_fleet.client_key))
|
||||
assert status == 200, path
|
||||
|
||||
|
||||
def test_health_is_probeable_with_the_root_ca_alone_while_every_other_path_is_not(
|
||||
tls_fleet: SimpleNamespace, tls_agent_server, tmp_path: Path) -> None:
|
||||
"""The `CERT_OPTIONAL` assumption: a probe with no client certificate must still work.
|
||||
|
||||
`install-agent.sh --status` and the systemd probe hold the CA and were never handed a client
|
||||
certificate; `CERT_REQUIRED` would make every TLS install look dead. Everything that carries
|
||||
metrics is 403 — including the status page, which repeats hostname, kernel and CPU model.
|
||||
"""
|
||||
base = tls_agent_server(load_bundle(tmp_path, tls_fleet.bundle))
|
||||
probe = client_context(tls_fleet.root_cert)
|
||||
status, body = _https_get(base + "/health", probe)
|
||||
assert status == 200 and body["ok"] is True
|
||||
for path in ("/snapshot", "/info", "/schema", "/", "/nope"):
|
||||
refused, detail = _https_get(base + path, probe)
|
||||
assert refused == 403, path
|
||||
assert detail["detail"] == "client certificate required", path
|
||||
|
||||
|
||||
def test_a_client_signed_by_another_authority_is_refused(
|
||||
tls_fleet: SimpleNamespace, tls_agent_server, tmp_path: Path) -> None:
|
||||
"""A certificate that does not chain to the bundle's root never reaches a handler.
|
||||
|
||||
The server verifies at the handshake, so the answer is an aborted connection rather than an
|
||||
HTTP code — the assertion is only ever `not 200` because which alert arrives is the peer
|
||||
OpenSSL's choice, not ours.
|
||||
"""
|
||||
base = tls_agent_server(load_bundle(tmp_path, tls_fleet.bundle))
|
||||
stranger = client_context(tls_fleet.root_cert, tls_fleet.foreign_cert, tls_fleet.foreign_key)
|
||||
with pytest.raises((OSError, http.client.HTTPException)) as caught:
|
||||
_https_get(base + "/snapshot", stranger)
|
||||
assert not isinstance(caught.value, urllib.error.HTTPError)
|
||||
|
||||
|
||||
def test_a_plaintext_request_to_a_tls_agent_gets_nothing(
|
||||
tls_fleet: SimpleNamespace, tls_agent_server, tmp_path: Path) -> None:
|
||||
"""A port probe cannot read the answer, and does not get a helpful error either."""
|
||||
base = tls_agent_server(load_bundle(tmp_path, tls_fleet.bundle))
|
||||
port = int(base.rsplit(":", 1)[-1])
|
||||
with socket.create_connection(("127.0.0.1", port), timeout=15) as raw:
|
||||
raw.sendall(b"GET /snapshot HTTP/1.1\r\nHost: x\r\n\r\n")
|
||||
raw.settimeout(5)
|
||||
try:
|
||||
payload = raw.recv(200)
|
||||
except OSError:
|
||||
payload = b""
|
||||
assert b"hostname" not in payload
|
||||
|
||||
|
||||
def test_a_revoked_client_certificate_is_refused_and_the_bundle_is_re_read(
|
||||
tls_fleet: SimpleNamespace, tls_agent_server, tmp_path: Path) -> None:
|
||||
"""The two halves: the denylist bites, and re-deploying the bundle takes effect no later
|
||||
than the next request — no restart, which is the only reason the lines live in the file."""
|
||||
path = write_bundle(tmp_path, tls_fleet.bundle
|
||||
+ f"{host_agent.REVOKED_LINE}{tls_fleet.stolen.serial}\n")
|
||||
bundle, error = host_agent.TlsBundle.load(str(path))
|
||||
assert error is None and bundle is not None
|
||||
base = tls_agent_server(bundle)
|
||||
stolen = client_context(tls_fleet.root_cert, str(tls_fleet.stolen.cert), str(tls_fleet.stolen.key))
|
||||
|
||||
status, body = _https_get(base + "/snapshot", stolen)
|
||||
assert status == 403
|
||||
assert body["detail"] == "client certificate revoked"
|
||||
# A withdrawn identity is refused even on the open path: the exemption is for a probe that
|
||||
# holds the CA and no certificate, not for a certificate the operator has just taken away.
|
||||
assert _https_get(base + "/health", stolen)[0] == 403
|
||||
# The dashboard's own identity is untouched — the list names a serial, not a host.
|
||||
assert _https_get(base + "/snapshot", client_context(tls_fleet.root_cert, tls_fleet.client_cert,
|
||||
tls_fleet.client_key))[0] == 200
|
||||
|
||||
path.write_text(tls_fleet.bundle, encoding="ascii") # a redeploy: the line is gone
|
||||
status, body = _https_get(base + "/snapshot", stolen)
|
||||
assert status == 200, "the denylist is re-read when the file changes, without a restart"
|
||||
|
||||
|
||||
def test_a_renamed_denylist_entry_is_re_read_on_the_next_request(
|
||||
tls_fleet: SimpleNamespace, tmp_path: Path) -> None:
|
||||
"""An unreadable bundle keeps the list this agent already had rather than forget it."""
|
||||
path = write_bundle(tmp_path, tls_fleet.bundle
|
||||
+ f"{host_agent.REVOKED_LINE}{tls_fleet.stolen.serial}\n")
|
||||
bundle, error = host_agent.TlsBundle.load(str(path))
|
||||
assert error is None and bundle is not None
|
||||
assert bundle.revoked_serials() == frozenset({str(tls_fleet.stolen.serial)})
|
||||
path.unlink()
|
||||
assert bundle.revoked_serials() == frozenset({str(tls_fleet.stolen.serial)})
|
||||
|
||||
|
||||
def test_the_token_is_not_read_on_the_tls_path(
|
||||
tls_fleet: SimpleNamespace, tls_agent_server, tmp_path: Path) -> None:
|
||||
"""Bundle plus token: the handshake is the auth, so a stolen token cannot outlive a revoke.
|
||||
|
||||
Checking the bearer on the TLS path too would not be defence in depth — it would be a second
|
||||
credential that keeps working after the operator revoked the certificate, which is the one
|
||||
thing the denylist exists to prevent.
|
||||
"""
|
||||
base = tls_agent_server(load_bundle(tmp_path, tls_fleet.bundle), token="s3cret")
|
||||
status, _body = _https_get(base + "/snapshot",
|
||||
client_context(tls_fleet.root_cert, tls_fleet.client_cert,
|
||||
tls_fleet.client_key))
|
||||
assert status == 200
|
||||
|
||||
|
||||
def test_token_mode_reports_no_tls_fields(agent_server) -> None:
|
||||
"""Token mode is byte-identical: no `tls`, no `bundle_rev`, no new field that an agent
|
||||
installed before phase 15 does not have."""
|
||||
assert host_agent.Handler.bundle is None, "the token-mode fixture must not inherit a bundle"
|
||||
base, _port = agent_server(token="s3cret")
|
||||
_status, body = _get(base + "/snapshot", {"Authorization": "Bearer s3cret"})
|
||||
assert body["ok"] is True
|
||||
assert "tls" not in body and "bundle_rev" not in body
|
||||
|
||||
|
||||
def test_serve_wraps_the_socket_and_says_tls_and_the_revision(
|
||||
tls_fleet: SimpleNamespace, tmp_path: Path, capsys, monkeypatch) -> None:
|
||||
"""`serve()` is the function the unit runs, so its start line is what an operator reads.
|
||||
|
||||
The server is recorded rather than guessed at a port: the test needs the ephemeral port the
|
||||
OS chose *and* a way to stop `serve_forever`, which is the same object.
|
||||
"""
|
||||
created: list[ThreadingHTTPServer] = []
|
||||
|
||||
class Recording(ThreadingHTTPServer):
|
||||
def __init__(self, *args: object, **kwargs: object) -> None:
|
||||
super().__init__(*args, **kwargs) # type: ignore[arg-type]
|
||||
created.append(self)
|
||||
|
||||
monkeypatch.setattr(host_agent, "ThreadingHTTPServer", Recording)
|
||||
bundle = load_bundle(tmp_path, tls_fleet.bundle)
|
||||
thread = threading.Thread(target=host_agent.serve,
|
||||
args=("127.0.0.1", 0, host_agent.Collector("/", smart=False),
|
||||
None, bundle),
|
||||
daemon=True)
|
||||
thread.start()
|
||||
deadline = time.monotonic() + 15
|
||||
while not created and time.monotonic() < deadline:
|
||||
time.sleep(0.01)
|
||||
assert created, "serve() never built its server"
|
||||
httpd = created[0]
|
||||
port = httpd.server_address[1]
|
||||
status, _body = _https_get(f"https://127.0.0.1:{port}/health",
|
||||
client_context(tls_fleet.root_cert))
|
||||
assert status == 200
|
||||
httpd.shutdown()
|
||||
thread.join(timeout=15)
|
||||
printed = capsys.readouterr().out
|
||||
assert "tls (client cert required, rev " in printed
|
||||
# the scheme is the whole claim here (the port in the line is the requested one, which is 0
|
||||
# when the OS picks, exactly as in token mode)
|
||||
assert "[agent] listening on https://127.0.0.1" in printed
|
||||
assert "token" not in printed, "a TLS agent does not read a token"
|
||||
|
||||
|
||||
def test_a_bundle_allows_the_off_loopback_bind_without_a_token(
|
||||
tls_fleet: SimpleNamespace, tmp_path: Path, monkeypatch) -> None:
|
||||
"""The client certificate *is* the protection, so the exit-2 refusal does not apply."""
|
||||
served: list[tuple] = []
|
||||
monkeypatch.setattr(host_agent, "serve",
|
||||
lambda *args: served.append(args))
|
||||
path = write_bundle(tmp_path, tls_fleet.bundle)
|
||||
assert host_agent.main(["--host", "0.0.0.0", "--port", "0", "--cert-bundle", str(path)]) == 0
|
||||
assert served, "a bundle-configured agent must start"
|
||||
host_arg, _port, _collector, token_arg, bundle_arg = served[0]
|
||||
assert (host_arg, token_arg) == ("0.0.0.0", None)
|
||||
assert isinstance(bundle_arg, host_agent.TlsBundle)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("extra", [[], ["--cert-bundle", ""]])
|
||||
def test_off_loopback_without_a_token_or_a_bundle_is_refused_and_names_both(
|
||||
tmp_path: Path, capsys, monkeypatch, extra: list[str]) -> None:
|
||||
monkeypatch.delenv("AGENT_TOKEN", raising=False)
|
||||
monkeypatch.delenv("AGENT_CERT_BUNDLE", raising=False)
|
||||
args = ["--host", "0.0.0.0", "--port", "0", *extra]
|
||||
assert host_agent.main(args) == 2
|
||||
printed = capsys.readouterr().out
|
||||
assert "--token" in printed and "--cert-bundle" in printed
|
||||
|
||||
|
||||
def test_an_unreadable_bundle_is_fatal_at_startup(tmp_path: Path, capsys) -> None:
|
||||
"""Exit 2 naming the file: a degraded-to-plain-HTTP agent is the answer that is worse."""
|
||||
missing = tmp_path / "absent.certs"
|
||||
assert host_agent.main(["--host", "127.0.0.1", "--port", "0",
|
||||
"--cert-bundle", str(missing)]) == 2
|
||||
assert str(missing) in capsys.readouterr().out
|
||||
|
||||
|
||||
def schema_documents(field: str) -> bool:
|
||||
"""Is this field in the SCHEMA block? `GET /schema` answers with that text, self-documenting
|
||||
an install for whoever has to debug it, so a field the payload carries belongs there."""
|
||||
doc = host_agent.__doc__ or ""
|
||||
schema = doc.split("SCHEMA (GET /snapshot)", 1)[-1].split("Deployment:", 1)[0]
|
||||
return re.search(rf"^\s+{field}\s", schema, re.M) is not None
|
||||
|
||||
|
||||
def test_the_agent_version_and_schema_document_the_new_fields() -> None:
|
||||
"""1.1.0 because the payload grew, and the docstring is what `GET /schema` answers."""
|
||||
assert host_agent.AGENT_VERSION == "1.1.0"
|
||||
assert schema_documents("tls") and schema_documents("bundle_rev")
|
||||
assert "--cert-bundle" in (host_agent.__doc__ or "")
|
||||
|
||||
|
||||
+732
-21
@@ -7,6 +7,8 @@ asset the network can fail to deliver.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import ipaddress
|
||||
import json
|
||||
import re
|
||||
import time
|
||||
from collections.abc import Iterator
|
||||
@@ -16,13 +18,15 @@ from typing import Any
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from cryptography import x509
|
||||
from cryptography.x509.oid import NameOID
|
||||
from fastapi import HTTPException
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
import store_perf
|
||||
from auth_client import sign_in
|
||||
from auth_contract import ADMIN_PASSWORD, FAST_PBKDF2
|
||||
from dashboards import alerts, auth
|
||||
from dashboards import alerts, auth, certs
|
||||
from dashboards.api import MIN_PASSWORD_LENGTH, RANGE_SECONDS, parse_range
|
||||
from dashboards.app import create_app
|
||||
from dashboards.collectors import UpstreamError
|
||||
@@ -61,7 +65,11 @@ PAGES = ("/", "/system", "/disks", "/llm", "/ups")
|
||||
#: tab, a fixed tab order — which this page deliberately does not have. `MARKUP_PAGES` is the
|
||||
#: list for the checks that are about the bytes rather than the nav.
|
||||
SETTINGS_PAGE = "/settings"
|
||||
MARKUP_PAGES = (*PAGES, SETTINGS_PAGE)
|
||||
#: `/certs` is the second admin page (phase 15 task 05), kept out of `PAGES` for exactly the same
|
||||
#: reason: it is not a tab. It is in `MARKUP_PAGES` and `PAGE_SCRIPTS`, so the CDN, `.sev-`,
|
||||
#: flavour-text and orphan-`id` audits cover it like the five dashboards.
|
||||
CERTS_PAGE = "/certs"
|
||||
MARKUP_PAGES = (*PAGES, SETTINGS_PAGE, CERTS_PAGE)
|
||||
|
||||
|
||||
@contextmanager
|
||||
@@ -105,6 +113,25 @@ def ups_client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[Test
|
||||
yield c
|
||||
|
||||
|
||||
def wait_for_source_error(client: TestClient, key: str, fragment: str,
|
||||
*, timeout: float = 15.0) -> str:
|
||||
"""Wait for one poll loop to record an error saying `fragment`, and hand back the sentence.
|
||||
|
||||
The poll loops run in the app's event loop, so an assertion about what a source said has to
|
||||
wait for a poll rather than read the state at import time. The failure names the last error
|
||||
seen, because "it never said that" and "it said something else entirely" are different bugs.
|
||||
"""
|
||||
deadline = time.time() + timeout
|
||||
error = ""
|
||||
while time.time() < deadline:
|
||||
rows = {row["key"]: row for row in client.get("/api/health").json()["sources"]}
|
||||
error = str((rows.get(key) or {}).get("error") or "")
|
||||
if fragment in error:
|
||||
return error
|
||||
time.sleep(0.05)
|
||||
pytest.fail(f"{key} never reported {fragment!r} (last error: {error!r})")
|
||||
|
||||
|
||||
def wait_for_payload(client: TestClient, path: str, *, timeout: float = 5.0) -> dict[str, Any]:
|
||||
deadline = time.time() + timeout
|
||||
while time.time() < deadline:
|
||||
@@ -721,6 +748,49 @@ def test_a_dead_host_never_writes_its_token_into_the_feed(tmp_path: Path) -> Non
|
||||
store.close()
|
||||
|
||||
|
||||
def test_a_revoked_agent_fails_both_of_its_sources_without_a_dial(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""The chain end to end, wiring included: revoke it, and the next poll stops talking to it.
|
||||
|
||||
`collectors.agent_tls_map` is what `build_hub` hands the jobs, so this is the proof that a
|
||||
revocation reaches a *running* app through its own `meta` table — no restart, and no second
|
||||
process telling the first. The spy replaces `httpx.AsyncClient.get` on the class, so it
|
||||
catches the shared client and the per-dial TLS client alike: what must not happen is any
|
||||
request at all, because a dial that then gets refused still put a socket on a host the
|
||||
operator struck out.
|
||||
"""
|
||||
db = str(tmp_path / "revoked-agent.sqlite3")
|
||||
store = SampleStore(db)
|
||||
settings = Settings(db_path=db, admin_password=ADMIN_PASSWORD,
|
||||
pbkdf2_iterations=FAST_PBKDF2,
|
||||
host_interval=0.05, disks_interval=0.05,
|
||||
host_targets=(Target(name="deskwork", url="https://127.0.0.1:1"),))
|
||||
with TestClient(create_app(settings, store)) as client:
|
||||
sign_in(client)
|
||||
# Before the revoke the same dead host fails as an unreachable upstream. That sentence is
|
||||
# the control: it is what a dial produces, so the one after it has to mean something else.
|
||||
wait_for_source_error(client, "host:deskwork", "unreachable")
|
||||
state = certs.issue_agent_cert(store, certs.ensure_certs(store), "deskwork",
|
||||
["127.0.0.1"])
|
||||
certs.revoke(store, state, "deskwork")
|
||||
|
||||
calls: list[str] = []
|
||||
|
||||
async def spy_get(self: Any, url: Any, **kwargs: Any) -> Any:
|
||||
calls.append(str(url))
|
||||
raise AssertionError(f"a revoked agent was dialled at {url}")
|
||||
|
||||
monkeypatch.setattr(httpx.AsyncClient, "get", spy_get)
|
||||
for key in ("host:deskwork", "disks:deskwork"):
|
||||
error = wait_for_source_error(client, key, "agent certificate revoked")
|
||||
assert "unreachable" not in error, "the refusal replaced the dial's sentence"
|
||||
assert calls == [], "the gate is in front of the network, not behind it"
|
||||
# and the overview says the operator revoked it rather than that the box went down
|
||||
messages = [row["message"] for row in client.get("/api/summary").json()["alerts"]]
|
||||
assert any("certificate revoked" in text for text in messages), messages
|
||||
store.close()
|
||||
|
||||
|
||||
def test_engine_urls_are_redacted_the_same_way_hosts_are(tmp_path: Path) -> None:
|
||||
"""`GUFO_BASE_URL`/`VLLM_BASE_URL` go to the browser verbatim in `/api/config`, and unlike
|
||||
`HOST_TARGETS` nothing parses a userinfo token out of them into a header — so a URL that
|
||||
@@ -926,25 +996,176 @@ def test_the_settings_page_refuses_a_caller_with_no_session(tmp_path: Path) -> N
|
||||
store.close()
|
||||
|
||||
|
||||
def test_every_dashboard_page_links_to_settings(client) -> None:
|
||||
"""The `⚙ Account` link is the only way to the session page, so it is on every dashboard.
|
||||
def test_every_page_links_to_both_admin_pages(client) -> None:
|
||||
"""`⚙ Account` and `⚷ Certs` are the only way to the two admin pages, so they are everywhere.
|
||||
|
||||
It sits in `.masthead-right` before the theme toggle and **outside** `.tabs`, because the tab
|
||||
They sit in `.masthead-right` before the theme toggle and **outside** `.tabs`, because the tab
|
||||
strip's structure — five dashboards, one of them current — is pinned by
|
||||
`test_the_disks_tab_is_in_the_masthead_of_every_page` and a sixth entry with no
|
||||
`aria-current` would break what that marker means.
|
||||
|
||||
Neither page links to itself, which is the `/settings` precedent from phase 11 and the reason
|
||||
the two loops below are over different tuples: an `⚷ Certs` link on `/certs` is a no-op the
|
||||
operator learns to ignore, and ignored chrome is worse than none.
|
||||
"""
|
||||
link = '<a class="ghost account" href="/settings"'
|
||||
for path in PAGES:
|
||||
account = '<a class="ghost account" href="/settings"'
|
||||
certs_link = '<a class="ghost account" href="/certs"'
|
||||
for path in (*PAGES, SETTINGS_PAGE):
|
||||
html = client.get(path).text
|
||||
assert link in html, f"{path} has no route to the account page"
|
||||
assert certs_link in html, f"{path} has no route to the certificate authority"
|
||||
assert certs_link in html[:html.index('id="theme-toggle"')], \
|
||||
f"{path} put the Certs link after the theme toggle"
|
||||
tabs = html[html.index('<nav class="tabs"'):html.index("</nav>")]
|
||||
assert "/settings" not in tabs, f"{path} made Account a dashboard tab"
|
||||
assert html.index(link) < html.index('id="theme-toggle"'), \
|
||||
f"{path} put the account link after the theme toggle"
|
||||
# the stylesheet styles the link with the button's own rule, not a copy of it
|
||||
css = client.get("/static/dash.css").text
|
||||
assert "a.ghost" in css
|
||||
assert "/certs" not in tabs, f"{path} made Certs a dashboard tab"
|
||||
for path in (*PAGES, CERTS_PAGE):
|
||||
html = client.get(path).text
|
||||
assert account in html, f"{path} has no route to the account page"
|
||||
assert account in html[:html.index('id="theme-toggle"')], \
|
||||
f"{path} put the Account link after the theme toggle"
|
||||
assert account not in client.get(SETTINGS_PAGE).text, "/settings links to itself"
|
||||
assert certs_link not in client.get(CERTS_PAGE).text, "/certs links to itself"
|
||||
|
||||
|
||||
def test_the_certs_page_is_a_page_and_not_a_sixth_dashboard(client) -> None:
|
||||
"""`/certs` borrows the dashboard chrome and none of the dashboard meaning (task 15.05).
|
||||
|
||||
Five tabs with nothing current, no `#nav-status` (a CA page has no fleet reachability to
|
||||
report), and the same head and stylesheet as every other page — this is the
|
||||
`test_the_settings_page_is_not_one_of_the_dashboards` assertion for the second admin page,
|
||||
and it is what a sixth tab in the strip would fail on.
|
||||
"""
|
||||
html = client.get(CERTS_PAGE).text
|
||||
assert html.count('aria-current="page"') == 0
|
||||
tabs = re.findall(r'<a href="([^"]+)"[^>]*aria-current="page"', html)
|
||||
assert tabs == [], f"/certs marked a tab current: {tabs}"
|
||||
assert 'id="nav-status"' not in html, "/certs grew fleet dots it has nothing to fill"
|
||||
for path in PAGES:
|
||||
assert f'href="{path}"' in html, f"/certs cannot navigate to {path}"
|
||||
assert '/static/dash.css' in html
|
||||
|
||||
|
||||
def test_the_certs_page_carries_the_authority_and_the_agents_table(client) -> None:
|
||||
"""The three blocks the page is: the root, the client line, and one row per host.
|
||||
|
||||
Pinned on the served bytes rather than on a browser: the root and client panels and the table
|
||||
are markup, and the two rows of the table body are built from `GET /api/certs` by
|
||||
`page-certs.js` — so the per-row action labels are asserted on the served **script**, the way
|
||||
`/ups` pins its command buttons. `tests/test_api.py` has no JS runner (a locked anchor), and a
|
||||
label that shipped in neither file would be an invisible button either way.
|
||||
"""
|
||||
html = client.get(CERTS_PAGE).text
|
||||
js = client.get("/static/page-certs.js").text
|
||||
|
||||
assert "Root CA" in html and 'id="root-serial"' in html
|
||||
assert 'id="root-expires"' in html and 'id="root-left"' in html
|
||||
assert "Rotate root CA" in html, "the CA-wide action disappeared"
|
||||
assert "Dashboard client" in html and 'id="client-serial"' in html
|
||||
assert 'id="client-expires"' in html, "/certs lost the client expiry"
|
||||
assert 'id="agents-table"' in html and "<tbody></tbody>" in html
|
||||
assert 'id="agents-empty"' in html and "hidden" in html
|
||||
|
||||
for label in ("Rotate", "Revoke", "bundle"):
|
||||
assert f"'{label}'" in js, f"the {label} action lost its label"
|
||||
for route in ("/rotate", "/revoke", "/bundle"):
|
||||
assert route in js, f"the page no longer reaches {route}"
|
||||
assert "/api/certs/root/rotate" in js
|
||||
|
||||
|
||||
def test_the_certs_page_prints_the_names_the_authority_mints(client) -> None:
|
||||
"""The CN of each certificate is markup, so the literal is pinned to `certs.py`.
|
||||
|
||||
`MIN_PASSWORD_LENGTH`'s idiom: there is no build step to share a Python constant with a page,
|
||||
so a page that restates one says it in markup and a test compares the two. Printing a CN the
|
||||
module did not sign with would be the page inventing a fact about a stored certificate.
|
||||
"""
|
||||
html = client.get(CERTS_PAGE).text
|
||||
assert f">{certs.ROOT_CN}<" in html, "the root panel's CN is not what certs.py signs"
|
||||
assert f">{certs.CLIENT_CN}<" in html, "the client line's CN is not what certs.py signs"
|
||||
|
||||
|
||||
def test_the_certs_page_reads_only_fields_the_cert_api_publishes(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""End to end, in both directions: the page's column table and `GET /api/certs` are one shape.
|
||||
|
||||
The served script's `COLUMNS` entries name every payload key the page indexes, so this is the
|
||||
`SLOT_FIELDS` device on the surface where a rename is worst: a `deployed_bundle_rev` that
|
||||
quietly became `agent_bundle_rev` would print `n/a` beside a healthy fleet forever.
|
||||
"""
|
||||
script = "/static/page-certs.js"
|
||||
with build_client(tmp_path, monkeypatch, host_targets=cert_hosts(3),
|
||||
host_interval=3600.0) as client:
|
||||
js = client.get(script).text
|
||||
names: set[str] = set()
|
||||
for listed in re.findall(r"fields: \[([^\]]*)\]", js):
|
||||
names.update(re.findall(r"'([^']+)'", listed))
|
||||
assert names == CERT_AGENT_KEYS, sorted(names ^ CERT_AGENT_KEYS)
|
||||
|
||||
client.post("/api/certs/deskwork/rotate")
|
||||
for row in cert_agent_rows(client):
|
||||
assert len(row) == 8
|
||||
for name in names:
|
||||
assert name in row, f"the page indexes `{name}` and the API does not publish it"
|
||||
|
||||
# …and the two authority blocks: the page prints a serial and an expiry for each, which is
|
||||
# exactly the key set the API answers there.
|
||||
body = cert_document(client)
|
||||
for block in ("root", "client"):
|
||||
assert set(body[block]) == CERT_TIMESERIAL_KEYS
|
||||
for name in CERT_TIMESERIAL_KEYS:
|
||||
assert f".{name}" in js, f"the page stopped reading `{name}`"
|
||||
|
||||
|
||||
def test_the_certs_page_never_colours_a_certificate(client) -> None:
|
||||
"""No severity vocabulary reaches this page at all — an expiry is a date, not a verdict.
|
||||
|
||||
`/api/certs` sends no `severity` anywhere (task 04, and `severity.py` rates no certificate),
|
||||
so a page that coloured an expiring cert would be inventing a threshold in JavaScript. The one
|
||||
colour on the page is the payload's `status` word through a word→class map, and `unissued`
|
||||
maps to no class at all — a certificate that does not exist must not borrow green.
|
||||
"""
|
||||
js = _code(client.get("/static/page-certs.js").text)
|
||||
assert "severity" not in js, "/certs grew a verdict the API does not send"
|
||||
assert "severityFrom" not in js and "severityFor" not in js
|
||||
table = js[js.index("const STATUS_CHIP = {"):js.index("};")]
|
||||
assert "active: 'ok'" in table and "revoked: 'fail'" in table
|
||||
assert "unissued: ''" in table, "an unissued host would be coloured"
|
||||
assert "not_after" not in table and "serial" not in table, "an expiry became a verdict"
|
||||
|
||||
|
||||
def test_the_certs_page_asks_before_it_writes_and_offers_nothing_the_api_refuses(
|
||||
client) -> None:
|
||||
"""Every write is confirmed, and no row offers a control its own payload rules out.
|
||||
|
||||
`confirm()` and not the typed `DANGER` word: that gate belongs to commands that cut power,
|
||||
and this page has none — the worst outcome here is a host that is not dialled until somebody
|
||||
installs a bundle. The guards are the API's, restated: a row outside `HOST_TARGETS` gets no
|
||||
buttons at all (`404 {"configured": false}` is all three routes' first line), a never-issued
|
||||
host has nothing to revoke and no bundle to download, and the bundle download is a plain
|
||||
`GET` link because a session cookie is all a signed `GET` needs.
|
||||
"""
|
||||
js = _code(client.get("/static/page-certs.js").text)
|
||||
assert "window.confirm(" in js
|
||||
assert "DANGER" not in js, "the power-cut word migrated to a page with no power cut"
|
||||
assert "entry.configured" in js and "not in HOST_TARGETS" in js
|
||||
assert "entry.status !== 'unissued'" in js, "a row could revoke nothing"
|
||||
assert "entry.serial" in js
|
||||
assert "bundle.href = '/api/certs/'" in js
|
||||
assert "method: 'POST'" in js
|
||||
# …and the refusal is text, never markup: these are server sentences
|
||||
assert "innerHTML" not in js
|
||||
|
||||
|
||||
def test_the_certs_page_prints_the_restart_sentence_the_api_spoke(client) -> None:
|
||||
"""`restart_required` prints the route's `hint`, not a sentence the page wrote for itself.
|
||||
|
||||
The restart fact lives in `api.py` (the CA and the client pair are startup material, so the
|
||||
table is new before the socket is). A page copy of it is the version that quietly stops
|
||||
saying "and every bundle needs re-installing".
|
||||
"""
|
||||
js = client.get("/static/page-certs.js").text
|
||||
assert "restart_required" in js
|
||||
assert "body.hint" in js
|
||||
assert "restart the dashboard" not in js.lower(), "the page invented the restart sentence"
|
||||
|
||||
|
||||
def test_the_password_floor_the_form_checks_is_the_one_the_api_enforces(client) -> None:
|
||||
@@ -1000,7 +1221,7 @@ def test_a_rotated_secret_clears_the_cached_session_every_page_uses(client) -> N
|
||||
ASSETS = ("/static/dash.css", "/static/app.js", "/static/charts.js",
|
||||
"/static/page-index.js", "/static/page-disks.js", "/static/page-system.js",
|
||||
"/static/page-llm.js", "/static/page-ups.js", "/static/page-login.js",
|
||||
"/static/page-settings.js")
|
||||
"/static/page-settings.js", "/static/page-certs.js")
|
||||
|
||||
|
||||
def _css(client: TestClient) -> str:
|
||||
@@ -2894,6 +3115,9 @@ PAGE_SCRIPTS = {
|
||||
# `/settings` is audited like a dashboard even though it is not one: an id nothing reads is
|
||||
# dead markup whatever page it sits on, and this page has five new ones.
|
||||
SETTINGS_PAGE: "page-settings.js",
|
||||
# `/certs` too (phase 15 task 05): an authority panel and an agents table are markup, and
|
||||
# every id in them is written by this page's own script.
|
||||
CERTS_PAGE: "page-certs.js",
|
||||
}
|
||||
# `/login` is deliberately **not** in this table even though it is audited (phase 11): a
|
||||
# signed-in client asking for it gets a 303 to the dashboard, so this helper's own
|
||||
@@ -3459,14 +3683,19 @@ def test_the_disks_surface_is_gets_only_and_the_write_stays_the_ups_one(
|
||||
assert response.status_code == 405, f"{verb} /api/disks/snapshot was accepted"
|
||||
writes = sorted(str(route.path) for route in routes
|
||||
if "POST" in (getattr(route, "methods", None) or set()))
|
||||
# Invariant 1 is about the fleet, and phase 11 did not touch the fleet: the three `/auth/`
|
||||
# writes create, re-credential, and clear a session in this process and dial nothing. They
|
||||
# are named here instead of filtered out of sight, because the failure this assertion catches
|
||||
# is a POST arriving quietly — a list the reader recognises is worth more than a short one.
|
||||
upstream = [path for path in writes if not path.startswith("/api/auth/")]
|
||||
# Invariant 1 is about the fleet. The `/auth/` writes create, re-credential, and clear a
|
||||
# session in this process; the `/certs/` writes (phase 15) mint, reissue, and revoke a
|
||||
# certificate in this process's `meta` table. Neither dials anything — and a revocation's whole
|
||||
# effect is that the *next poll* refuses to dial — so they are named here rather than filtered
|
||||
# out of sight, because the failure this assertion catches is a POST arriving quietly.
|
||||
authority = [path for path in writes if path.startswith("/api/certs/")]
|
||||
session = [path for path in writes if path.startswith("/api/auth/")]
|
||||
upstream = [path for path in writes
|
||||
if not path.startswith(("/api/auth/", "/api/certs/"))]
|
||||
assert upstream == ["/api/ups/command"], "one write endpoint, four gates, unchanged"
|
||||
assert session == ["/api/auth/login", "/api/auth/logout", "/api/auth/password"]
|
||||
assert authority == ["/api/certs/root/rotate", "/api/certs/{name}/revoke",
|
||||
"/api/certs/{name}/rotate"], "the CA surface grew a write nobody named"
|
||||
# …and nothing anywhere grew the verbs this app has no use for, session included.
|
||||
assert not sorted(path for route in routes
|
||||
for path in [str(route.path)]
|
||||
@@ -3478,7 +3707,8 @@ def test_the_disks_surface_is_gets_only_and_the_write_stays_the_ups_one(
|
||||
and not str(route.path).startswith("/api/"))
|
||||
# `/settings` (phase 11) is a fifth dashboard-shaped **page** and it grew no verb: a page
|
||||
# route is a GET, so the write surface this test guards stays the one UPS command.
|
||||
assert pages == ["/", "/disks", "/llm", "/login", "/settings", "/system", "/ups"]
|
||||
assert pages == ["/", "/certs", "/disks", "/llm", "/login", "/settings", "/system",
|
||||
"/ups"]
|
||||
|
||||
|
||||
def test_config_prints_the_disks_cadence_the_page_explains_itself_with(
|
||||
@@ -4160,6 +4390,487 @@ def test_samples_written_by_the_poller_are_queryable(tmp_path: Path,
|
||||
store.close()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# The certificate authority surface (phase 15 task 04)
|
||||
#
|
||||
# These routes mint, re-sign, and revoke the identities every agent in the fleet is dialled with,
|
||||
# so the assertions are about **stored rows and wire effects** rather than about JSON being shaped
|
||||
# nicely: a read that issued a certificate, a typo that wrote a row, a revocation that did not
|
||||
# reach the collector, and a key that left `meta` are the four ways this surface can be wrong while
|
||||
# every status code still reads 200.
|
||||
#
|
||||
# The fleet here is loopback ports nobody listens on. That is deliberate: nothing on this surface
|
||||
# is allowed to need a live agent, so a refused connection is both the whole environment and the
|
||||
# control that says "a dial was attempted, and it was not this".
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
#: Three configured hosts, none of them reachable, on the scheme a protected exporter is dialled
|
||||
#: by. The URL host is also what a rotated certificate's SAN must carry, so these are the SANs the
|
||||
#: assertions below read.
|
||||
DEAD_TARGETS = (Target(name="deskwork", url="https://127.0.0.1:1"),
|
||||
Target(name="driveripper", url="https://127.0.0.1:2"),
|
||||
Target(name="gateway", url="https://127.0.0.1:3"))
|
||||
|
||||
#: The whole `GET /api/certs` document and one agent row, pinned as a table so a panel that
|
||||
#: retires can never shrink the payload (`/api/disks`' idiom) and task 05's page has a contract.
|
||||
CERT_DOCUMENT_KEYS = {"root", "client", "agents"}
|
||||
CERT_TIMESERIAL_KEYS = {"serial", "not_after"}
|
||||
CERT_AGENT_KEYS = {"name", "configured", "serial", "not_after", "status", "bundle_rev",
|
||||
"deployed_bundle_rev", "san"}
|
||||
|
||||
|
||||
def cert_document(client: TestClient) -> dict[str, Any]:
|
||||
"""`GET /api/certs`, with the status asserted: a refusal here would pass a `.get()` chain."""
|
||||
response = client.get("/api/certs")
|
||||
assert response.status_code == 200, response.text
|
||||
body = response.json()
|
||||
assert set(body) == CERT_DOCUMENT_KEYS, "the CA document grew or lost a top-level key"
|
||||
for block in ("root", "client"):
|
||||
assert set(body[block]) == CERT_TIMESERIAL_KEYS, block
|
||||
assert int(body[block]["serial"]) > 0, "a serial is the decimal text of a positive int"
|
||||
assert body[block]["not_after"] > time.time(), "a stored CA may not already be expired"
|
||||
return body
|
||||
|
||||
|
||||
def cert_agent_rows(client: TestClient, name: str | None = None) -> list[dict[str, Any]]:
|
||||
"""The agent rows, every one of them shaped, optionally just one host's."""
|
||||
rows = cert_document(client)["agents"]
|
||||
for row in rows:
|
||||
assert set(row) == CERT_AGENT_KEYS, sorted(row)
|
||||
if name is None:
|
||||
return rows
|
||||
return [row for row in rows if row["name"] == name]
|
||||
|
||||
|
||||
def cert_context(client: TestClient) -> Any:
|
||||
"""The app's `AppContext`.
|
||||
|
||||
`TestClient.app` is typed as the ASGI callable it wraps, so the one cast lives here instead of
|
||||
one `# type: ignore` at every assertion that needs the store or a job.
|
||||
"""
|
||||
return client.app.state.ctx # type: ignore[attr-defined]
|
||||
|
||||
|
||||
def cert_meta(store: SampleStore) -> dict[str, str]:
|
||||
"""Every `certs.*` row, for the before/after comparisons that prove a route wrote nothing.
|
||||
|
||||
Coalesced to `""` because the comparison is "did any byte of these five rows move", and an
|
||||
absent row is the empty answer to that question rather than a second kind of nothing.
|
||||
"""
|
||||
return {key: store.get_meta(key) or "" for key in certs.META_KEYS}
|
||||
|
||||
|
||||
def revoked_serials(bundle: str) -> list[str]:
|
||||
"""The denylist of a bundle, as the agent reads it: the `# REVOKED <serial>` lines."""
|
||||
return [line.removeprefix(certs.REVOKED_LINE).strip()
|
||||
for line in bundle.splitlines() if line.startswith(certs.REVOKED_LINE)]
|
||||
|
||||
|
||||
#: One PEM block out of a bundle, by text — the same read task 02's agent does with stdlib.
|
||||
_PEM_BLOCK = re.compile(r"-----BEGIN [A-Z0-9 ]+-----.*?-----END [A-Z0-9 ]+-----\n", re.DOTALL)
|
||||
|
||||
|
||||
def bundle_certs(bundle: str) -> list[x509.Certificate]:
|
||||
"""The two certificates of a bundle, in the order the format writes them.
|
||||
|
||||
`[0]` is the agent's own server certificate and `[1]` the root it chains to (`certs.py`'s
|
||||
`_bundle_text` order: metadata lines, agent cert, agent key, root cert). Reading them out of the
|
||||
served bytes rather than trusting the response's echo is what makes "did the rotation reach the
|
||||
artifact?" a checked claim.
|
||||
"""
|
||||
blocks = [block for block in _PEM_BLOCK.findall(bundle) if "CERTIFICATE" in block]
|
||||
assert len(blocks) == 2, f"a bundle is two certificates and a key, got {len(blocks)}"
|
||||
return [x509.load_pem_x509_certificate(block.encode()) for block in blocks]
|
||||
|
||||
|
||||
def bundle_of(client: TestClient, name: str) -> str:
|
||||
"""The served bundle, asserted to still be a bundle (2 certs, 1 key) before it is compared."""
|
||||
response = client.get(f"/api/certs/{name}/bundle")
|
||||
assert response.status_code == 200, response.text
|
||||
assert response.headers["content-type"].startswith("application/x-pem-file")
|
||||
assert "no-store" in response.headers["cache-control"], "a private key may not be cached"
|
||||
blocks = re.findall(r"-----BEGIN [A-Z0-9 ]+-----", response.text)
|
||||
assert blocks.count("-----BEGIN CERTIFICATE-----") == 2, blocks
|
||||
assert blocks.count("-----BEGIN PRIVATE KEY-----") == 1, blocks
|
||||
return response.text
|
||||
|
||||
|
||||
def stored_bundle(store: SampleStore, name: str) -> str:
|
||||
"""The bundle exactly as the `certs.agents` row holds it."""
|
||||
row: dict[str, Any] = json.loads(store.get_meta(certs.META_AGENTS) or "{}")
|
||||
return str(row[name]["bundle"])
|
||||
|
||||
|
||||
def cert_events(client: TestClient) -> list[dict[str, Any]]:
|
||||
return client.get("/api/events", params={"source": "certs"}).json()["events"]
|
||||
|
||||
|
||||
def cert_hosts(count: int) -> tuple[Target, ...]:
|
||||
return DEAD_TARGETS[:count]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("hosts", [0, 1, 3], ids=["none", "one", "three"])
|
||||
def test_the_cert_list_is_the_stored_table_for_zero_one_and_three_hosts(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, hosts: int) -> None:
|
||||
"""The list works with no fleet at all, and says *unissued* rather than omitting a host.
|
||||
|
||||
The CA exists whether or not `HOST_TARGETS` does — a fresh database boots with a root and a
|
||||
client certificate — so this is not the `/api/disks` "not configured" surface and does not 404.
|
||||
A configured host with nothing issued still gets a row (invariant 3: a machine that is deployed
|
||||
and not yet installed is not an absent machine), with `None` and not `0` where there is no
|
||||
certificate (rule 2).
|
||||
"""
|
||||
with build_client(tmp_path, monkeypatch, host_targets=cert_hosts(hosts),
|
||||
host_interval=3600.0) as client:
|
||||
rows = cert_agent_rows(client)
|
||||
assert len(rows) == hosts, "one row per configured host, and no more"
|
||||
assert [row["name"] for row in rows] == sorted(row["name"] for row in rows), (
|
||||
"one order in the payload and one in the table")
|
||||
for row in rows:
|
||||
assert row["configured"] is True
|
||||
assert row["status"] == "unissued", "nothing has been signed for it yet"
|
||||
assert row["serial"] is None and row["bundle_rev"] is None, "no invented zeros"
|
||||
assert row["deployed_bundle_rev"] is None, "nobody has answered, so nobody reported"
|
||||
assert row["san"] == [], "no names on record, not a certificate with no names"
|
||||
|
||||
|
||||
def test_the_cert_list_carries_the_rev_the_agent_reported(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""`deployed_bundle_rev` is what the host said about itself, not what the CA hopes.
|
||||
|
||||
Task 05's "stale bundle" sentence is a comparison of two numbers that come from two different
|
||||
places: the row this dashboard issued and the `agent` block of the last `system` payload. Both
|
||||
halves are pinned here, including the direction that matters — issued 2 against deployed 1 is
|
||||
the operator's "install the new bundle" case, and it has to be visible rather than smoothed
|
||||
over.
|
||||
"""
|
||||
with build_client(tmp_path, monkeypatch, host_targets=cert_hosts(1),
|
||||
host_interval=3600.0) as client:
|
||||
assert client.post("/api/certs/deskwork/rotate").status_code == 200
|
||||
job = cert_context(client).hub.jobs["host:deskwork"]
|
||||
assert cert_agent_rows(client, "deskwork")[0]["deployed_bundle_rev"] is None, (
|
||||
"a host that has never answered reports nothing")
|
||||
job.state.payload = {"name": "deskwork", "agent": {"tls": True, "bundle_rev": 1}}
|
||||
assert cert_agent_rows(client, "deskwork")[0]["deployed_bundle_rev"] == 1
|
||||
assert client.post("/api/certs/deskwork/rotate").json()["bundle_rev"] == 2
|
||||
row = cert_agent_rows(client, "deskwork")[0]
|
||||
assert (row["bundle_rev"], row["deployed_bundle_rev"]) == (2, 1), "the stale pair, both"
|
||||
# A token-mode agent publishes no `agent.tls` block at all, and that is `n/a`, not rev 0.
|
||||
job.state.payload = {"name": "deskwork", "agent": {"version": "1.0.0"}}
|
||||
assert cert_agent_rows(client, "deskwork")[0]["deployed_bundle_rev"] is None
|
||||
|
||||
|
||||
def test_an_entry_for_a_host_that_left_the_config_is_listed_and_not_writable(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""A row issued for a machine that has gone is a key still sitting in the table.
|
||||
|
||||
Listing it is the honest answer (`/certs` is where you learn you still hold one); refusing to
|
||||
rotate or revoke it is the same invariant from the other side — `HOST_TARGETS` is what this
|
||||
dashboard dials, and a name it no longer dials is not a name to install a bundle on. The row is
|
||||
written straight to the store because that is the only way this state arises: a host removed
|
||||
from the config, never a route that created it.
|
||||
"""
|
||||
with build_client(tmp_path, monkeypatch, host_targets=cert_hosts(1),
|
||||
host_interval=3600.0) as client:
|
||||
store: SampleStore = cert_context(client).store
|
||||
certs.issue_agent_cert(store, certs.read_state(store), "old-host", ["10.9.8.7"])
|
||||
|
||||
rows = cert_agent_rows(client)
|
||||
assert [row["name"] for row in rows] == ["deskwork", "old-host"]
|
||||
ghost = rows[1]
|
||||
assert ghost["configured"] is False, "listed, but this dashboard does not dial it"
|
||||
assert ghost["status"] == certs.ACTIVE and ghost["bundle_rev"] == 1
|
||||
assert ghost["deployed_bundle_rev"] is None, "no job means no report, not rev 0"
|
||||
assert client.post("/api/certs/old-host/rotate").status_code == 404
|
||||
assert client.post("/api/certs/old-host/revoke").status_code == 404
|
||||
assert cert_events(client) == [], "a refusal is not an audit line"
|
||||
|
||||
|
||||
def test_a_cert_route_for_an_unknown_name_dials_nothing_and_writes_nothing(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""The invariant-3 gate on the per-host routes, taken before anything can have a side effect.
|
||||
|
||||
`not-installed` is refused in the first line of each route, so no client is built, no store row
|
||||
is written, and no key is minted. All three are measured: the two network spies raise, `set_meta`
|
||||
is watched for any `certs.*` write, and the five rows are compared byte for byte. A typo that
|
||||
minted a certificate would be invisible on every status code and permanent in the table.
|
||||
"""
|
||||
with build_client(tmp_path, monkeypatch, host_targets=cert_hosts(1),
|
||||
host_interval=3600.0) as client:
|
||||
store: SampleStore = cert_context(client).store
|
||||
calls: list[str] = []
|
||||
written: list[str] = []
|
||||
|
||||
async def spy_get(self: Any, url: Any, **kwargs: Any) -> Any:
|
||||
calls.append(f"GET {url}")
|
||||
raise AssertionError(f"an unconfigured cert name dialled {url}")
|
||||
|
||||
def spy_fetch(*args: Any, **kwargs: Any) -> Any:
|
||||
calls.append("fetch_snapshot")
|
||||
raise AssertionError("an unconfigured cert name fetched an agent snapshot")
|
||||
|
||||
real_set_meta = SampleStore.set_meta
|
||||
|
||||
def spy_set_meta(self: SampleStore, key: str, value: str) -> None:
|
||||
written.append(key)
|
||||
real_set_meta(self, key, value)
|
||||
|
||||
monkeypatch.setattr(httpx.AsyncClient, "get", spy_get)
|
||||
monkeypatch.setattr("dashboards.collectors.fetch_snapshot", spy_fetch)
|
||||
monkeypatch.setattr(SampleStore, "set_meta", spy_set_meta)
|
||||
before = cert_meta(store)
|
||||
|
||||
for response in (client.get("/api/certs/not-configured/bundle"),
|
||||
client.post("/api/certs/not-configured/rotate"),
|
||||
client.post("/api/certs/not-configured/revoke")):
|
||||
assert response.status_code == 404, response.text
|
||||
body = response.json()
|
||||
assert body["configured"] is False
|
||||
assert "HOST_TARGETS" in body["detail"], "the sentence names the fix"
|
||||
|
||||
assert calls == [], "the gate is in front of the network, not behind it"
|
||||
assert [key for key in written if key.startswith("certs.")] == [], "a write happened"
|
||||
assert cert_meta(store) == before, "the stored CA and agent table did not move"
|
||||
assert cert_events(client) == [], "and no audit line was written for a name that is not here"
|
||||
# …and the same route still works for the host that *is* configured, so the refusal above
|
||||
# is the gate and not a broken router.
|
||||
assert client.post("/api/certs/deskwork/rotate").status_code == 200
|
||||
|
||||
|
||||
def test_rotating_a_host_mints_the_bundle_it_asks_for(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Rotate is also *issue*: first call is `bundle_rev` 1, and the SAN is the config's address.
|
||||
|
||||
The SAN assertion is the one a rollout fails on silently — a certificate that does not name the
|
||||
address the dashboard dials is a handshake error and a red source, not a warning — so it is read
|
||||
out of the served bundle's own certificate rather than out of the response's echo.
|
||||
"""
|
||||
with build_client(tmp_path, monkeypatch, host_targets=cert_hosts(1),
|
||||
host_interval=3600.0) as client:
|
||||
store: SampleStore = cert_context(client).store
|
||||
body = client.post("/api/certs/deskwork/rotate").json()
|
||||
assert body["ok"] is True and body["status"] == certs.ACTIVE
|
||||
assert body["bundle_rev"] == 1 and body["serial"] and body["not_after"] > time.time()
|
||||
assert body["san"] == ["127.0.0.1", "deskwork"], "the dialled address and the name"
|
||||
|
||||
cert = bundle_certs(bundle_of(client, "deskwork"))[0]
|
||||
names = cert.extensions.get_extension_for_class(
|
||||
x509.SubjectAlternativeName).value.get_values_for_type(x509.IPAddress)
|
||||
assert names == [ipaddress.ip_address("127.0.0.1")], "an IP SAN, or an IP dial fails"
|
||||
assert cert.subject == x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "deskwork")])
|
||||
|
||||
events = cert_events(client)
|
||||
assert len(events) == 1 and events[0]["kind"] == "cert-rotated"
|
||||
assert events[0]["host"] == "deskwork" and events[0]["message"] == (
|
||||
"agent certificate rotated for deskwork")
|
||||
assert events[0]["data"] == {"bundle_rev": 1}
|
||||
|
||||
second = client.post("/api/certs/deskwork/rotate").json()
|
||||
assert second["bundle_rev"] == 2 and second["serial"] != body["serial"]
|
||||
# The new identity is not on the denylist: rotation replaces an identity, it does not strike
|
||||
# one out, and a bundle that listed its own serial would refuse the client it ships.
|
||||
bundle = bundle_of(client, "deskwork")
|
||||
assert revoked_serials(bundle) == [], revoked_serials(bundle)
|
||||
assert second["serial"] not in bundle.splitlines(), "no denylist line for the live serial"
|
||||
assert stored_bundle(store, "deskwork") == bundle, "served bytes are the stored bytes"
|
||||
|
||||
|
||||
def test_revoking_a_host_marks_the_table_and_the_next_poll_refuses_without_a_dial(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""The API half of the revocation chain: this route is what task 03's refusal reads.
|
||||
|
||||
The control is the sentence from *before* the click (`unreachable`, which is what a dial to a
|
||||
dead port says), and after it the poll has to say `agent certificate revoked` with a spy on
|
||||
`httpx.AsyncClient.get` proving no socket was opened. The stored denylist is the other half: an
|
||||
agent holding a bundle that omits the serial keeps accepting the client it was given.
|
||||
"""
|
||||
with build_client(tmp_path, monkeypatch, host_targets=cert_hosts(1),
|
||||
host_interval=0.05, disks_interval=0.05) as client:
|
||||
store: SampleStore = cert_context(client).store
|
||||
wait_for_source_error(client, "host:deskwork", "unreachable")
|
||||
serial = client.post("/api/certs/deskwork/rotate").json()["serial"]
|
||||
|
||||
body = client.post("/api/certs/deskwork/revoke").json()
|
||||
assert body["ok"] is True and body["status"] == certs.REVOKED
|
||||
assert body["serial"] == serial, "revocation marks the identity it has, it does not replace"
|
||||
assert body["hint"] and "next poll" in body["hint"]
|
||||
|
||||
calls: list[str] = []
|
||||
|
||||
async def spy_get(self: Any, url: Any, **kwargs: Any) -> Any:
|
||||
calls.append(str(url))
|
||||
raise AssertionError(f"a revoked agent was dialled at {url}")
|
||||
|
||||
monkeypatch.setattr(httpx.AsyncClient, "get", spy_get)
|
||||
for key in ("host:deskwork", "disks:deskwork"):
|
||||
error = wait_for_source_error(client, key, "agent certificate revoked")
|
||||
assert "unreachable" not in error
|
||||
assert calls == []
|
||||
|
||||
assert revoked_serials(bundle_of(client, "deskwork")) == [serial]
|
||||
events = cert_events(client)
|
||||
assert [row["kind"] for row in events] == ["cert-revoked", "cert-rotated"]
|
||||
assert events[0]["message"] == "agent certificate revoked for deskwork"
|
||||
assert certs.agent_status(store, "deskwork") == certs.REVOKED, (
|
||||
"the collector's own read agrees with the row the route wrote")
|
||||
|
||||
|
||||
def test_root_rotation_reissues_every_host_and_answers_restart_required(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""One action re-arms the fleet — and says the part it cannot do from inside the process.
|
||||
|
||||
`restart_required` is not decoration: the CA and the client pair are startup material, so the
|
||||
table is new immediately while the sockets keep the old identity. The page (task 05) prints
|
||||
that sentence because the API answered it, and the alternative is an operator who rotated a
|
||||
root and watches a fleet that still chains to the previous one.
|
||||
"""
|
||||
with build_client(tmp_path, monkeypatch, host_targets=cert_hosts(3),
|
||||
host_interval=3600.0) as client:
|
||||
store: SampleStore = cert_context(client).store
|
||||
for target in DEAD_TARGETS:
|
||||
assert client.post(f"/api/certs/{target.name}/rotate").status_code == 200
|
||||
before = cert_document(client)
|
||||
rows_before = {row["name"]: row for row in before["agents"]}
|
||||
root_pem_before = store.get_meta(certs.META_ROOT_CERT) or ""
|
||||
key_before = store.get_meta(certs.META_ROOT_KEY) or ""
|
||||
assert root_pem_before and key_before, "both halves of the CA exist before the rotation"
|
||||
|
||||
body = client.post("/api/certs/root/rotate").json()
|
||||
assert body["ok"] is True and body["restart_required"] is True
|
||||
assert body["agents"] == 3
|
||||
assert body["root"]["serial"] != before["root"]["serial"], "a fresh root"
|
||||
|
||||
after = cert_document(client)
|
||||
assert after["root"] == body["root"], "the list reports what the rotation wrote"
|
||||
assert after["client"]["serial"] != before["client"]["serial"], "and a fresh client leaf"
|
||||
for row in after["agents"]:
|
||||
was = rows_before[row["name"]]
|
||||
assert row["bundle_rev"] == was["bundle_rev"] + 1, "every host reissued, rev bumped"
|
||||
assert row["serial"] != was["serial"] and row["status"] == certs.ACTIVE
|
||||
bundle = bundle_of(client, row["name"])
|
||||
leaf, root = bundle_certs(bundle)
|
||||
assert leaf.serial_number == int(row["serial"]), "the served bundle is the listed one"
|
||||
assert root.serial_number == int(after["root"]["serial"]), (
|
||||
"and it chains to the authority the list now reports")
|
||||
assert revoked_serials(bundle) == [], "the denylist resets with the generation"
|
||||
assert store.get_meta(certs.META_ROOT_KEY) != key_before
|
||||
assert store.get_meta(certs.META_ROOT_CERT) != root_pem_before
|
||||
assert root_pem_before not in "".join(cert_meta(store).values()), (
|
||||
"the old root is gone")
|
||||
events = cert_events(client)
|
||||
assert events[0]["kind"] == "cert-root-rotated"
|
||||
assert events[0]["message"] == (
|
||||
"root certificate rotated — every agent certificate reissued, restart required")
|
||||
assert events[0]["data"] == {"agents": 3}
|
||||
|
||||
|
||||
def test_the_bundle_route_names_a_file_and_changes_when_the_identity_does(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Byte-identical to the stored row, an `attachment`, and different bytes after a reissue.
|
||||
|
||||
This is the installer's input (task 06), so the claim is about bytes and not about a bundle
|
||||
that merely re-verifies: `install-agent.sh` is handed the file and compares nothing else.
|
||||
"""
|
||||
with build_client(tmp_path, monkeypatch, host_targets=cert_hosts(1),
|
||||
host_interval=3600.0) as client:
|
||||
store: SampleStore = cert_context(client).store
|
||||
missing = client.get("/api/certs/deskwork/bundle")
|
||||
assert missing.status_code == 404, "nothing issued yet, so nothing to download"
|
||||
body = missing.json()
|
||||
assert body["configured"] is True and body["issued"] is False, "not the same 404"
|
||||
assert "rotate" in body["detail"], "and the answer names the way to get one"
|
||||
|
||||
first = client.post("/api/certs/deskwork/rotate").json()["serial"]
|
||||
response = client.get("/api/certs/deskwork/bundle")
|
||||
assert response.headers["content-disposition"] == (
|
||||
'attachment; filename="agent-deskwork.bundle"')
|
||||
assert response.text == stored_bundle(store, "deskwork")
|
||||
|
||||
client.post("/api/certs/deskwork/rotate")
|
||||
second = client.post("/api/certs/deskwork/rotate").json()["serial"]
|
||||
assert second != first
|
||||
updated = client.get("/api/certs/deskwork/bundle")
|
||||
assert updated.text != response.text, "a reissue changes the served bytes"
|
||||
assert updated.text == stored_bundle(store, "deskwork")
|
||||
|
||||
|
||||
def test_no_certificate_material_leaves_meta_through_the_cert_surface(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""invariant 6 on the API that owns the keys: serials and dates go out, PEMs do not.
|
||||
|
||||
Every write here is exercised first, so the grep is over a table that has been rotated and
|
||||
revoked rather than a pristine one, and it covers both the payloads a browser reads and the
|
||||
stored event rows those payloads are built from — an event row is kept forever, which is what
|
||||
makes a PEM in a message permanent rather than transient.
|
||||
"""
|
||||
with build_client(tmp_path, monkeypatch, host_targets=cert_hosts(2),
|
||||
host_interval=3600.0) as client:
|
||||
store: SampleStore = cert_context(client).store
|
||||
client.post("/api/certs/deskwork/rotate")
|
||||
client.post("/api/certs/driveripper/rotate")
|
||||
client.post("/api/certs/driveripper/revoke")
|
||||
client.post("/api/certs/root/rotate")
|
||||
secrets = [value for value in cert_meta(store).values() if value]
|
||||
assert any("PRIVATE KEY" in value for value in secrets), "the table really does hold keys"
|
||||
|
||||
for path in ("/api/certs", "/api/health", "/api/events", "/api/summary", "/api/config"):
|
||||
text = client.get(path).text
|
||||
for marker in ("BEGIN CERTIFICATE", "PRIVATE KEY", "BUNDLE-REV"):
|
||||
assert marker not in text, f"{path} served {marker}"
|
||||
assert client.get("/api/health").json()["db"], "health still answers, keys or no keys"
|
||||
for event in store.recent_events(50, source="certs"):
|
||||
assert "BEGIN" not in event.message and "PRIVATE KEY" not in event.message
|
||||
assert not any(secret in json.dumps(event.as_json()) for secret in secrets)
|
||||
|
||||
|
||||
def test_a_host_named_root_cannot_be_rotated_past_the_certificate_authority(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""`/api/certs/root/rotate` and `/api/certs/{name}/rotate` are one URL for a host called root.
|
||||
|
||||
The CA route is registered first, so it wins — which means the wrong button would rotate the
|
||||
whole authority instead of one host's leaf. Refusing with the reason is the only honest answer:
|
||||
the alternative is a fleet-wide outage behind a click that looked like routine maintenance.
|
||||
"""
|
||||
with build_client(tmp_path, monkeypatch,
|
||||
host_targets=(Target(name="root", url="https://127.0.0.1:1"),),
|
||||
host_interval=3600.0) as client:
|
||||
store: SampleStore = cert_context(client).store
|
||||
before = cert_document(client)["root"]["serial"]
|
||||
refused = client.post("/api/certs/root/rotate")
|
||||
assert refused.status_code == 409, refused.text
|
||||
assert "rename" in refused.json()["detail"], "the sentence says what to do"
|
||||
assert cert_document(client)["root"]["serial"] == before, "and the CA did not move"
|
||||
assert cert_events(client) == []
|
||||
assert cert_meta(store)[certs.META_AGENTS] == "{}", "nothing was issued for the name either"
|
||||
|
||||
|
||||
def test_a_read_of_the_cert_surface_issues_nothing_and_writes_nothing(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""A GET that mints a certificate is the surprise this route must never be.
|
||||
|
||||
`certs.ensure_certs` is the bootstrapping read and `certs.read_state` is the request-time one;
|
||||
making `ensure_certs` explode is what proves the difference rather than asserting it in prose,
|
||||
and the meta rows are compared byte for byte across three reads.
|
||||
"""
|
||||
with build_client(tmp_path, monkeypatch, host_targets=cert_hosts(1),
|
||||
host_interval=3600.0) as client:
|
||||
store: SampleStore = cert_context(client).store
|
||||
before = cert_meta(store)
|
||||
|
||||
def explode(*args: Any, **kwargs: Any) -> Any:
|
||||
raise AssertionError("a GET tried to bootstrap the certificate authority")
|
||||
|
||||
monkeypatch.setattr(certs, "ensure_certs", explode)
|
||||
assert cert_agent_rows(client)
|
||||
assert cert_document(client)["root"]["serial"] == str(
|
||||
x509.load_pem_x509_certificate(
|
||||
str(before[certs.META_ROOT_CERT]).encode()).serial_number)
|
||||
# A bundle that has not been issued is still a 404 rather than a mint, on a read too.
|
||||
assert client.get("/api/certs/deskwork/bundle").status_code == 404
|
||||
assert cert_meta(store) == before, "three reads, no write"
|
||||
assert cert_events(client) == []
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# phase 14 task 04: the endpoint budgets
|
||||
#
|
||||
|
||||
+112
-6
@@ -42,7 +42,7 @@ from auth_contract import FAST_PBKDF2
|
||||
from dashboards import auth
|
||||
from dashboards.api import MIN_PASSWORD_LENGTH, AppContext
|
||||
from dashboards.app import PUBLIC_PATHS, PUBLIC_PREFIXES, create_app, is_public
|
||||
from dashboards.config import Settings
|
||||
from dashboards.config import Settings, Target
|
||||
from dashboards.poller import Hub
|
||||
from dashboards.sources import nut
|
||||
from dashboards.store import SampleStore
|
||||
@@ -58,11 +58,17 @@ PROTECTED_API = (
|
||||
"/api/system/snapshot", "/api/system/history", "/api/disks/snapshot",
|
||||
"/api/disks/history", "/api/llm/snapshot", "/api/llm/history", "/api/ups/snapshot",
|
||||
"/api/ups/history", "/api/ups/commands", "/api/auth/status",
|
||||
# Phase 15: the certificate authority's own surface. `/api/certs/<name>/bundle` is every
|
||||
# agent's private key, so "it needs a session" is the whole point of listing it here.
|
||||
"/api/certs", "/api/certs/{name}/bundle", "/api/certs/{name}/rotate",
|
||||
"/api/certs/{name}/revoke", "/api/certs/root/rotate",
|
||||
)
|
||||
#: Every HTML page behind the gate. `/settings` is a **page** and not one of the four
|
||||
#: dashboards (it carries no `aria-current` tab), which is exactly why it is in this list: the
|
||||
#: one thing that must never regress is a new route answering anonymously.
|
||||
PROTECTED_PAGES = ("/", "/system", "/disks", "/llm", "/ups", "/settings")
|
||||
#: Every HTML page behind the gate. `/settings` and `/certs` are **pages** and not dashboards
|
||||
#: (neither carries an `aria-current` tab), which is exactly why they are in this list: the one
|
||||
#: thing that must never regress is a new route answering anonymously. `/certs` is the sharper
|
||||
#: case of the two — an anonymous caller reaching it would see the CA's serials and every host
|
||||
#: the dashboard dials.
|
||||
PROTECTED_PAGES = ("/", "/system", "/disks", "/llm", "/ups", "/settings", "/certs")
|
||||
|
||||
|
||||
#: The one write this app has, with the fields the route needs (the UPS is unconfigured, so a
|
||||
@@ -1256,6 +1262,105 @@ def test_the_danger_word_is_still_what_cuts_power_after_the_gate(nut_app: TestCl
|
||||
assert WireRecorder.sent == [("apc", "driver.killpower")]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# The certificate authority's writes (phase 15 task 04)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cert_app(tmp_path: Path) -> Iterator[TestClient]:
|
||||
"""A signed-in client on an app with **one configured agent** and no live host.
|
||||
|
||||
What these routes *do* is `tests/test_api.py`'s business; this file asks only whether the gate
|
||||
stops a write before it reaches `certs.py`. The target is a dead loopback port because a
|
||||
certificate route is never allowed to dial anything (invariant 1) — the stored agent table is
|
||||
the wire here, which is what the assertions below read.
|
||||
"""
|
||||
db = tmp_path / "certs-gate.sqlite3"
|
||||
store = SampleStore(str(db))
|
||||
try:
|
||||
app = create_app(Settings(db_path=str(db), admin_password=ADMIN_PASSWORD,
|
||||
pbkdf2_iterations=FAST_PBKDF2,
|
||||
host_targets=(Target(name="deskwork",
|
||||
url="https://127.0.0.1:1"),),
|
||||
host_interval=3600.0, disks_interval=3600.0), store)
|
||||
with TestClient(app, follow_redirects=False) as client:
|
||||
sign_in(client)
|
||||
yield client
|
||||
finally:
|
||||
store.close()
|
||||
|
||||
|
||||
def test_anonymous_calls_see_no_certificate_list_and_no_bundle(anon: TestClient) -> None:
|
||||
"""The named, explicit version of the route-walk proof.
|
||||
|
||||
`test_every_route_that_is_not_public_requires_a_session` walks the served table and cannot miss
|
||||
a route; this test is here so the *bundle* route — every agent's private key, behind a login —
|
||||
has one assertion that says what it is about, and so a redirect that would leak the existence of
|
||||
a host name into a `Location` header is caught as deliberately as the 401 is.
|
||||
"""
|
||||
listing = anon.get("/api/certs", headers=JSON_ACCEPT)
|
||||
assert listing.status_code == 401
|
||||
assert listing.json() == {"ok": False, "detail": "sign in required", "login": "/login"}
|
||||
assert "no-store" in listing.headers["cache-control"]
|
||||
|
||||
bundle = anon.get("/api/certs/deskwork/bundle", headers=JSON_ACCEPT)
|
||||
assert bundle.status_code == 401, "a bundle is a private key with a filename on it"
|
||||
assert "location" not in bundle.headers
|
||||
|
||||
for write in (anon.post("/api/certs/deskwork/rotate"),
|
||||
anon.post("/api/certs/deskwork/revoke"),
|
||||
anon.post("/api/certs/root/rotate")):
|
||||
assert write.status_code == 401, write.url
|
||||
assert write.json() == {"ok": False, "detail": "sign in required", "login": "/login"}
|
||||
assert "no-store" in write.headers["cache-control"]
|
||||
|
||||
|
||||
def test_a_cert_write_without_the_csrf_token_changes_no_stored_row(
|
||||
cert_app: TestClient) -> None:
|
||||
"""The UPS wire test's shape on the CA's own writes, with the row as the wire.
|
||||
|
||||
Proven in both directions on purpose: a gate that refused every write would pass a test that
|
||||
only asserted the refusal, and a route that wrote before checking would pass one that only
|
||||
asserted the success. The stored agent table is what a rotation changes, so it is what is read.
|
||||
"""
|
||||
def issued() -> dict[str, Any]:
|
||||
rows: list[dict[str, Any]] = cert_app.get("/api/certs").json()["agents"]
|
||||
return rows[0]
|
||||
|
||||
assert issued()["status"] == "unissued" and issued()["bundle_rev"] is None
|
||||
|
||||
csrf = str(cert_app.headers.get(auth.CSRF_HEADER))
|
||||
del cert_app.headers[auth.CSRF_HEADER]
|
||||
refused = cert_app.post("/api/certs/deskwork/rotate")
|
||||
assert refused.status_code == 403
|
||||
assert "csrf" in refused.json()["detail"].lower()
|
||||
assert issued()["status"] == "unissued", "a key was minted by a write the gate refused"
|
||||
|
||||
cert_app.headers[auth.CSRF_HEADER] = csrf
|
||||
accepted = cert_app.post("/api/certs/deskwork/rotate")
|
||||
assert accepted.status_code == 200, accepted.text
|
||||
assert accepted.json()["bundle_rev"] == 1
|
||||
assert issued()["bundle_rev"] == 1
|
||||
|
||||
|
||||
def test_a_cross_origin_cert_write_is_refused_before_the_authority_is_touched(
|
||||
cert_app: TestClient) -> None:
|
||||
"""The origin half of the gate on a route whose effect is durable and fleet-wide."""
|
||||
cross_site = cert_app.post("/api/certs/deskwork/revoke",
|
||||
headers={"origin": "https://elsewhere.example"})
|
||||
assert cross_site.status_code == 403
|
||||
assert "cross-origin" in cross_site.json()["detail"]
|
||||
|
||||
stalled = cert_app.post("/api/certs/root/rotate", headers={"sec-fetch-site": "cross-site"})
|
||||
assert stalled.status_code == 403
|
||||
assert cert_app.get("/api/certs").json()["agents"][0]["status"] == "unissued", (
|
||||
"nothing moved: not the host's identity and not the root")
|
||||
|
||||
# …and the same request with nothing forged is an ordinary gated write.
|
||||
assert cert_app.post("/api/certs/deskwork/revoke").status_code == 200
|
||||
|
||||
|
||||
def test_a_cross_origin_write_is_refused_even_with_a_session_and_a_csrf_token(
|
||||
authed: TestClient) -> None:
|
||||
response = authed.post("/api/ups/command", json=WRITE,
|
||||
@@ -1493,7 +1598,8 @@ def test_the_served_route_table_is_the_surface_the_gate_describes(anon: TestClie
|
||||
pages = sorted(str(route.path) for route in served_routes(anon)
|
||||
if isinstance(getattr(route, "path", None), str)
|
||||
and not str(route.path).startswith(("/api", "/static")))
|
||||
assert pages == ["/", "/disks", "/llm", "/login", "/settings", "/system", "/ups"]
|
||||
assert pages == ["/", "/certs", "/disks", "/llm", "/login", "/settings", "/system",
|
||||
"/ups"]
|
||||
for path in pages:
|
||||
response = anon.get(path, headers=BROWSER_HEADERS)
|
||||
if path == "/login":
|
||||
|
||||
+1094
File diff suppressed because it is too large
Load Diff
@@ -10,21 +10,30 @@ transitions); the socket protocol itself is in test_nut.py.
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import threading
|
||||
from http.server import ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import Any, ClassVar
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from dashboards import certs
|
||||
from dashboards.collectors import (
|
||||
AgentTls,
|
||||
NutMonitor,
|
||||
UpstreamError,
|
||||
agent_tls_map,
|
||||
disks_jobs,
|
||||
fetch_disk_snapshot,
|
||||
fetch_engine,
|
||||
fetch_host_snapshot,
|
||||
fetch_snapshot,
|
||||
host_jobs,
|
||||
)
|
||||
from dashboards.config import Target
|
||||
from dashboards.poller import Result
|
||||
from dashboards.rates import RateEstimator
|
||||
from dashboards.sources import nut
|
||||
from dashboards.sources.disks import DisksCollector
|
||||
@@ -351,6 +360,403 @@ async def test_smart_addon_errors_reach_the_disk_payload_verbatim(
|
||||
assert "degraded" not in system.payload, "and /system stays unbothered by it"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# TLS: the dashboard dials a protected exporter with its own client certificate
|
||||
# --------------------------------------------------------------------------- #
|
||||
#
|
||||
# The certificates come from `dashboards.certs` over a dict store and the server is the real
|
||||
# `agent/host_agent.py` in the TLS mode task 02 added, so this proves the dial itself: the
|
||||
# client certificate is presented, the agent is verified against the root this dashboard minted,
|
||||
# and a host the operator revoked is refused before a socket opens. `httpx.MockTransport` is
|
||||
# kept for the sentences (a 403 body is the agent's words, not a handshake).
|
||||
|
||||
|
||||
class DictMeta:
|
||||
"""The three-method `certs.MetaStore` slice, backed by a dict (the `test_certs.py` idiom)."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.rows: dict[str, str] = {}
|
||||
|
||||
def get_meta(self, key: str) -> str | None:
|
||||
return self.rows.get(key)
|
||||
|
||||
def set_meta(self, key: str, value: str) -> None:
|
||||
self.rows[key] = value
|
||||
|
||||
def delete_meta(self, key: str) -> int:
|
||||
return 1 if self.rows.pop(key, None) is not None else 0
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tls_fleet(tmp_path: Path, host_agent_module: Any) -> Any:
|
||||
"""One dashboard CA + client pair, one issued agent bundle, and the agent serving it.
|
||||
|
||||
Everything production has: the material written to a private directory (`httpx` takes
|
||||
paths), the bundle installed where `install-agent.sh` will put it, and the `AgentTls` the
|
||||
jobs get — built by `agent_tls_map`, the same factory `app.build_hub` is handed, so the
|
||||
tests dial with the production wiring and not a hand-built guess of it.
|
||||
"""
|
||||
store = DictMeta()
|
||||
state = certs.ensure_certs(store)
|
||||
state = certs.issue_agent_cert(store, state, "localhost", ["127.0.0.1", "localhost"])
|
||||
entry = state.agent("localhost")
|
||||
assert entry is not None
|
||||
material = certs.write_material(state, tmp_path)
|
||||
bundle_path = tmp_path / "dashboards-agent.certs"
|
||||
bundle_path.write_text(entry.bundle, encoding="ascii")
|
||||
bundle, error = host_agent_module.TlsBundle.load(str(bundle_path))
|
||||
assert error is None and bundle is not None
|
||||
host_agent_module.Handler.collector = host_agent_module.Collector("/", smart=False)
|
||||
host_agent_module.Handler.token = None
|
||||
host_agent_module.Handler.bundle = bundle
|
||||
httpd = ThreadingHTTPServer(("127.0.0.1", 0), host_agent_module.Handler)
|
||||
httpd.socket = bundle.context.wrap_socket(httpd.socket, server_side=True)
|
||||
thread = threading.Thread(target=httpd.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
base = f"https://127.0.0.1:{httpd.server_address[1]}"
|
||||
target = Target("localhost", base)
|
||||
try:
|
||||
yield SimpleNamespace(
|
||||
store=store,
|
||||
state=state,
|
||||
material=material,
|
||||
bundle=bundle,
|
||||
bundle_rev=entry.bundle_rev,
|
||||
base=base,
|
||||
target=target,
|
||||
tls=agent_tls_map(material, store, [target])["localhost"],
|
||||
)
|
||||
finally:
|
||||
httpd.shutdown()
|
||||
httpd.server_close()
|
||||
# `Handler` is process-wide state: a later test must not inherit this bundle.
|
||||
host_agent_module.Handler.bundle = None
|
||||
host_agent_module.Handler.token = None
|
||||
certs.remove_material(material)
|
||||
|
||||
|
||||
async def test_an_https_agent_answers_the_dashboard_that_presents_its_certificate(
|
||||
tls_fleet: SimpleNamespace) -> None:
|
||||
"""The whole point of the phase, collector side: our cert opens the agent.
|
||||
|
||||
The two fields the agent adds in TLS mode come back too, because task 05's page compares the
|
||||
revision `/certs` issued with the revision the host is actually running.
|
||||
"""
|
||||
async with httpx.AsyncClient(timeout=15.0) as client:
|
||||
body = await fetch_snapshot(client, tls_fleet.target, tls=tls_fleet.tls)
|
||||
assert body["ok"] is True
|
||||
assert body["tls"] is True
|
||||
assert body["bundle_rev"] == tls_fleet.bundle_rev
|
||||
result = await fetch_host_snapshot(client, tls_fleet.target,
|
||||
HostCollector(RateEstimator()), tls=tls_fleet.tls)
|
||||
assert result.payload["agent"]["tls"] is True
|
||||
assert result.payload["agent"]["bundle_rev"] == tls_fleet.bundle_rev
|
||||
assert result.payload["memory"]["total_bytes"] > 0, "the snapshot is a real one"
|
||||
|
||||
|
||||
async def test_an_agent_signed_by_another_authority_is_not_trusted(
|
||||
tls_fleet: SimpleNamespace, tmp_path: Path) -> None:
|
||||
"""A root that is not ours verifies nothing here, so the dial fails instead of accepting.
|
||||
|
||||
Trusting any certificate would make the whole phase a no-op with extra steps, and the
|
||||
failure has to reach the poller as an `UpstreamError` naming the host rather than as a
|
||||
traceback out of `ssl`.
|
||||
"""
|
||||
stranger_base = tmp_path / "stranger"
|
||||
stranger_base.mkdir()
|
||||
stranger = certs.write_material(certs.ensure_certs(DictMeta()), stranger_base)
|
||||
foreign = agent_tls_map(stranger, tls_fleet.store, [tls_fleet.target])["localhost"]
|
||||
async with httpx.AsyncClient(timeout=15.0) as client:
|
||||
with pytest.raises(UpstreamError) as caught:
|
||||
await fetch_snapshot(client, tls_fleet.target, tls=foreign)
|
||||
assert "unreachable" in caught.value.message
|
||||
assert tls_fleet.base in caught.value.message
|
||||
certs.remove_material(stranger)
|
||||
|
||||
|
||||
async def test_dashboard_material_that_cannot_be_read_fails_the_poll(
|
||||
tls_fleet: SimpleNamespace, tmp_path: Path) -> None:
|
||||
"""A missing client-cert file is a red source with a sentence, not a dead poll loop.
|
||||
|
||||
`httpx` loads the CA and the client chain when the client is *built*, so this arrives as a
|
||||
`FileNotFoundError` from inside the fetch. A TLS dial that can only fail with a non-HTTP
|
||||
exception is the invariant-4 failure mode: the task would die and the source would look
|
||||
healthy while its charts stopped.
|
||||
"""
|
||||
broken = AgentTls(verify=str(tls_fleet.material.ca),
|
||||
cert=(str(tmp_path / "gone-cert.pem"), str(tmp_path / "gone-key.pem")),
|
||||
status=lambda: certs.ACTIVE)
|
||||
async with httpx.AsyncClient(timeout=15.0) as client:
|
||||
with pytest.raises(UpstreamError, match="unreachable") as caught:
|
||||
await fetch_snapshot(client, tls_fleet.target, tls=broken)
|
||||
assert "FileNotFoundError" in caught.value.message
|
||||
assert tls_fleet.base in caught.value.message, "the failure still names the host's URL"
|
||||
|
||||
|
||||
async def test_an_https_agent_is_not_readable_without_the_material(tls_fleet: SimpleNamespace) -> None:
|
||||
"""A shared client with no client certificate gets nowhere — "only this dashboard can read
|
||||
an agent" has to hold for the collector's own client too, not just for a stranger's.
|
||||
"""
|
||||
async with httpx.AsyncClient(timeout=15.0) as client:
|
||||
with pytest.raises(UpstreamError) as caught:
|
||||
await fetch_snapshot(client, tls_fleet.target)
|
||||
assert "unreachable" in caught.value.message
|
||||
|
||||
|
||||
async def test_a_revoked_agent_is_refused_before_anything_is_dialled(
|
||||
tls_fleet: SimpleNamespace, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""The revocation gate is in front of the network, not behind it.
|
||||
|
||||
The spy replaces `httpx.AsyncClient.get` on the class, so it catches the shared client and
|
||||
the per-dial TLS client alike: what is being proved is that no request is built at all, not
|
||||
that one particular client was unused. A dial-then-refuse would still put a socket on a host
|
||||
the operator just struck out.
|
||||
"""
|
||||
calls: list[str] = []
|
||||
|
||||
async def spy_get(self: Any, url: Any, **kwargs: Any) -> Any:
|
||||
calls.append(str(url))
|
||||
raise AssertionError(f"a revoked agent was dialled at {url}")
|
||||
|
||||
monkeypatch.setattr(httpx.AsyncClient, "get", spy_get)
|
||||
certs.revoke(tls_fleet.store, tls_fleet.state, "localhost")
|
||||
async with httpx.AsyncClient(timeout=15.0) as client:
|
||||
with pytest.raises(UpstreamError, match="agent certificate revoked") as caught:
|
||||
await fetch_snapshot(client, tls_fleet.target, tls=tls_fleet.tls)
|
||||
with pytest.raises(UpstreamError, match="agent certificate revoked"):
|
||||
await fetch_disk_snapshot(client, tls_fleet.target, DisksCollector(),
|
||||
tls=tls_fleet.tls)
|
||||
assert calls == [], "a revoked host must not be dialled on either cadence"
|
||||
assert "revoked" in caught.value.message
|
||||
|
||||
|
||||
async def test_the_revocation_answer_is_a_live_read_of_the_store(tls_fleet: SimpleNamespace) -> None:
|
||||
"""`status` is a callable because the operator clicks revoke while this process is running.
|
||||
|
||||
A status string read at startup would keep dialling a struck-out host until the next
|
||||
restart, and a dashboard that needs a restart to enforce a security decision does not
|
||||
enforce it.
|
||||
"""
|
||||
tls = tls_fleet.tls
|
||||
assert tls.status() == certs.ACTIVE
|
||||
assert tls.revoked is False
|
||||
certs.revoke(tls_fleet.store, tls_fleet.state, "localhost")
|
||||
assert tls.status() == certs.REVOKED
|
||||
assert tls.revoked is True, "the same object, with no rebuild and no restart"
|
||||
|
||||
|
||||
def test_the_revocation_is_per_host_and_the_material_is_shared(tls_fleet: SimpleNamespace) -> None:
|
||||
"""One identity for every dial, one revocation answer per name.
|
||||
|
||||
`certs.py` gives the dashboard a single client certificate, so `verify`/`cert` are the same
|
||||
paths for every host; what differs is whose identity the operator struck out. A name with no
|
||||
entry is active rather than refused — never-issued is not the same fact as revoked, and
|
||||
answering revoked for it would take a working host red with no operator action behind it
|
||||
(invariant 3).
|
||||
"""
|
||||
targets = [tls_fleet.target, Target("deskwork", "https://10.1.0.51:9120")]
|
||||
mapping = agent_tls_map(tls_fleet.material, tls_fleet.store, targets)
|
||||
assert mapping["localhost"].verify == str(tls_fleet.material.ca)
|
||||
assert mapping["localhost"].cert == mapping["deskwork"].cert == (
|
||||
str(tls_fleet.material.client_cert), str(tls_fleet.material.client_key))
|
||||
certs.revoke(tls_fleet.store, tls_fleet.state, "localhost")
|
||||
assert mapping["localhost"].status() == certs.REVOKED
|
||||
assert mapping["deskwork"].status() == certs.ACTIVE
|
||||
assert agent_tls_map(None, tls_fleet.store, targets) == {}, (
|
||||
"no material means no TLS dial, exactly the pre-phase-15 path")
|
||||
|
||||
|
||||
def scripted_refusal(body: Any, *, status: int = 403) -> AgentTls:
|
||||
"""An agent that answers every `/snapshot` with a refusal, in whatever shape it sent the body."""
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
if isinstance(body, dict):
|
||||
return httpx.Response(status, json=body)
|
||||
return httpx.Response(status, text=body)
|
||||
|
||||
return AgentTls(verify="/unused/root.pem", cert=("/unused/cert.pem", "/unused/key.pem"),
|
||||
status=lambda: certs.ACTIVE, transport=httpx.MockTransport(handler))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("body,expected", [
|
||||
({"ok": False, "detail": "client certificate required"}, "client certificate required"),
|
||||
({"ok": False, "detail": "client certificate revoked"}, "client certificate revoked"),
|
||||
("<html><body>403 Forbidden</body></html>", "403 Forbidden"),
|
||||
("", "no reason given"),
|
||||
])
|
||||
async def test_a_refusal_from_a_protected_agent_quotes_its_own_words(
|
||||
body: Any, expected: str) -> None:
|
||||
"""The 403 sentence names the refusal the *agent* made, redacted and never blank.
|
||||
|
||||
An `UpstreamError` string goes into the health record and the event table, so it has to say
|
||||
which of the agent's two refusals this was (a required certificate and a revoked one are two
|
||||
different operator actions) while quoting nothing but the URL it was dialling
|
||||
(AGENTS.md invariant 6). The `verify`/`cert` paths here do not exist: a scripted refusal
|
||||
proves the wording, and the tests above prove the handshake.
|
||||
"""
|
||||
tls = scripted_refusal(body)
|
||||
target = Target("gw", "https://SEKRIT-AGENT-TOKEN@agent.test")
|
||||
async with httpx.AsyncClient(timeout=15.0) as client:
|
||||
with pytest.raises(UpstreamError) as caught:
|
||||
await fetch_snapshot(client, target, tls=tls)
|
||||
message = caught.value.message
|
||||
assert "agent refused the client certificate" in message
|
||||
assert expected in message
|
||||
assert "SEKRIT-AGENT-TOKEN" not in message, "the token leaked into a browser-readable string"
|
||||
assert "https://agent.test/snapshot" in message, "and it still says where"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("status", [401, 403, 500])
|
||||
async def test_a_refusal_from_an_unprotected_agent_keeps_the_status_wording(status: int) -> None:
|
||||
"""The certificate sentence is only for a dial where a certificate was presented.
|
||||
|
||||
A token-mode host answers 401 for a bad token and would answer 403 for something that has
|
||||
nothing to do with certificates; calling that "refused the client certificate" would send
|
||||
the operator to `/certs` for a token problem.
|
||||
"""
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(status, text="nope")
|
||||
|
||||
target = Target("gw", "http://agent.test")
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
||||
with pytest.raises(UpstreamError) as caught:
|
||||
await fetch_snapshot(client, target,
|
||||
tls=AgentTls(verify="/unused/root.pem",
|
||||
cert=("/unused/c.pem", "/unused/k.pem"),
|
||||
status=lambda: certs.ACTIVE))
|
||||
if status == 401:
|
||||
assert "token" in caught.value.message
|
||||
else:
|
||||
assert f"HTTP {status}" in caught.value.message
|
||||
assert "client certificate" not in caught.value.message
|
||||
|
||||
|
||||
async def test_a_token_mode_host_still_uses_the_shared_client(tls_fleet: SimpleNamespace) -> None:
|
||||
"""An `http://` entry is dialled exactly as it was before phase 15.
|
||||
|
||||
The material exists in this deployment (one CA, one client pair, every host), so the scheme
|
||||
is the only thing that may decide the path — a rollout is per host, and the plain hosts
|
||||
cannot be allowed to pick up a client certificate they never asked for.
|
||||
"""
|
||||
seen: list[str] = []
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
seen.append(str(request.url))
|
||||
return httpx.Response(200, json=agent_body())
|
||||
|
||||
plain = Target("localhost", "http://agent.test")
|
||||
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
|
||||
body = await fetch_snapshot(client, plain, tls=tls_fleet.tls)
|
||||
assert body["ok"] is True
|
||||
assert seen == ["http://agent.test/snapshot"], "the shared client, the shared transport"
|
||||
|
||||
|
||||
async def test_a_revoked_agent_fails_both_jobs_with_one_sentence(store: Any,
|
||||
tls_fleet: SimpleNamespace,
|
||||
monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""The phase's criterion, at the level the health record is written: `host:` and `disks:`.
|
||||
|
||||
The two jobs dial one agent on two cadences, so a revocation has to fail both with the same
|
||||
sentence (one dead source must not blink the other, and both must stop talking to it).
|
||||
"""
|
||||
calls: list[str] = []
|
||||
|
||||
async def spy_get(self: Any, url: Any, **kwargs: Any) -> Any:
|
||||
calls.append(str(url))
|
||||
raise AssertionError(f"a revoked agent was dialled at {url}")
|
||||
|
||||
monkeypatch.setattr(httpx.AsyncClient, "get", spy_get)
|
||||
certs.revoke(tls_fleet.store, tls_fleet.state, "localhost")
|
||||
async with httpx.AsyncClient(timeout=15.0) as client:
|
||||
jobs = host_jobs([tls_fleet.target], client, HostCollector(RateEstimator()), 10.0,
|
||||
tls={"localhost": tls_fleet.tls})
|
||||
jobs += disks_jobs([tls_fleet.target], client, DisksCollector(), 300.0,
|
||||
tls={"localhost": tls_fleet.tls})
|
||||
assert [job.key for job in jobs] == ["host:localhost", "disks:localhost"]
|
||||
for job in jobs:
|
||||
assert await job.run_once(store) is False
|
||||
assert [job.state.error for job in jobs] == [
|
||||
"UpstreamError: localhost: agent certificate revoked"] * 2
|
||||
assert calls == []
|
||||
|
||||
|
||||
async def test_host_and_disk_jobs_hand_each_host_its_own_material(
|
||||
monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""The plumbing, pinned: a host in the map gets its entry, a host outside it gets `None`.
|
||||
|
||||
Two hosts and one wrong-key miss is the minimum that shows the lookup is by name rather than
|
||||
"whatever was built first", which is how a revoked host would keep being dialled with
|
||||
another host's revocation answer.
|
||||
"""
|
||||
seen: dict[str, Any] = {}
|
||||
|
||||
def record(name: str) -> Any:
|
||||
async def scrape(client: Any, target: Target, *_args: Any, **kwargs: Any) -> Result:
|
||||
seen[f"{name}:{target.name}"] = kwargs.get("tls")
|
||||
return Result(payload={}, samples=[], events=[])
|
||||
|
||||
return scrape
|
||||
|
||||
fetch_host = record("system")
|
||||
fetch_disk = record("disks")
|
||||
monkeypatch.setattr("dashboards.collectors.fetch_host_snapshot", fetch_host)
|
||||
monkeypatch.setattr("dashboards.collectors.fetch_disk_snapshot", fetch_disk)
|
||||
entry = AgentTls(verify="/ca.pem", cert=("/c.pem", "/k.pem"),
|
||||
status=lambda: certs.ACTIVE)
|
||||
targets = [Target("gw", "https://gw.test"), Target("desk", "http://desk.test")]
|
||||
async with httpx.AsyncClient() as client:
|
||||
for job in (host_jobs(targets, client, HostCollector(RateEstimator()), 10.0,
|
||||
tls={"gw": entry})
|
||||
+ disks_jobs(targets, client, DisksCollector(), 300.0, tls={"gw": entry})):
|
||||
await job.fn()
|
||||
assert seen == {"system:gw": entry, "system:desk": None,
|
||||
"disks:gw": entry, "disks:desk": None}
|
||||
|
||||
|
||||
async def test_the_tls_self_report_reaches_both_payloads(disk_snapshot: Any) -> None:
|
||||
"""`tls` + `bundle_rev` travel into `/api/system/snapshot` and `/api/disks/snapshot`.
|
||||
|
||||
Task 05's page has to answer "is the installed bundle the one I just issued?", and both
|
||||
dashboards read the same snapshot, so both carry the deployed revision under the same
|
||||
`agent` key. Nothing grades it: a revision is an identity counter, not a threshold.
|
||||
"""
|
||||
def stamped(snapshot: dict[str, Any]) -> dict[str, Any]:
|
||||
"""What a TLS-mode agent's `/snapshot` looks like on the wire."""
|
||||
snapshot["tls"] = True
|
||||
snapshot["bundle_rev"] = 3
|
||||
return snapshot
|
||||
|
||||
target = Target("driveripper", "http://agent.test")
|
||||
async with client_for(lambda request: httpx.Response(200,
|
||||
json=stamped(agent_body()))) as client:
|
||||
system = (await fetch_host_snapshot(client, target,
|
||||
HostCollector(RateEstimator()))).payload
|
||||
assert system["agent"]["tls"] is True
|
||||
assert system["agent"]["bundle_rev"] == 3
|
||||
|
||||
# the `/disks` payload is built from a capture that also carries SMART blocks, so it gets
|
||||
# its own stamped snapshot rather than sharing the one above
|
||||
with_smart = disk_snapshot(mutate=stamped)
|
||||
async with client_for(lambda request: httpx.Response(200, json=with_smart)) as client:
|
||||
disks = (await fetch_disk_snapshot(client, target, DisksCollector())).payload
|
||||
assert disks["agent"] == {"tls": True, "bundle_rev": 3}
|
||||
|
||||
|
||||
async def test_a_token_mode_snapshot_gains_no_tls_fields() -> None:
|
||||
"""The captures are token-mode agents, and their payloads must stay byte-identical.
|
||||
|
||||
`false` and `0` would be numbers the agent never published (rule 2), and an empty block
|
||||
would be a container that means nothing.
|
||||
"""
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(200, json=agent_body())
|
||||
|
||||
target = Target("driveripper", "http://agent.test")
|
||||
async with client_for(handler) as client:
|
||||
system = (await fetch_host_snapshot(client, target, HostCollector(RateEstimator()))).payload
|
||||
assert "tls" not in system["agent"] and "bundle_rev" not in system["agent"]
|
||||
assert system["agent"]["version"], "the block is still the block it was"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# LLM engines
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
@@ -12,15 +12,21 @@ apart from the path the script was invoked through.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import stat
|
||||
import subprocess
|
||||
import threading
|
||||
from http.server import ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from dashboards import certs
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
DEPLOY = ROOT / "deploy"
|
||||
INSTALLER = DEPLOY / "install-agent.sh"
|
||||
@@ -291,6 +297,7 @@ ROOT_BOUNDARY_STUBS = ("id", "getent", "useradd", "usermod", "chown", "systemctl
|
||||
SANDBOX_REWRITES = {
|
||||
'AGENT_DIR="/usr/local/lib/dashboards-agent"': 'AGENT_DIR="{root}/lib"',
|
||||
'TOKEN_FILE="/etc/dashboards-agent.token"': 'TOKEN_FILE="{root}/token"',
|
||||
'CERT_FILE="/etc/dashboards-agent.certs"': 'CERT_FILE="{root}/certs"',
|
||||
'UNIT="/etc/systemd/system/$SERVICE.service"': 'UNIT="{root}/dashboards-agent.service"',
|
||||
}
|
||||
|
||||
@@ -361,6 +368,10 @@ def test_install_path_installs_into_its_own_paths_and_prints_the_token_once(tmp_
|
||||
assert "User=dashboards-agent" in unit
|
||||
assert f"--token-file {token_path}" in unit, "the unit must use the file, never --token"
|
||||
assert "--host 10.0.0.5 --port 9120" in unit
|
||||
# A token-only install runs no TLS, so it gets neither the bundle flag nor the directory
|
||||
# the bundle unpacks into (the agent falls back to a private mkdtemp without one).
|
||||
assert "--cert-bundle" not in unit, "no bundle installed means no TLS flag on a token install"
|
||||
assert "RuntimeDirectory=dashboards-agent" not in unit
|
||||
# DevicePolicy=strict denies every node not listed, and systemd opens /dev/null for stdin
|
||||
# before it execs python: without this line the unit dies at spawn (status 208/STDIN).
|
||||
assert "DevicePolicy=strict" in unit
|
||||
@@ -396,3 +407,370 @@ def test_token_file_is_readable_by_the_service_user_and_nobody_else(tmp_path: Pa
|
||||
mode = stat.S_IMODE((tmp_path / "target" / "token").stat().st_mode)
|
||||
assert mode == 0o640, f"token file mode is {oct(mode)}, expected 0o640 (group = the agent user)"
|
||||
assert not mode & 0o007, "the token file must never be world-readable"
|
||||
|
||||
|
||||
# --- the certificate bundle (phase 15 task 06) --------------------------------------------
|
||||
|
||||
#: Where the installed bundle lives — one constant in the script, pinned here so a rename
|
||||
#: fails a test instead of stranding every already-written unit file.
|
||||
CERT_FILE_PATH = "/etc/dashboards-agent.certs"
|
||||
|
||||
|
||||
def _fake_bundle_text(revision: str = "3") -> str:
|
||||
"""A syntactically real bundle: agent cert, agent key, root CA, and the rev line.
|
||||
|
||||
Shape only. The installer's validator greps PEM blocks by text exactly like
|
||||
`resolve_bundle` does (a monitored host has no `cryptography`), so a well-formed *fake*
|
||||
is all an unprivileged validation run needs; the functional probe tests below use a
|
||||
real bundle minted by `dashboards.certs`.
|
||||
"""
|
||||
|
||||
def block(kind: str, body: str) -> str:
|
||||
return f"-----BEGIN {kind}-----\n{body}\n-----END {kind}-----\n"
|
||||
|
||||
return (
|
||||
f"# BUNDLE-REV {revision}\n"
|
||||
+ block("CERTIFICATE", "AGENTCERT")
|
||||
+ block("PRIVATE KEY", "AGENTKEY")
|
||||
+ block("CERTIFICATE", "ROOTCA")
|
||||
)
|
||||
|
||||
|
||||
def _installer_status(stdout: str) -> str:
|
||||
"""The line `status()` prints as `health: …`, parsed as JSON when it is JSON."""
|
||||
for line in stdout.splitlines():
|
||||
if line.startswith("health: "):
|
||||
body = line.removeprefix("health: ").strip()
|
||||
try:
|
||||
return str(json.loads(body)["ok"])
|
||||
except (json.JSONDecodeError, KeyError):
|
||||
return body
|
||||
pytest.fail(f"the probe printed no health line:\n{stdout}")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("shape", "reason"),
|
||||
[
|
||||
("missing", "cannot be read"),
|
||||
("empty", "is empty"),
|
||||
("no-revision", "BUNDLE-REV"),
|
||||
("two-keys", "the agent certificate, the agent key and the root CA"),
|
||||
],
|
||||
)
|
||||
def test_a_bad_cert_bundle_is_refused_with_its_reason_before_anything_else(
|
||||
tmp_path: Path, shape: str, reason: str
|
||||
) -> None:
|
||||
"""Completion criterion: `--cert-bundle /nonexistent` refuses with the reason, touches nothing.
|
||||
|
||||
The validation runs *before* `need_root`, so the failure names the file rather than
|
||||
sending the operator to add a sudoers line for a typo in the flag's argument — and
|
||||
nothing after the validation (user, files, unit) can have run.
|
||||
"""
|
||||
if shape == "missing":
|
||||
path = tmp_path / "absent.bundle"
|
||||
elif shape == "empty":
|
||||
path = tmp_path / "empty.bundle"
|
||||
path.write_text("\n", encoding="ascii")
|
||||
elif shape == "no-revision":
|
||||
path = tmp_path / "revless.bundle"
|
||||
path.write_text(_fake_bundle_text().replace("# BUNDLE-REV 3\n", ""), encoding="ascii")
|
||||
else: # a bundle with the root *key* in it is exactly what the shape check refuses
|
||||
path = tmp_path / "leaky.bundle"
|
||||
extra = "-----BEGIN PRIVATE KEY-----\nROOTKEY\n-----END PRIVATE KEY-----\n"
|
||||
path.write_text(_fake_bundle_text() + extra, encoding="ascii")
|
||||
result = _installer_without_root("--cert-bundle", str(path))
|
||||
assert result.returncode == 2, "the bundle refusal keeps its own exit code (2)"
|
||||
assert result.stdout == "", "a refusal goes to stderr, not stdout"
|
||||
assert reason in result.stderr
|
||||
assert str(path) in result.stderr, "the message names the file the operator passed"
|
||||
assert GRANT_PREFIX not in result.stderr, "a bad bundle is not a sudoers problem"
|
||||
|
||||
|
||||
def test_a_well_formed_bundle_passes_validation_and_reaches_the_root_check(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Validation is shape-only and never dials: the next refusal is the usual sudoers one."""
|
||||
path = tmp_path / "agent.bundle"
|
||||
path.write_text(_fake_bundle_text(), encoding="ascii")
|
||||
result = _installer_without_root("--bind", "10.0.0.5", "--cert-bundle", str(path))
|
||||
assert result.returncode == 1
|
||||
assert _grant_lines(result.stderr), "the bundle validated; what remains is the root boundary"
|
||||
|
||||
|
||||
def test_installer_help_documents_the_cert_bundle_and_its_path() -> None:
|
||||
result = _run([BASH, str(INSTALLER), "--help"])
|
||||
assert result.returncode == 0
|
||||
assert "--cert-bundle" in result.stdout
|
||||
assert CERT_FILE_PATH in result.stdout
|
||||
|
||||
|
||||
def test_the_bundle_lands_at_the_token_file_idiom_in_the_script() -> None:
|
||||
"""`0640 root:$AGENT_USER` + chown: the agent opens the bundle itself and re-reads it."""
|
||||
text = INSTALLER.read_text()
|
||||
assert f'CERT_FILE="{CERT_FILE_PATH}"' in text
|
||||
assert 'install -m 0640 "$BUNDLE_SRC" "$CERT_FILE"' in text
|
||||
assert 'chown root:"$AGENT_USER" "$CERT_FILE"' in text
|
||||
# 0600 would fail the unit the way it would fail the token file: the service user opens
|
||||
# the bundle at startup and re-reads the denylist live. Pin the reasoning, not just the mode.
|
||||
assert "0600 root-owned file would fail the unit" in text
|
||||
assert "re-reads it when its stamp changes" in text
|
||||
|
||||
|
||||
def test_the_tls_unit_lines_come_from_one_bundle_branch() -> None:
|
||||
"""Flag + RuntimeDirectory are emitted together, or the unit is a token-mode unit."""
|
||||
text = INSTALLER.read_text()
|
||||
match = re.search(r'\n if \[ -f "\$CERT_FILE" \]; then(.*?)\n fi\n', text, re.DOTALL)
|
||||
assert match, "the TLS unit lines must come from one `if [ -f \"$CERT_FILE\" ]` branch"
|
||||
block = match.group(1)
|
||||
assert 'TLS_EXEC_ARGS=" --cert-bundle $CERT_FILE"' in block
|
||||
assert "RuntimeDirectory=dashboards-agent" in block
|
||||
assert "${TLS_EXEC_ARGS}" in text, "ExecStart must carry the flag from that branch"
|
||||
|
||||
|
||||
def test_status_probes_tls_with_the_bundle_ca_and_no_client_cert() -> None:
|
||||
"""The status branch: unit says --cert-bundle → python3 ssl probe, never openssl, no cert."""
|
||||
status_body = INSTALLER.read_text().split("status() {", 1)[1].split("\n}", 1)[0]
|
||||
assert "grep -q -- '--cert-bundle'" in status_body
|
||||
assert "PROTOCOL_TLS_CLIENT" in status_body
|
||||
assert "load_verify_locations" in status_body
|
||||
assert "https://" in status_body
|
||||
assert "load_cert_chain" not in status_body, "the installer holds no client certificate"
|
||||
assert 'http://$BIND:$PORT/health' in status_body, "token mode keeps the plain probe"
|
||||
assert "openssl" not in INSTALLER.read_text(), "no new binary dependency on a monitored host"
|
||||
|
||||
|
||||
@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
|
||||
def test_a_bundle_install_lands_the_file_the_unit_flag_and_the_runtime_directory(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""The install body with a bundle: byte-identical file at 0640, TLS unit, https hint."""
|
||||
bundle = tmp_path / "agent.bundle"
|
||||
bundle.write_text(_fake_bundle_text(), encoding="ascii")
|
||||
script, call_log, env = _sandbox_installer(tmp_path)
|
||||
result = subprocess.run(
|
||||
[BASH, str(script), "--bind", "10.0.0.5", "--port", "9120", "--cert-bundle", str(bundle)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
timeout=60,
|
||||
check=False,
|
||||
)
|
||||
assert result.returncode == 0, f"install failed:\n{result.stdout}\n{result.stderr}"
|
||||
|
||||
cert_path = tmp_path / "target" / "certs"
|
||||
assert cert_path.is_file(), "install -m never landed the bundle"
|
||||
assert cert_path.read_text(encoding="ascii") == _fake_bundle_text(), (
|
||||
"the installed bundle must be byte-identical to the downloaded one"
|
||||
)
|
||||
mode = stat.S_IMODE(cert_path.stat().st_mode)
|
||||
assert mode == 0o640, f"bundle mode is {oct(mode)}, expected 0o640 (group = the agent user)"
|
||||
assert not mode & 0o007, "the bundle holds the agent's key: never world-readable"
|
||||
|
||||
unit = (tmp_path / "target" / "dashboards-agent.service").read_text()
|
||||
assert f"--cert-bundle {cert_path}" in unit
|
||||
assert "RuntimeDirectory=dashboards-agent" in unit, (
|
||||
"ProtectSystem=strict leaves the agent no writable file but this, and the bundle "
|
||||
"unpacks its PEM blocks there"
|
||||
)
|
||||
assert "--token-file" in unit, "the dual-mode line stays — rollout is host by host"
|
||||
assert 'HOST_TARGETS="hostname=https://10.0.0.5:9120"' in result.stdout, (
|
||||
"a TLS install points the dashboard at https, with no token in the URL"
|
||||
)
|
||||
|
||||
calls = call_log.read_text()
|
||||
assert calls.count("chown root:dashboards-agent") == 2, "token and bundle both chowned"
|
||||
|
||||
|
||||
@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
|
||||
def test_an_unrelated_rerun_reuses_the_installed_bundle(tmp_path: Path) -> None:
|
||||
"""The token-reuse idiom: a re-run without the flag neither rewrites nor drops the bundle."""
|
||||
script, _call_log, env = _sandbox_installer(tmp_path)
|
||||
target = tmp_path / "target"
|
||||
target.mkdir(parents=True)
|
||||
cert_path = target / "certs"
|
||||
cert_path.write_text(_fake_bundle_text(revision="7"), encoding="ascii")
|
||||
result = subprocess.run(
|
||||
[BASH, str(script), "--bind", "10.0.0.5"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
timeout=60,
|
||||
check=False,
|
||||
)
|
||||
assert result.returncode == 0, f"re-run failed:\n{result.stdout}\n{result.stderr}"
|
||||
assert cert_path.read_text(encoding="ascii") == _fake_bundle_text(revision="7")
|
||||
assert "reusing the installed certificate bundle" in result.stdout
|
||||
unit = (target / "dashboards-agent.service").read_text()
|
||||
assert f"--cert-bundle {cert_path}" in unit, "an installed bundle keeps the unit in TLS mode"
|
||||
|
||||
|
||||
@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
|
||||
def test_uninstall_removes_the_bundle_with_the_token(tmp_path: Path) -> None:
|
||||
"""An uninstalled agent leaves no private key behind."""
|
||||
script, call_log, env = _sandbox_installer(tmp_path)
|
||||
target = tmp_path / "target"
|
||||
(target / "lib").mkdir(parents=True)
|
||||
leftovers = {
|
||||
target / "token": "tok\n",
|
||||
target / "certs": _fake_bundle_text(),
|
||||
target / "lib" / "host_agent.py": "# stub\n",
|
||||
target / "dashboards-agent.service": "[Unit]\n",
|
||||
}
|
||||
for path, body in leftovers.items():
|
||||
path.write_text(body, encoding="ascii")
|
||||
result = subprocess.run(
|
||||
[BASH, str(script), "--uninstall"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
timeout=60,
|
||||
check=False,
|
||||
)
|
||||
assert result.returncode == 0, f"uninstall failed:\n{result.stdout}\n{result.stderr}"
|
||||
for path in leftovers:
|
||||
assert not path.exists(), f"uninstall left {path} on disk"
|
||||
assert "systemctl disable --now dashboards-agent" in call_log.read_text()
|
||||
|
||||
|
||||
# --- the --status probe, end to end against the real TLS agent -----------------------------
|
||||
|
||||
|
||||
class DictMeta:
|
||||
"""The three-method `certs.MetaStore` slice, backed by a dict (the `test_auth.py` idiom)."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.rows: dict[str, str] = {}
|
||||
|
||||
def get_meta(self, key: str) -> str | None:
|
||||
return self.rows.get(key)
|
||||
|
||||
def set_meta(self, key: str, value: str) -> None:
|
||||
self.rows[key] = value
|
||||
|
||||
def delete_meta(self, key: str) -> int:
|
||||
return 1 if self.rows.pop(key, None) is not None else 0
|
||||
|
||||
|
||||
def _mint_bundle() -> str:
|
||||
"""A real bundle from the dashboard's own CA — the artifact the operator downloads."""
|
||||
store = DictMeta()
|
||||
state = certs.ensure_certs(store)
|
||||
state = certs.issue_agent_cert(store, state, "localhost", ["127.0.0.1", "localhost"])
|
||||
entry = state.agent("localhost")
|
||||
assert entry is not None
|
||||
return entry.bundle
|
||||
|
||||
|
||||
def _serve_tls_agent(host_agent: Any, bundle_text: str, where: Path) -> tuple[Any, int]:
|
||||
"""The real agent handler on a TLS-wrapped loopback port, exactly as the unit runs it."""
|
||||
path = where / "installed.certs"
|
||||
path.write_text(bundle_text, encoding="ascii")
|
||||
bundle, error = host_agent.TlsBundle.load(str(path))
|
||||
assert error is None and bundle is not None
|
||||
host_agent.Handler.collector = host_agent.Collector("/", smart=False)
|
||||
host_agent.Handler.token = None
|
||||
host_agent.Handler.bundle = bundle
|
||||
httpd = ThreadingHTTPServer(("127.0.0.1", 0), host_agent.Handler)
|
||||
httpd.socket = bundle.context.wrap_socket(httpd.socket, server_side=True)
|
||||
threading.Thread(target=httpd.serve_forever, daemon=True).start()
|
||||
return httpd, httpd.server_address[1]
|
||||
|
||||
|
||||
def _run_status(script: Path, port: int, env: dict[str, str]) -> str:
|
||||
result = subprocess.run(
|
||||
[BASH, str(script), "--status", "--bind", "127.0.0.1", "--port", str(port)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
timeout=60,
|
||||
check=False,
|
||||
)
|
||||
assert result.returncode == 0, f"--status exited {result.returncode}:\n{result.stderr}"
|
||||
return result.stdout
|
||||
|
||||
|
||||
@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
|
||||
def test_the_tls_status_probe_reads_health_with_the_ca_and_no_client_cert(
|
||||
tmp_path: Path, host_agent_module: Any
|
||||
) -> None:
|
||||
"""The task-02 ASSUMPTION, proven through the installer: CERT_OPTIONAL + the /health
|
||||
exemption let a probe that holds only the bundle's CA read /health on a TLS agent."""
|
||||
bundle = _mint_bundle()
|
||||
httpd, port = _serve_tls_agent(host_agent_module, bundle, tmp_path)
|
||||
try:
|
||||
script, _call_log, env = _sandbox_installer(tmp_path / "installer")
|
||||
target = tmp_path / "installer" / "target"
|
||||
target.mkdir(parents=True)
|
||||
(target / "dashboards-agent.service").write_text(
|
||||
f"ExecStart=python3 host_agent.py --cert-bundle {target / 'certs'}\n", encoding="ascii"
|
||||
)
|
||||
(target / "certs").write_text(bundle, encoding="ascii")
|
||||
assert _installer_status(_run_status(script, port, env)) == "True"
|
||||
finally:
|
||||
httpd.shutdown()
|
||||
httpd.server_close()
|
||||
host_agent_module.Handler.bundle = None
|
||||
|
||||
|
||||
@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
|
||||
def test_the_tls_status_probe_verifies_the_chain_and_says_no_to_another_authority(
|
||||
tmp_path: Path, host_agent_module: Any
|
||||
) -> None:
|
||||
"""The probe loads the CA for a reason: a server signed by a different root is not health."""
|
||||
httpd, port = _serve_tls_agent(host_agent_module, _mint_bundle(), tmp_path)
|
||||
try:
|
||||
script, _call_log, env = _sandbox_installer(tmp_path / "installer")
|
||||
target = tmp_path / "installer" / "target"
|
||||
target.mkdir(parents=True)
|
||||
(target / "dashboards-agent.service").write_text(
|
||||
f"ExecStart=python3 host_agent.py --cert-bundle {target / 'certs'}\n", encoding="ascii"
|
||||
)
|
||||
(target / "certs").write_text(_mint_bundle(), encoding="ascii") # a *different* CA
|
||||
stdout = _run_status(script, port, env)
|
||||
finally:
|
||||
httpd.shutdown()
|
||||
httpd.server_close()
|
||||
host_agent_module.Handler.bundle = None
|
||||
line = _installer_status(stdout)
|
||||
assert line != "True" and line.startswith("(no answer"), f"the probe must refuse: {line}"
|
||||
|
||||
|
||||
@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
|
||||
def test_the_tls_status_probe_names_the_bundle_when_it_cannot_read_it(tmp_path: Path) -> None:
|
||||
"""The bundle is 0640 root:agent — an unprivileged --status cannot read it, and says so."""
|
||||
script, _call_log, env = _sandbox_installer(tmp_path)
|
||||
target = tmp_path / "target"
|
||||
target.mkdir(parents=True)
|
||||
(target / "dashboards-agent.service").write_text(
|
||||
f"ExecStart=python3 host_agent.py --cert-bundle {target / 'certs'}\n", encoding="ascii"
|
||||
)
|
||||
cert = target / "certs"
|
||||
cert.write_text(_fake_bundle_text(), encoding="ascii")
|
||||
cert.chmod(0o000)
|
||||
line = _installer_status(_run_status(script, 1, env))
|
||||
assert "cannot read" in line and "run --status as root" in line, line
|
||||
|
||||
|
||||
@pytest.mark.skipif(shutil.which("curl") is None, reason="curl is not installed on this box")
|
||||
@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
|
||||
def test_a_token_mode_status_probe_stays_plain_http(tmp_path: Path) -> None:
|
||||
"""The other branch: no bundle in the unit → the unchanged plain probe, not https."""
|
||||
script, _call_log, env = _sandbox_installer(tmp_path)
|
||||
target = tmp_path / "target"
|
||||
target.mkdir(parents=True)
|
||||
(target / "dashboards-agent.service").write_text(
|
||||
"ExecStart=python3 host_agent.py --token-file /etc/dashboards-agent.token\n",
|
||||
encoding="ascii",
|
||||
)
|
||||
result = subprocess.run(
|
||||
[BASH, str(script), "--status", "--bind", "127.0.0.1", "--port", "1"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
timeout=60,
|
||||
check=False,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
assert _installer_status(result.stdout).startswith("(no answer"), (
|
||||
"token mode must probe plain http — nothing listens here, and the refusal is the point"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user