server/server-config
3
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Fix flake update (#11)
diff-tag / dix (push) Successful in 6m33s
Details
flake-check / flake-check (push) Successful in 59s
Details
lint / lint (push) Successful in 1m47s
Details

Browse Source 
Co-authored-by: flowey <floweydev@gmail.com>
Co-authored-by: gitea-actions[bot] <gitea-actions@noreply.local>
Reviewed-on: #11
This commit was merged in pull request #11.
This commit is contained in:
flowey
2026-09-22 21:44:25 +00:00
co-authored by flowey gitea-actions[bot]
parent 8551b8ad90
commit d9b4bab1f8
10 changed files with 329 additions and 296 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/server-n0/integration.py
+146 -107
View File
@@ -29,12 +29,15 @@ if TYPE_CHECKING:
KIND: dict[str, str] = {}
DEVICES: dict[str, list[str]] = {}
EDGES: list[tuple[str, str, list[int], list[int]]] = []
EXPOSED_TCP: dict[str, list[int]] = {}
HOST_ACCESS_TCP: dict[str, list[int]] = {}
FORWARD_TCP: list[tuple[int, str, int]] = []
NGINX_BACKENDS: dict[str, int] = {}
NGINX_DOMAINS: dict[str, str] = {}
SSH_KEY: str = ""
SSH_OPTS: str = ""
TEST_SSH_KEY_STORE: str = ""
SSHPIPER_TEST_USER: str = ""
HOST_EXPORTER_PORT: int = 0
HOST_CADVISOR_PORT: int = 0
HOST_SSHD_PORT: int = 0
SW = "/run/current-system/sw/bin"
@@ -42,6 +45,12 @@ SH = f"{SW}/sh"
PEER_EDGES = {(src, to) for src, to, _, _ in EDGES if to not in ("@host", "@internet")}
# pre-compute the set of all host-access ports (union across containers)
ALL_HOST_ACCESS_PORTS = {p for ps in HOST_ACCESS_TCP.values() for p in ps}
# published host ports
FWD_HOST_PORTS = {hp for hp, _, _ in FORWARD_TCP}
def failed_units(node_arg=""):
out = server.succeed(f"systemctl {node_arg} list-units --state=failed --no-legend --plain || true")
@@ -144,49 +153,38 @@ with subtest("(isolation): internet egress follows the access list"):
server.fail(net_cmd(name, f"{IP} route show default | grep -q ."))
with subtest("(functional): container DNS still resolves"):
src = enterable[0]
dst = next(c for c in included if c != src)
resolved = server.succeed(
net_cmd("git", f"{DIG} +short +time=3 @{HOST_ADDR['git']} nginx.containers")
net_cmd(src, f"{DIG} +short +time=3 @{HOST_ADDR[src]} {dst}.containers")
).strip()
assert resolved == ADDR["nginx"], f"nginx.containers resolved to {resolved!r}, expected {ADDR['nginx']}"
assert resolved == ADDR[dst], f"{dst}.containers resolved to {resolved!r}, expected {ADDR[dst]}"
with subtest("(functional): web services respond"):
for name, port in [("devenv", 8080), ("maven", 8080), ("git", 3000)]:
server.succeed(net_cmd(name, f"{CURL} -sf --max-time 5 -o /dev/null http://{ADDR[name]}:{port}/ >&2"))
# TODO: re-enable mail
"""
with subtest("(functional): mail accepts connections"):
smtp_probe = net_cmd("mail", f"{SH} -c 'exec 3<>/dev/tcp/{ADDR['mail']}/25 && head -1 <&3'")
server.wait_until_succeeds(f"{smtp_probe} | grep -q '^220'", timeout=60)
imap_probe = net_cmd(
"mail",
f"{TIMEOUT} 3 {OPENSSL} s_client -connect {ADDR['mail']}:993 -quiet -verify_quiet 2>/dev/null",
)
def imap_ready(_):
greeting = server.execute(f"echo | {imap_probe} | head -1")[1]
return greeting.strip().startswith("* OK")
retry(imap_ready, timeout_seconds=60)
"""
for name, port in NGINX_BACKENDS.items():
if name in enterable:
server.succeed(net_cmd(name, f"{CURL} -sf --max-time 5 -o /dev/null http://{ADDR[name]}:{port}/ >&2"))
with subtest("(functional): nginx proxies to backend containers"):
server.succeed(
net_cmd(
"nginx",
f"{CURL} -sfk --max-time 5 --resolve git.floweytf.com:443:{ADDR['nginx']} "
"https://git.floweytf.com/ >&2",
)
)
if "nginx" in enterable:
for domain, backend in NGINX_DOMAINS.items():
if backend in included:
server.succeed(
net_cmd(
"nginx",
f"{CURL} -sfk --max-time 5 --resolve {domain}:443:{ADDR['nginx']} "
f"https://{domain}/ >&2",
)
)
with subtest("(functional): sshpiper authenticates/routes properly"):
whoami = server.succeed(f"{SSH} {SSH_OPTS} -p 2222 {SSHPIPER_TEST_USER}@{ADDR['sshpiper']} whoami").strip()
assert whoami == "runner", f"expected to land as runner, got {whoami}"
hostname = server.succeed(
f"{SSH} {SSH_OPTS} -p 2222 {SSHPIPER_TEST_USER}@{ADDR['sshpiper']} hostname"
).strip()
assert hostname == "ssh-admin", f"expected to land on ssh-admin, got {hostname}"
if "sshpiper" in included:
whoami = server.succeed(f"{SSH} {SSH_OPTS} -p 2222 {SSHPIPER_TEST_USER}@{ADDR['sshpiper']} whoami").strip()
assert whoami == "runner", f"expected to land as runner, got {whoami}"
hostname = server.succeed(
f"{SSH} {SSH_OPTS} -p 2222 {SSHPIPER_TEST_USER}@{ADDR['sshpiper']} hostname"
).strip()
assert hostname == "ssh-admin", f"expected to land on ssh-admin, got {hostname}"
with subtest("(isolation): every declared container edge connects"):
for src, to, tcp_ports, _ in EDGES:
@@ -196,89 +194,136 @@ with subtest("(isolation): every declared container edge connects"):
server.succeed(ct_connect(src, ADDR[to], port))
with subtest("(isolation): undeclared container edges are refused"):
for src, dst, port in [
("devenv", "git", 3000),
("git", "devenv", 8080),
("maven", "git", 3000),
# ("nginx", "mail", 25),
("ssh-botm", "ssh-admin", 22),
("overseer", "git", 3000),
]:
assert (src, dst) not in PEER_EDGES, f"{src} -> {dst} is a declared edge, not a negative case"
server.fail(ct_connect(src, ADDR[dst], port, timeout=2))
tested = 0
for src in enterable:
for dst in enterable:
if src == dst or (src, dst) in PEER_EDGES:
continue
ports = EXPOSED_TCP.get(dst, [])
if not ports:
continue
server.fail(ct_connect(src, ADDR[dst], ports[0], timeout=2))
tested += 1
assert tested > 0, "no negative edge cases to test"
with subtest("(isolation): edges are one-way"):
assert ("nginx", "git") in PEER_EDGES
assert ("git", "nginx") not in PEER_EDGES
server.succeed(ct_connect("nginx", ADDR["git"], 3000))
server.fail(ct_connect("git", ADDR["nginx"], 443, timeout=2))
tested = 0
for src, dst, tcp_ports, _ in EDGES:
if dst in ("@host", "@internet") or src not in enterable or dst not in enterable:
continue
if (dst, src) in PEER_EDGES or not tcp_ports:
continue
server.succeed(ct_connect(src, ADDR[dst], tcp_ports[0]))
tested += 1
rev_ports = EXPOSED_TCP.get(src, [])
if rev_ports:
server.fail(ct_connect(dst, ADDR[src], rev_ports[0], timeout=2))
assert tested > 0, "no one-way edge cases to test"
with subtest("(isolation): host access allowlist"):
server.succeed(ct_connect("overseer", HOST_ADDR["overseer"], HOST_EXPORTER_PORT))
for port in (HOST_EXPORTER_PORT, HOST_CADVISOR_PORT):
server.fail(ct_connect("git", HOST_ADDR["git"], port, timeout=2))
server.fail(ct_connect("git", HOST_ADDR["overseer"], port, timeout=2))
for name, ports in HOST_ACCESS_TCP.items():
if name in enterable:
for port in ports:
server.succeed(ct_connect(name, HOST_ADDR[name], port))
no_access = [c for c in enterable if c not in HOST_ACCESS_TCP]
if no_access and ALL_HOST_ACCESS_PORTS:
ct = no_access[0]
for port in ALL_HOST_ACCESS_PORTS:
server.fail(ct_connect(ct, HOST_ADDR[ct], port, timeout=2))
# reaching the port through a different container's host addr is also blocked
for other in HOST_ACCESS_TCP:
if other in included:
server.fail(ct_connect(ct, HOST_ADDR[other], port, timeout=2))
with subtest("(isolation): host services outside the access list"):
internet.succeed(connect(server_ip, HOST_SSHD_PORT))
for name in ("git", "overseer"):
server.fail(ct_connect(name, HOST_ADDR[name], HOST_SSHD_PORT, timeout=2))
server.fail(ct_connect(name, server_ip, HOST_SSHD_PORT, timeout=2))
for name in enterable:
if HOST_SSHD_PORT not in HOST_ACCESS_TCP.get(name, []):
server.fail(ct_connect(name, HOST_ADDR[name], HOST_SSHD_PORT, timeout=2))
server.fail(ct_connect(name, server_ip, HOST_SSHD_PORT, timeout=2))
with subtest("(isolation): source address spoofing is dropped"):
src = enterable[0]
other = enterable[1]
before = spoof_counter()
server.succeed(net_cmd("git", f"{PING} -c 1 -W 2 {HOST_ADDR['git']} >&2"))
server.succeed(net_cmd(src, f"{PING} -c 1 -W 2 {HOST_ADDR[src]} >&2"))
assert spoof_counter() == before, "an unspoofed packet hit the source guard"
iface = IFACE["git"]
server.succeed(net_cmd("git", f"{IP} addr add {ADDR['nginx']}/32 dev {iface}"))
iface = IFACE[src]
server.succeed(net_cmd(src, f"{IP} addr add {ADDR[other]}/32 dev {iface}"))
try:
server.execute(
net_cmd("git", f"{PING} -c 2 -W 2 -I {ADDR['nginx']} {HOST_ADDR['git']}")
net_cmd(src, f"{PING} -c 2 -W 2 -I {ADDR[other]} {HOST_ADDR[src]}")
)
assert spoof_counter() > before, "a forged source address was not dropped"
finally:
server.succeed(net_cmd("git", f"{IP} addr del {ADDR['nginx']}/32 dev {iface}"))
server.succeed(net_cmd(src, f"{IP} addr del {ADDR[other]}/32 dev {iface}"))
with subtest("(isolation): renaming the interface changes nothing"):
iface, host_addr = IFACE["nginx"], HOST_ADDR["nginx"]
# find a container with a declared peer edge
rename_src = rename_dst = rename_port = None
for src, dst, tcp_ports, _ in EDGES:
if dst in ("@host", "@internet") or src not in enterable or dst not in included:
continue
if tcp_ports:
rename_src, rename_dst, rename_port = src, dst, tcp_ports[0]
break
assert rename_src is not None and rename_dst is not None and rename_port is not None, \
"no peer edge to test interface rename"
iface, host_addr = IFACE[rename_src], HOST_ADDR[rename_src]
def rename(frm, to):
server.succeed(net_cmd("nginx", f"{IP} link set dev {frm} down"))
server.succeed(net_cmd("nginx", f"{IP} link set dev {frm} name {to}"))
server.succeed(net_cmd("nginx", f"{IP} link set dev {to} up"))
server.succeed(net_cmd(rename_src, f"{IP} link set dev {frm} down"))
server.succeed(net_cmd(rename_src, f"{IP} link set dev {frm} name {to}"))
server.succeed(net_cmd(rename_src, f"{IP} link set dev {to} up"))
for route in (host_addr, f"10.0.0.0/8 via {host_addr}", f"default via {host_addr}"):
server.succeed(net_cmd("nginx", f"{IP} route replace {route} dev {to}"))
server.succeed(net_cmd(rename_src, f"{IP} route replace {route} dev {to}"))
rename(iface, "evil0")
try:
server.succeed(ct_connect("nginx", ADDR["git"], 3000))
# server.fail(ct_connect("nginx", ADDR["mail"], 25, timeout=2))
server.succeed(ct_connect(rename_src, ADDR[rename_dst], rename_port))
finally:
rename("evil0", iface)
with subtest("(isolation): the veth cannot be moved out of the namespace"):
server.fail(net_cmd("git", f"{IP} link set dev {IFACE['git']} netns 1"))
ct = enterable[0]
server.fail(net_cmd(ct, f"{IP} link set dev {IFACE[ct]} netns 1"))
with subtest("(isolation): route bypass"):
iface = IFACE["hopper"]
bypass_ct = next((c for c in enterable if c in no_gateway), None)
if bypass_ct is None:
bypass_ct = next((c for c in enterable if c not in allow_internet), None)
assert bypass_ct is not None, "no container without internet to test route bypass"
iface = IFACE[bypass_ct]
server.succeed(
net_cmd("hopper", f"{IP} route replace default via {HOST_ADDR['hopper']} dev {iface} onlink")
net_cmd(bypass_ct, f"{IP} route replace default via {HOST_ADDR[bypass_ct]} dev {iface} onlink")
)
server.fail(net_cmd("hopper", f"{CURL} -sf --max-time 5 http://{internet_ip}/ >&2"))
server.fail(ct_connect("hopper", ADDR["git"], 3000, timeout=2))
server.fail(net_cmd(bypass_ct, f"{CURL} -sf --max-time 5 http://{internet_ip}/ >&2"))
unreachable = next(
(c for c in included if c != bypass_ct and (bypass_ct, c) not in PEER_EDGES and EXPOSED_TCP.get(c, [])),
None,
)
if unreachable:
server.fail(ct_connect(bypass_ct, ADDR[unreachable], EXPOSED_TCP[unreachable][0], timeout=2))
with subtest("(isolation): no IPv6 link-local addresses"):
for name in ["git", "overseer"]:
for name in enterable:
assert not has_ipv6_link_local(name), f"{name} still has an IPv6 link-local address"
with subtest("(isolation): published ports grant no reach between containers"):
for port in (80, 443, 25565, 22, 25):
server.fail(ct_connect("overseer", HOST_ADDR["overseer"], port, timeout=2))
server.fail(ct_connect("git", server_ip, port, timeout=2))
if FWD_HOST_PORTS:
ct = next((c for c in enterable if c not in HOST_ACCESS_TCP), enterable[0])
for port in FWD_HOST_PORTS:
server.fail(ct_connect(ct, HOST_ADDR[ct], port, timeout=2))
server.fail(ct_connect(ct, server_ip, next(iter(FWD_HOST_PORTS)), timeout=2))
with subtest("(isolation): privilege escalation / container escape"):
leader = container_pid("git")
ct = enterable[0]
leader = container_pid(ct)
host_uid = server.succeed(f"awk '/^Uid:/{{print $2}}' /proc/{leader}/status").strip()
assert host_uid != "0", "container's PID 1 is not idmapped away from host uid 0"
@@ -287,28 +332,16 @@ with subtest("(isolation): privilege escalation / container escape"):
container_pidns = server.succeed(f"readlink /proc/{leader}/ns/pid").strip()
assert host_pidns != container_pidns, "container shares the host's PID namespace"
server.succeed(
f"{NSENTER} --mount=/proc/{leader}/ns/mnt --root=/proc/{leader}/root -- "
f"{SH} -c 'echo x >> /var/lib/gitea/canary' >&2"
)
sshpiper_pid = container_pid("sshpiper")
server.fail(
f"{NSENTER} --mount=/proc/{sshpiper_pid}/ns/mnt --root=/proc/{sshpiper_pid}/root -- "
f"{SH} -c 'echo x >> /etc/ssh/ssh_host_ed25519_key' >&2"
)
server.fail(
f"{NSENTER} --mount=/proc/{leader}/ns/mnt --root=/proc/{leader}/root -- "
f"{SH} -c 'touch /nix/store/canary' >&2"
)
server.fail(ct_cmd(ct, "touch /nix/store/canary"))
server.fail(
f"{NSENTER} --mount=/proc/{leader}/ns/mnt --root=/proc/{leader}/root -- "
"/run/current-system/sw/bin/test -d /persist >&2"
)
with subtest("(functional): external clients reach published ports"):
for port in (80, 25565, 22):
internet.succeed(ext_connect(port))
for hp, _, _ in FORWARD_TCP:
internet.succeed(ext_connect(hp))
# a non-published port should be unreachable
internet.fail(ext_connect(8080))
with subtest("(functional): no failing-reason restarts"):
@@ -317,21 +350,27 @@ with subtest("(functional): no failing-reason restarts"):
assert (
failed_units(f"-M {name}") == []
), f"{name} has failed units: {failed_units(f'-M {name}')}"
restarts = server.succeed(f"systemctl show container@{name}.service -p NRestarts --value").strip()
assert restarts == "0", f"container@{name}.service restarted {restarts} times"
unit = "container" if KIND[name] == "nspawn" else "microvm"
restarts = server.succeed(f"systemctl show {unit}@{name}.service -p NRestarts --value").strip()
assert restarts == "0", f"{unit}@{name}.service restarted {restarts} times"
with subtest("(functional): port forwards survive an nftables reload"):
server.succeed("systemctl reload nftables.service")
elements = server.succeed("nft list map ip container-portforward tcp-fwd")
assert re.search(r"\b80\s*:", elements), f"nftables reload emptied the map: {elements!r}"
internet.wait_until_succeeds(ext_connect(80), timeout=30)
if FORWARD_TCP:
hp, ct_name, _ = FORWARD_TCP[0]
server.succeed("systemctl reload nftables.service")
elements = server.succeed("nft list map ip container-portforward tcp-fwd")
assert re.search(rf"\b{hp}\s*:", elements), f"nftables reload emptied the map: {elements!r}"
internet.wait_until_succeeds(ext_connect(hp), timeout=30)
with subtest("(functional): published ports follow the container's lifetime"):
server.succeed("systemctl stop container@nginx")
internet.fail(ext_connect(80))
server.succeed("systemctl start container@nginx")
server.wait_for_unit("container@nginx.service")
internet.wait_until_succeeds(ext_connect(80), timeout=30)
if FORWARD_TCP:
hp, ct_name, _ = FORWARD_TCP[0]
unit = "container" if KIND[ct_name] == "nspawn" else "microvm"
server.succeed(f"systemctl stop {unit}@{ct_name}")
internet.fail(ext_connect(hp))
server.succeed(f"systemctl start {unit}@{ct_name}")
server.wait_for_unit(f"{unit}@{ct_name}.service")
internet.wait_until_succeeds(ext_connect(hp), timeout=30)
with subtest("(functional): declared devices are usable in-container"):
for name, devices in DEVICES.items():