The agent host's vault and SilverBullet (M5.4)
deploy/agent-host sets up the vault on nisse-host (infra/tofu#77) and SilverBullet over it: the operator writes the vault, the agents' memory/ is read-only to it, and only the operator's Caddy reaches it, which serves vault.dongar.net to the LAN. scripts/agent-host.sh installs and checks it from here. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk
This commit is contained in:
1 parent
da9097a4dd
commit
eafe1b9f4a
8 files changed
+342
-7
No files matched your search
@@ -0,0 +1,117 @@
|
||||
"""The agent host's vault and SilverBullet (deploy/agent-host; L10, M5.4):
|
||||
the operator writes the vault, never the agents' memory; nothing but the
|
||||
vault and SilverBullet's own data is mounted; the image changes only in a
|
||||
PR; only the operator's Caddy reaches it."""
|
||||
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
|
||||
HOST = Path(__file__).resolve().parents[2] / "deploy" / "agent-host"
|
||||
VAULT = "/srv/nisse/vault"
|
||||
|
||||
|
||||
def silverbullet() -> dict[str, Any]:
|
||||
compose = yaml.safe_load((HOST / "silverbullet" / "compose.yaml").read_text())
|
||||
return compose["services"]["silverbullet"]
|
||||
|
||||
|
||||
def mounts() -> list[tuple[str, str, str]]:
|
||||
out = []
|
||||
for volume in silverbullet()["volumes"]:
|
||||
source, target, *mode = volume.split(":")
|
||||
out.append((source, target, mode[0] if mode else "rw"))
|
||||
return out
|
||||
|
||||
|
||||
def setting(name: str) -> str:
|
||||
match = re.search(rf"^{name}=(\S+)", (HOST / "setup.sh").read_text(), re.MULTILINE)
|
||||
assert match, name
|
||||
return match.group(1)
|
||||
|
||||
|
||||
def test_the_agents_memory_is_read_only_and_the_rest_of_the_vault_is_not() -> None:
|
||||
by_source = {source: (target, mode) for source, target, mode in mounts()}
|
||||
vault_target, vault_mode = by_source[VAULT]
|
||||
assert vault_mode == "rw"
|
||||
# memory/ is mounted over the same place in the space, read-only.
|
||||
assert by_source[f"{VAULT}/memory"] == (f"{vault_target}/memory", "ro")
|
||||
# Nothing else of the vault is mounted, so nothing gets around that.
|
||||
others = [s for s in by_source if s.startswith(VAULT)]
|
||||
assert sorted(others) == [VAULT, f"{VAULT}/memory"]
|
||||
|
||||
|
||||
def test_it_sees_nothing_of_the_host_but_the_vault_and_its_own_data() -> None:
|
||||
assert sorted(source for source, _, _ in mounts()) == [
|
||||
"/srv/nisse/silverbullet",
|
||||
VAULT,
|
||||
f"{VAULT}/memory",
|
||||
]
|
||||
|
||||
|
||||
def test_the_image_is_pinned_to_a_release() -> None:
|
||||
# The docker role updates images nightly; a pinned release changes only here.
|
||||
image = silverbullet()["image"]
|
||||
assert re.fullmatch(
|
||||
r"ghcr\.io/silverbulletmd/silverbullet:\d+\.\d+\.\d+(-slim)?", image
|
||||
)
|
||||
|
||||
|
||||
def test_no_secrets_in_the_compose_file() -> None:
|
||||
environment = silverbullet().get("environment", {})
|
||||
assert not [k for k in environment if re.search(r"USER|TOKEN|PASS|SECRET|KEY", k)]
|
||||
|
||||
|
||||
def test_its_port_caddys_and_the_firewalls_agree() -> None:
|
||||
port = setting("SB_PORT")
|
||||
assert silverbullet()["ports"] == [f"0.0.0.0:{port}:3000"] # IPv4 only
|
||||
caddy = (HOST / "caddy" / "vault.dongar.net.caddy").read_text()
|
||||
assert f"reverse_proxy nisse-host.garrett-hall.com:{port}" in caddy
|
||||
assert "@outside not remote_ip private_ranges" in caddy # the LAN only
|
||||
|
||||
|
||||
def test_logs_go_to_the_hosts_journal() -> None:
|
||||
assert silverbullet()["logging"] == {"driver": "journald"}
|
||||
|
||||
|
||||
def _can_use_a_network_namespace() -> bool:
|
||||
if not (shutil.which("unshare") and shutil.which("iptables")):
|
||||
return False
|
||||
probe = subprocess.run(
|
||||
["unshare", "-n", "iptables", "-S"], capture_output=True, check=False
|
||||
)
|
||||
return probe.returncode == 0
|
||||
|
||||
|
||||
@pytest.mark.skipif(
|
||||
not _can_use_a_network_namespace(),
|
||||
reason="needs root, unshare and iptables to try the rule in a namespace",
|
||||
)
|
||||
def test_the_firewall_lets_only_one_address_in_and_replaces_its_own_rule() -> None:
|
||||
script = HOST / "nisse-firewall"
|
||||
steps = (
|
||||
"iptables -N DOCKER-USER; iptables -A DOCKER-USER -j RETURN; "
|
||||
f"{script} 192.168.6.13 3000; {script} 192.168.6.13 3000; "
|
||||
f"{script} 192.168.6.14 3000; iptables -S DOCKER-USER"
|
||||
)
|
||||
run = subprocess.run(
|
||||
["unshare", "-n", "sh", "-ec", steps],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
rules = [r for r in run.stdout.splitlines() if "nisse-firewall" in r]
|
||||
assert len(rules) == 1, run.stdout # run again: still one; changed: replaced
|
||||
[rule] = rules
|
||||
assert "! -s 192.168.6.14/32" in rule
|
||||
assert "--ctorigdstport 3000" in rule
|
||||
assert rule.endswith("-j DROP")
|
||||
# Ahead of Docker's own return, so it's checked first.
|
||||
assert run.stdout.splitlines().index(rule) < run.stdout.splitlines().index(
|
||||
"-A DOCKER-USER -j RETURN"
|
||||
)
|
||||
Reference in new issue
Block a user