feat(gate): systemd socket activation + daemon client + protocol test + runbook updates

docs/RELEASES.md

@@ -48,6 +48,7 @@ We consider **Truth Plane v0.1** achieved when:
 - A compatibility alias remains for at least one minor cycle.
 
 Example:
+
 - `_nft_set_elements_json_from_obj` is a compatibility alias for `parse_nft_set_elements_json` and will be removed after v0.1.
 
 ---

docs/TRUTH_PLANE_RUNBOOK.md

@@ -3,6 +3,8 @@
 This runbook assumes the v0 tools exist in `tools/`:
 
 - `tools/sourceos_gate_egress.py`
+- `tools/sourceos_gate_egressd.py`
+- `tools/sourceos_gate_egressctl.py`
 - `tools/sourceos_truth_surface.py`
 - `tools/sourceos_delta_surface.py`
 - `tools/sourceos_incident.py`
@@ -60,7 +62,37 @@ sudo python tools/sourceos_truth_plane_smoke.py --store-root /tmp/sourceos-smoke
 
 ---
 
-## 0b) Tick orchestrator (periodic surfaces + delta)
+## 0b) Daemon mode (recommended for host operation)
+
+The egress gate can run as a host-local unix socket daemon.
+
+### Start the daemon (dev)
+
+```bash
+sudo python tools/sourceos_gate_egressd.py --store-root /var/lib/sourceos --socket /run/sourceos/gate-egress.sock
+```
+
+### Talk to the daemon
+
+```bash
+python tools/sourceos_gate_egressctl.py --socket /run/sourceos/gate-egress.sock health
+python tools/sourceos_gate_egressctl.py snapshot
+python tools/sourceos_gate_egressctl.py grant --token-id tok --nonce n1 --exp 9999999999 --proto tcp --target 1.2.3.4/32 --port 443 --apply
+python tools/sourceos_gate_egressctl.py verify
+```
+
+### systemd socket activation (packaging lane)
+
+If deployed via systemd socket activation, use:
+
+- `systemd/sourceos-gate-egress.socket`
+- `systemd/sourceos-gate-egress.service`
+
+The daemon will use `LISTEN_FDS` when present.
+
+---
+
+## 0c) Tick orchestrator (periodic surfaces + delta)
 
 The tick orchestrator is the unit of periodic work intended for systemd timers. It:
 
@@ -109,60 +141,13 @@ Then the gate can apply allowlist changes by mutating only the allow sets.
 
 ---
 
-## 3) Install a dry-run egress grant
-
-```bash
-# exp is epoch seconds
-python tools/sourceos_gate_egress.py grant --store-root /tmp/sourceos \
-  --token-id tok_demo --nonce n_0001 --exp 1893456000 \
-  --target 1.2.3.4/32 --port 443
-```
-
-Expected:
-
-- updates allowlist.state.json
-- rejects replay of the same token+nonce
-
----
-
-## 3a) Install and apply a TCP grant (requires root)
+## 3) CLI mode (direct)
 
 ```bash
+python tools/sourceos_gate_egress.py init --store-root /tmp/sourceos
 sudo python tools/sourceos_gate_egress.py grant --apply --proto tcp --store-root /tmp/sourceos \
   --token-id tok_demo --nonce n_0002 --exp 1893456000 \
   --target 1.2.3.4/32 --port 443
-```
-
-Expected:
-
-- updates allowlist.state.json
-- applies allowlist sets to nft (no ruleset flush)
-- writes an audit line under `/tmp/sourceos/audit/events/<date>/gate.egress.ndjson`
-
----
-
-## 3a-udp) Install and apply a UDP grant (DNS example)
-
-```bash
-sudo python tools/sourceos_gate_egress.py grant --apply --proto udp --store-root /tmp/sourceos \
-  --token-id tok_dns --nonce n_dns --exp 1893456000 \
-  --target 1.1.1.1/32 --port 53
-```
-
----
-
-## 3b) Prune expired grants (and optionally apply)
-
-```bash
-python tools/sourceos_gate_egress.py prune --store-root /tmp/sourceos
-sudo python tools/sourceos_gate_egress.py prune --apply --store-root /tmp/sourceos
-```
-
----
-
-## 3c) Verify nft allow sets match allowlist state
-
-```bash
 sudo python tools/sourceos_gate_egress.py verify --store-root /tmp/sourceos
 ```
 

src/sourceos_gate/daemon.py

@@ -4,15 +4,20 @@
 
 This is intended for host-local orchestration (systemd socket activation or
 static socket path), and is not exposed to the network.
