import importlib import os import json import queue import socket import tempfile import threading import time import unittest from types import SimpleNamespace from unittest import mock from pathlib import Path from agent_guard.events import Event class SidecarTests(unittest.TestCase): @staticmethod def _start_socket_server(mod, path, snapshot, work_queue): stop = threading.Event() thread = threading.Thread( target=mod.socket_server, args=(path, work_queue, stop, snapshot), daemon=True, ) thread.start() for _ in range(100): if path.exists(): break time.sleep(0.01) return stop, thread @staticmethod def _send(path, payload, read_reply=True): with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client: client.connect(str(path)) client.sendall(payload) client.shutdown(socket.SHUT_WR) if not read_reply: return None return json.loads(client.recv(65536).decode()) def test_socket_poll_returns_handled_agent_guard_events_and_culprits(self): mod = importlib.import_module("agent_guard.daemon") with tempfile.TemporaryDirectory() as td: path = Path(td) / "agent-guard.sock" snapshot = mod.PollSnapshot() snapshot.record(Event( "warning", "cpu_runaway", "nonbuild CPU hog detected", culprit={"name": "cursor-agent", "rss_kb": 512000}, remediation="kill", )) stop, thread = self._start_socket_server(mod, path, snapshot, queue.Queue()) self.assertTrue(path.exists()) response = self._send(path, b'{"type":"poll"}\n') stop.set() thread.join(timeout=2) self.assertEqual(response["events"], [{ "tier": "warning", "source": "cpu_runaway", "reason": "nonbuild CPU hog detected", "remediation": "kill", "culprit": {"name": "cursor-agent", "rss_kb": 512000}, }]) self.assertEqual(response["culprits"], [{"name": "cursor-agent", "rss_kb": 512000}]) def test_socket_server_survives_hostile_clients_and_keeps_accepting(self): mod = importlib.import_module("agent_guard.daemon") work_queue = queue.Queue() with tempfile.TemporaryDirectory() as td: path = Path(td) / "agent-guard.sock" snapshot = mod.PollSnapshot() stop, thread = self._start_socket_server(mod, path, snapshot, work_queue) self.assertTrue(path.exists()) with mock.patch.object(mod, "journal_send") as journal: self._send(path, b'[1,2]\n', read_reply=False) self._send(path, b'not json\n', read_reply=False) self._send(path, b'{"type":"poll"}\n', read_reply=False) response = self._send(path, b'{"type":"poll","version":2}\n') self._send( path, b'{"type":"event","event":{"tier":"warning","source":"s","reason":"r"}}\n', read_reply=False, ) event = work_queue.get(timeout=2) stop.set() thread.join(timeout=2) self.assertEqual(response, {"events": [], "culprits": []}) self.assertEqual((event.tier, event.source, event.reason), ("warning", "s", "r")) actions = [call.kwargs["SM_ACTION"] for call in journal.call_args_list] self.assertEqual([a for a in actions if a == "bad_ipc"], ["bad_ipc", "bad_ipc"]) self.assertLessEqual(set(actions), {"bad_ipc", "ipc_transport"}) def test_disk_alarm_does_not_offer_tmp_cleanup(self): mod = importlib.import_module("agent_guard.cli") with mock.patch.object(mod, "send_event", return_value=0) as send: self.assertEqual( mod.main(["alarm", "--name", "disk_capacity", "--status", "CRITICAL", "--value", "91"]), 0, ) event = send.call_args.args[0] self.assertEqual(event.remediation, "none") self.assertNotIn("Clean /tmp (safe)", event.actions) def test_journald_matcher_flags_kernel_oom(self): mod = importlib.import_module("agent_guard.events.journald") event = mod.match_line("kernel: Out of memory: Killed process 1234 (python3)") self.assertEqual(event.tier, "critical") self.assertEqual(event.source, "kernel") self.assertIn("Out of memory", event.reason) def test_ports_diff_ignores_dev_allowlist_and_flags_other_listener(self): mod = importlib.import_module("agent_guard.events.ports") baseline = {("tcp", "127.0.0.1", 5173)} current = {("tcp", "127.0.0.1", 5173), ("tcp", "0.0.0.0", 2222)} events = mod.diff_listeners(baseline, current, ["3000-9999", "5173"]) self.assertEqual(len(events), 1) self.assertIn("2222", events[0].reason) def test_ports_diff_skips_tailscale_host_listener(self): mod = importlib.import_module("agent_guard.events.ports") cidrs = mod.parse_host_cidrs(["100.64.0.0/10", "fd7a:115c:a1e0::/48"]) baseline = set() current = { ("tcp", "100.126.128.50", 54659), ("tcp6", "fd7a:115c:a1e0::b3a:8033", 49161), ("tcp", "0.0.0.0", 2222), } events = mod.diff_listeners(baseline, current, [], cidrs) self.assertEqual(len(events), 1) self.assertIn("0.0.0.0:2222", events[0].reason) def test_ports_diff_skips_new_loopback_listener(self): # A new listener on loopback (dev servers, LSPs, AI agents) is benign and # must NOT desktop-notify; only non-loopback binds are the security signal. mod = importlib.import_module("agent_guard.events.ports") baseline = set() current = {("tcp", "127.0.0.1", 8888), ("tcp6", "::1", 9999), ("tcp", "0.0.0.0", 2222)} events = mod.diff_listeners(baseline, current, []) self.assertEqual(len(events), 1) self.assertIn("0.0.0.0:2222", events[0].reason) def test_state_save_survives_concurrent_writers(self): # 6 daemon threads call save() concurrently; a shared tmp name races and # crashes the daemon. Atomic per-writer tmp + lock must survive the storm. import threading as _t mod = importlib.import_module("agent_guard.state") with tempfile.TemporaryDirectory() as td: st = mod.State(Path(td) / "agent-guard.json") errors = [] def hammer(i): try: for j in range(50): self.assertIn(st.allow_now(f"k{i}-{j}", 0), (True, False)) except Exception as exc: # noqa: BLE001 - capture race crash errors.append(exc) threads = [_t.Thread(target=hammer, args=(i,)) for i in range(6)] for t in threads: t.start() for t in threads: t.join() self.assertEqual(errors, []) self.assertTrue((Path(td) / "agent-guard.json").exists()) def test_culprit_picks_fastest_growing_non_protected_group(self): mod = importlib.import_module("agent_guard.culprit") before = {"python3": 100, "Xorg": 50} after = {"python3": 240, "Xorg": 500} culprit = mod.pick_fastest_grower(before, after, protect={"Xorg"}) self.assertEqual(culprit["name"], "python3") self.assertEqual(culprit["growth_kb"], 140) def test_kill_guard_refuses_protected_and_start_time_mismatch(self): mod = importlib.import_module("agent_guard.culprit") proc = { 100: {"exe": "systemd", "start_time": "10", "cgroup": "0::/a"}, 200: {"exe": "python3", "start_time": "11", "cgroup": "0::/b"}, 201: {"exe": "python3", "start_time": "12", "cgroup": "0::/b"}, } target = {"name": "python3", "pids": [{"pid": 200, "start_time": "old", "cgroup": "0::/b"}, {"pid": 201, "start_time": "12", "cgroup": "0::/b"}]} self.assertEqual(mod.safe_signal_plan(target, proc, protect={"systemd", "bash"}), [201]) protected = {"name": "systemd", "pids": [{"pid": 100, "start_time": "10", "cgroup": "0::/a"}]} self.assertEqual(mod.safe_signal_plan(protected, proc, protect={"systemd"}), []) def test_safe_tmp_clean_only_removes_owned_old_files(self): mod = importlib.import_module("agent_guard.notifier") with tempfile.TemporaryDirectory() as td: root = Path(td) old_owned = root / "old" old_owned.write_text("x") old = time.time() - 80 * 3600 os.utime(old_owned, (old, old)) fresh = root / "fresh" fresh.write_text("x") removed = mod.clean_tmp_safe(root, uid=os.getuid(), older_than_seconds=72 * 3600) self.assertEqual(removed, 1) self.assertFalse(old_owned.exists()) self.assertTrue(fresh.exists()) def test_gdbus_notify_passes_replaces_id(self): mod = importlib.import_module("agent_guard.notifier") proc = SimpleNamespace(stdout="(uint32 77,)") with mock.patch("agent_guard.notifier.gate_allows", return_value=True), \ mock.patch("agent_guard.notifier.subprocess.run", return_value=proc) as run: nid = mod.gdbus_notify("summary", "body", ["Info", "Dismiss"], replaces_id=41) self.assertEqual(nid, 77) args = run.call_args.args[0] self.assertIn("41", args) def test_gdbus_notify_stays_off_the_bus_without_approval(self): mod = importlib.import_module("agent_guard.notifier") with mock.patch("agent_guard.notifier.gate_allows", return_value=False), \ mock.patch("agent_guard.notifier.subprocess.run") as run: nid = mod.gdbus_notify("summary", "body", ["Dismiss"]) self.assertEqual(nid, 0) run.assert_not_called() def test_info_action_opens_url_and_re_notifies(self): mod = importlib.import_module("agent_guard.daemon") event = Event("warning", "grafana", "Memory low", info_url="http://127.0.0.1:3000/d/abc") cfg = SimpleNamespace(tmp_stale_hours=72, protect={"systemd"}) with mock.patch("agent_guard.daemon.subprocess.run") as run, mock.patch("agent_guard.daemon.notify_event") as notify: mod.handle_action(event, "info", cfg) run.assert_called_once() self.assertEqual(run.call_args.args[0][-1], event.info_url) notify.assert_called_once_with(event) def test_kill_remediation_rescans_culprit_and_sets_info_kill_actions(self): mod = importlib.import_module("agent_guard.daemon") culprit = {"name": "python3", "pids": [{"pid": 10}, {"pid": 11}]} event = Event("warning", "grafana", "Memory exhaustion ETA below 90s", remediation="kill", info_url="http://127.0.0.1:3000/d/system-monitor-overview") cfg = SimpleNamespace(protect={"systemd"}) with mock.patch("agent_guard.daemon.rescan_culprit", return_value=culprit) as rescan, mock.patch("agent_guard.daemon.notify_event") as notify: mod.handle_event(event, cfg) rescan.assert_called_once_with(cfg.protect) self.assertEqual(event.culprit, culprit) self.assertEqual(event.actions, ["Info", "Kill python3 (2 procs)"]) notify.assert_called_once_with(event) def test_clean_remediation_sets_clean_action_and_handle_action_calls_clean_tmp_safe(self): mod = importlib.import_module("agent_guard.daemon") event = Event("warning", "grafana", "/tmp pressure", remediation="clean", info_url="http://127.0.0.1:3000/d/system-monitor-overview") cfg = SimpleNamespace(tmp_stale_hours=72, protect={"systemd"}) with mock.patch("agent_guard.daemon.rescan_culprit") as rescan, mock.patch("agent_guard.daemon.notify_event") as notify: mod.handle_event(event, cfg) rescan.assert_not_called() self.assertEqual(event.actions, ["Info", "Clean /tmp"]) notify.assert_called_once_with(event) with mock.patch("agent_guard.daemon.clean_tmp_safe", return_value=5) as clean, mock.patch("agent_guard.daemon.journal_send"): mod.handle_action(event, "clean_/tmp", cfg) clean.assert_called_once_with(uid=1000, older_than_seconds=72 * 3600) if __name__ == "__main__": unittest.main()