+
+Security model:
+- Authentication is by filesystem permissions on the unix socket.
+- This daemon must not listen on TCP.
 """
 
 from __future__ import annotations
 
 import asyncio
 import json
+import os
+import socket as pysocket
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any
 
 from .egress import EgressGate
 from .protocol import err_response, map_error, ok_response, require_fields
@@ -26,7 +31,6 @@ class DaemonConfig:
 
 
 async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, gate: EgressGate) -> None:
-    peer = writer.get_extra_info("peername")
     try:
         while not reader.at_eof():
             line = await reader.readline()
@@ -89,15 +93,46 @@ async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWrit
             pass
 
 
+def _systemd_listen_fds() -> list[int]:
+    """Return systemd-passed listening fds, if present.
+
+    Systemd socket activation (sd_listen_fds) passes sockets starting at fd 3.
+    We implement the minimum required logic to avoid adding dependencies.
+    """
+
+    try:
+        listen_pid = int(os.environ.get("LISTEN_PID", "0"))
+        listen_fds = int(os.environ.get("LISTEN_FDS", "0"))
+    except ValueError:
+        return []
+
+    if listen_pid != os.getpid() or listen_fds <= 0:
+        return []
+
+    return list(range(3, 3 + listen_fds))
+
+
 async def serve(cfg: DaemonConfig) -> None:
+    gate = EgressGate.for_root(cfg.store_root)
+    gate.init()
+
+    fds = _systemd_listen_fds()
+    if fds:
+        # Prefer systemd socket activation. Use the first fd.
+        fd = fds[0]
+        sock = pysocket.socket(fileno=fd)
+        sock.setblocking(False)
+
+        server = await asyncio.start_unix_server(lambda r, w: handle_client(r, w, gate), sock=sock)
+        async with server:
+            await server.serve_forever()
+        return
+
+    # Fallback: create a unix socket at cfg.socket_path.
     cfg.socket_path.parent.mkdir(parents=True, exist_ok=True)
     if cfg.socket_path.exists():
         cfg.socket_path.unlink()
 
-    gate = EgressGate.for_root(cfg.store_root)
-    gate.init()
-
     server = await asyncio.start_unix_server(lambda r, w: handle_client(r, w, gate), path=str(cfg.socket_path))
-
     async with server:
         await server.serve_forever()

systemd/sourceos-gate-egress.service

@@ -0,0 +1,37 @@
+[Unit]
+Description=SourceOS egress gate daemon (Unix socket)
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+
+# Store root for state/audit.
+Environment=SOURCEOS_STORE_ROOT=/var/lib/sourceos
+
+# Create /run/sourceos for the socket path if needed.
+RuntimeDirectory=sourceos
+RuntimeDirectoryMode=0755
+
+# Run daemon; it will use systemd socket activation when available.
+ExecStart=/usr/bin/env python3 /usr/lib/sourceos/tools/sourceos_gate_egressd.py --store-root ${SOURCEOS_STORE_ROOT}
+
+Restart=on-failure
+RestartSec=2
+
+# Hardening
+NoNewPrivileges=yes
+PrivateTmp=yes
+ProtectSystem=strict
+ProtectHome=yes
+ReadWritePaths=/var/lib/sourceos /run/sourceos
+
+# The daemon must not listen on TCP; it only needs unix sockets + netlink (nft).
+RestrictAddressFamilies=AF_UNIX AF_NETLINK
+
+# Needs nft mutation when applying; run as root (socket perms gate callers).
+User=root
+Group=root
+
+[Install]
+WantedBy=multi-user.target

systemd/sourceos-gate-egress.socket

@@ -0,0 +1,12 @@
+[Unit]
+Description=SourceOS egress gate socket (Unix)
+
+[Socket]
+ListenStream=/run/sourceos/gate-egress.sock
+SocketMode=0660
+SocketUser=root
+SocketGroup=sourceos
+RemoveOnStop=true
+
+[Install]
+WantedBy=sockets.target

tests/test_gate_daemon.py

@@ -0,0 +1,59 @@
+import asyncio
+import json
+import tempfile
+import unittest
+from pathlib import Path
+import sys
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+SRC_DIR = REPO_ROOT / "src"
+if str(SRC_DIR) not in sys.path:
+    sys.path.insert(0, str(SRC_DIR))
+
+from sourceos_gate.daemon import DaemonConfig, serve
+
+
+async def _send(sock_path: str, msg: dict) -> dict:
+    reader, writer = await asyncio.open_unix_connection(sock_path)
+    writer.write((json.dumps(msg) + "\n").encode("utf-8"))
+    await writer.drain()
+    line = await reader.readline()
+    writer.close()
+    await writer.wait_closed()
+    return json.loads(line.decode("utf-8"))
+
+
+class GateDaemonTests(unittest.TestCase):
+    def test_health_and_snapshot(self):
+        async def run():
+            with tempfile.TemporaryDirectory() as td:
+                root = Path(td)
+                sock = root / "gate.sock"
+
+                cfg = DaemonConfig(socket_path=sock, store_root=root)
+                task = asyncio.create_task(serve(cfg))
+
+                # wait for socket
+                for _ in range(50):
+                    if sock.exists():
+                        break
+                    await asyncio.sleep(0.02)
+
+                resp = await _send(str(sock), {"id": "1", "method": "health", "params": {}})
+                self.assertTrue(resp.get("ok"))
+
+                snap = await _send(str(sock), {"id": "2", "method": "snapshot", "params": {}})
+                self.assertTrue(snap.get("ok"))
+                self.assertIn("active", snap.get("result", {}))
+
+                task.cancel()
+                try:
+                    await task
+                except asyncio.CancelledError:
+                    pass
+
+        asyncio.run(run())
+
+
+if __name__ == "__main__":
+    unittest.main()