diff --git a/development/tools/sam-cop/channels.py b/development/tools/sam-cop/channels.py new file mode 100644 index 00000000..78e576ff --- /dev/null +++ b/development/tools/sam-cop/channels.py @@ -0,0 +1,67 @@ +import abc +from typing import List + +import httpx + + +class Channel(abc.ABC): + """Delivery backend. Subclasses self-register; declare required_env and build_channels() picks them up.""" + + name = "channel" + required_env: tuple = () + registry: list = [] + + def __init_subclass__(cls, **kwargs): + super().__init_subclass__(**kwargs) + Channel.registry.append(cls) + + @classmethod + def from_env(cls, env): + # __init__ parameters must line up with required_env order. + return cls(*(env[var] for var in cls.required_env)) + + @abc.abstractmethod + async def send(self, message: str) -> None: ... + + +class SlackChannel(Channel): + name = "slack" + required_env = ("SLACK_WEBHOOK_URL",) + + def __init__(self, webhook_url: str): + self.webhook_url = webhook_url + + async def send(self, message: str) -> None: + async with httpx.AsyncClient() as http_client: + response = await http_client.post(self.webhook_url, json={"text": message}, timeout=10.0) + response.raise_for_status() + + +class TelegramChannel(Channel): + name = "telegram" + required_env = ("TELEGRAM_BOT_TOKEN", "TELEGRAM_CHAT_ID") + + def __init__(self, bot_token: str, chat_id: str): + self.bot_token = bot_token + self.chat_id = chat_id + + async def send(self, message: str) -> None: + url = f"https://api.telegram.org/bot{self.bot_token}/sendMessage" + async with httpx.AsyncClient() as http_client: + response = await http_client.post(url, json={"chat_id": self.chat_id, "text": message}, timeout=10.0) + response.raise_for_status() + + +class StdoutChannel(Channel): + name = "stdout" + + async def send(self, message: str) -> None: + print(f"[ALERT] {message}", flush=True) + + +def build_channels(env) -> List[Channel]: + # Stdout is always on so the log shows every alert, delivered or not. + channels = [cls.from_env(env) for cls in Channel.registry + if cls.required_env and all(env.get(var) for var in cls.required_env)] + channels.append(StdoutChannel()) + return channels diff --git a/development/tools/sam-cop/sam_cop.py b/development/tools/sam-cop/sam_cop.py new file mode 100644 index 00000000..994906ce --- /dev/null +++ b/development/tools/sam-cop/sam_cop.py @@ -0,0 +1,415 @@ +import asyncio +import json +import os +import re +import signal +import sys +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Dict, List, Optional, Any, Tuple + +from sam_mcp.client import SamClient + +from channels import Channel, build_channels + + +@dataclass +class SamCopConfig: + poll_interval: float + miss_threshold: int + min_peers: int + + +def load_config(env) -> SamCopConfig: + return SamCopConfig( + poll_interval=float(env.get("POLL_INTERVAL", "30")), + miss_threshold=int(env.get("MISS_THRESHOLD", "3")), + min_peers=int(env.get("MIN_PEERS", "1")), + ) + + +@dataclass +class Alert: + severity: str + title: str + description: str + peer_id: str = "" + service: str = "" + + +SEVERITY_EMOJI = {"critical": "🚨", "warning": "⚠ī¸", "info": "ℹī¸"} + +CONTROL_CHARS_RE = re.compile(r"[\x00-\x1f\x7f]+") + + +def sanitize_alert_text(text: str, limit: int = 200) -> str: + """Strips control chars (incl. newlines) so attacker-controlled strings can't forge alert lines.""" + cleaned = CONTROL_CHARS_RE.sub(" ", text).strip() + if len(cleaned) > limit: + cleaned = cleaned[:limit].rstrip() + "..." + return cleaned + + +def format_alert(alert: Alert, node_peer_id: str, timestamp: str) -> str: + title = sanitize_alert_text(alert.title) + lines = [f"{SEVERITY_EMOJI[alert.severity]} [{alert.severity.upper()}] {title}"] + if alert.description: + lines.append(sanitize_alert_text(alert.description)) + if alert.peer_id: + lines.append(f"peer: {sanitize_alert_text(alert.peer_id)}") + if alert.service: + lines.append(f"service: {sanitize_alert_text(alert.service)}") + lines.append(f"reported by {node_peer_id} at {timestamp}") + return "\n".join(lines) + + +async def deliver(channels: List[Channel], message: str) -> None: + async def deliver_to_channel(channel: Channel) -> None: + for attempt in range(3): + try: + await channel.send(message) + return + except Exception as error: + if attempt == 2: + print(f"[-] {channel.name} delivery failed after 3 attempts: {error}", flush=True) + else: + await asyncio.sleep(2**attempt) + + # Concurrent so one slow or failing channel's retries don't delay the others. + await asyncio.gather(*(deliver_to_channel(channel) for channel in channels)) + + +SEVERITY_BY_EVENT_TYPE = { + "banned": "critical", + "spoofing_attempt": "critical", + "stale_event": "warning", + "rate_limit_drop": "warning", + "key_rotation": "warning", + "policy_update": "info", +} + +# Event types that arrive as attacker-controlled floods and get aggregated per peer per cycle. +AGGREGATED_EVENT_TYPES = ("rate_limit_drop", "spoofing_attempt", "stale_event") + + +@dataclass +class SamCopState: + cursor: int = 0 + baseline: set = field(default_factory=set) + miss_counts: dict = field(default_factory=dict) + partitioned: bool = False + first_snapshot: bool = True + routers: set = field(default_factory=set) + signing_keys: set = field(default_factory=set) + control_plane_first_snapshot: bool = True + control_plane_misses: int = 0 + + +def detect_partition(state: SamCopState, connected_peer_count: int, min_peers: int) -> List[Alert]: + alerts = [] + if connected_peer_count < min_peers: + if not state.partitioned: + alerts.append(Alert("critical", "node partitioned", + f"connected peers dropped to {connected_peer_count} (min {min_peers}); churn detection suspended")) + state.partitioned = True + elif state.partitioned: + alerts.append(Alert("info", "connectivity restored", + f"{connected_peer_count} peer(s) connected; churn baseline reset")) + state.partitioned = False + state.first_snapshot = True + state.miss_counts = {} + return alerts + + +def detect_node_events(state: SamCopState, node_events: List[dict], latest_seq: int) -> List[Alert]: + alerts = [] + # Tally flood-prone event types per peer so N events collapse into one alert; other types alert one-to-one. + counts_by_type_and_peer: Dict[str, Dict[str, int]] = {event_type: {} for event_type in AGGREGATED_EVENT_TYPES} + for event in node_events: + event_type = event.get("type", "") + if event_type in counts_by_type_and_peer: + peer_id = event.get("peer_id", "") + counts_by_type_and_peer[event_type][peer_id] = counts_by_type_and_peer[event_type].get(peer_id, 0) + 1 + continue + severity = SEVERITY_BY_EVENT_TYPE.get(event_type, "info") + alerts.append(Alert(severity, f"node event: {event_type}", + event.get("message", ""), peer_id=event.get("peer_id", ""))) + for peer_id, count in counts_by_type_and_peer["rate_limit_drop"].items(): + alerts.append(Alert("warning", "node event: rate_limit_drop", + f"{count} message(s) dropped this cycle", peer_id=peer_id)) + for peer_id, count in counts_by_type_and_peer["spoofing_attempt"].items(): + alerts.append(Alert("critical", "node event: spoofing_attempt", + f"{count} spoofing attempt(s) (invalid event signature) this cycle", peer_id=peer_id)) + for peer_id, count in counts_by_type_and_peer["stale_event"].items(): + alerts.append(Alert("warning", "node event: stale_event", + f"{count} stale event(s) this cycle", peer_id=peer_id)) + if state.cursor > 0 and latest_seq - state.cursor > len(node_events): + lost = latest_seq - state.cursor - len(node_events) + alerts.append(Alert("warning", "node event buffer wrapped", + f"~{lost} node event(s) lost before this poll")) + state.cursor = latest_seq + return alerts + + +def detect_churn(state: SamCopState, services_snapshot: set, miss_threshold: int) -> List[Alert]: + if state.partitioned: + return [] + if state.first_snapshot: + state.baseline = set(services_snapshot) + state.miss_counts = {} + state.first_snapshot = False + return [] + + alerts = [] + for entry in sorted(services_snapshot - state.baseline): + service_type, service_name, peer_id = entry + alerts.append(Alert("info", "service appeared", f"{service_type}/{service_name} advertised on the mesh", + peer_id=peer_id, service=f"{service_type}/{service_name}")) + state.baseline.add(entry) + for entry in list(state.miss_counts): + if entry in services_snapshot: + del state.miss_counts[entry] + for entry in sorted(state.baseline - services_snapshot): + misses = state.miss_counts.get(entry, 0) + 1 + if misses >= miss_threshold: + service_type, service_name, peer_id = entry + alerts.append(Alert("warning", "service disappeared", + f"{service_type}/{service_name} missing for {misses} consecutive polls", + peer_id=peer_id, service=f"{service_type}/{service_name}")) + state.baseline.discard(entry) + state.miss_counts.pop(entry, None) + else: + state.miss_counts[entry] = misses + return alerts + + +# Sentinel distinguishing "control plane not polled" (tests) from "poll failed" (None). +CONTROL_PLANE_UNPOLLED = object() + + +def router_peer_ids(router_addresses) -> set: + """One router leases several multiaddrs; collapse them to the trailing /p2p/ peer ID.""" + routers = set() + for address in router_addresses: + if "/p2p/" in address: + routers.add(address.rsplit("/p2p/", 1)[1]) + else: + routers.add(address) + return routers + + +def detect_control_plane(state: SamCopState, control_plane_info, miss_threshold: int) -> List[Alert]: + if control_plane_info is CONTROL_PLANE_UNPOLLED: + return [] + + alerts = [] + if control_plane_info is None: + state.control_plane_misses += 1 + if state.control_plane_misses == miss_threshold: + alerts.append(Alert("warning", "control plane unreachable", + f"/info and /keys fetch failed for {miss_threshold} consecutive polls")) + return alerts + if state.control_plane_misses >= miss_threshold: + alerts.append(Alert("info", "control plane reachable again", + "control plane info fetch succeeded after repeated failures")) + state.control_plane_misses = 0 + + routers = router_peer_ids(control_plane_info.get("router_addresses") or []) + signing_keys = set(control_plane_info.get("signing_keys") or []) + + if state.control_plane_first_snapshot: + state.routers = routers + state.signing_keys = signing_keys + state.control_plane_first_snapshot = False + return alerts + + # No miss-threshold debounce: the control plane's lease TTL already debounces router churn. + for router in sorted(routers - state.routers): + alerts.append(Alert("info", "router appeared", "new router leased on the control plane", peer_id=router)) + for router in sorted(state.routers - routers): + alerts.append(Alert("warning", "router disappeared", + "router lease expired or was dropped on the control plane", peer_id=router)) + state.routers = routers + + for signing_key in sorted(signing_keys - state.signing_keys): + alerts.append(Alert("critical", "new control-plane signing key", + f"unrecognized signing key published on /keys: {signing_key[:16]}...")) + for signing_key in sorted(state.signing_keys - signing_keys): + alerts.append(Alert("warning", "control-plane signing key retired", + f"signing key removed from /keys: {signing_key[:16]}...")) + state.signing_keys = signing_keys + return alerts + + +def evaluate_cycle(state: SamCopState, connected_peer_count: int, node_events: List[dict], + latest_seq: int, services_snapshot: set, config: SamCopConfig, + control_plane_info=CONTROL_PLANE_UNPOLLED): + alerts = [] + alerts.extend(detect_partition(state, connected_peer_count, config.min_peers)) + alerts.extend(detect_node_events(state, node_events, latest_seq)) + alerts.extend(detect_churn(state, services_snapshot, config.miss_threshold)) + alerts.extend(detect_control_plane(state, control_plane_info, config.miss_threshold)) + return state, alerts + + +SERVICE_TYPES = ["mcp", "inference"] +DISCOVERY_PAGE_LIMIT = 200 + + +def parse_tool_json(result) -> Optional[Any]: + if result.get("isError"): + content = result.get("content") or [] + detail = content[0].get("text", "") if content else repr(result) + raise RuntimeError(f"tool error: {detail}") + content = result.get("content") or [] + if not content: + return None + text = content[0].get("text", "") + if not text: + return None + try: + return json.loads(text) + except json.JSONDecodeError: + return None + + +async def fetch_services(client) -> set: + snapshot = set() + for service_type in SERVICE_TYPES: + offset = 0 + while True: + result = await client.call_tool("discover_remote_services", + {"type": service_type, "limit": DISCOVERY_PAGE_LIMIT, "offset": offset}) + providers = parse_tool_json(result) or [] + for provider in providers: + snapshot.add((service_type, provider.get("srv_name", ""), provider.get("peer_id", ""))) + if len(providers) < DISCOVERY_PAGE_LIMIT: + break + offset += DISCOVERY_PAGE_LIMIT + return snapshot + + +async def run_cycle(client, state: SamCopState, channels: List[Channel], config: SamCopConfig, node_peer_id: str) -> SamCopState: + mesh_info = parse_tool_json(await client.call_tool("get_mesh_info", {})) or {} + connected_peer_count = len(mesh_info.get("connected_peers") or []) + + events_response = parse_tool_json(await client.call_tool("poll_node_events", {"since_seq": state.cursor})) or {} + node_events = events_response.get("events") or [] + latest_seq = events_response.get("latest_seq", state.cursor) + + services_snapshot = await fetch_services(client) + + # A control plane outage must not fail the cycle: node-event detection stays live + # and the failure must not feed the reconnect counter. + try: + control_plane_info = parse_tool_json(await client.call_tool("get_control_plane_info", {})) + except Exception as error: + control_plane_info = None + print(f"[-] control plane info fetch failed: {error}", flush=True) + + state, alerts = evaluate_cycle(state, connected_peer_count, node_events, latest_seq, + services_snapshot, config, control_plane_info) + + timestamp = datetime.now(timezone.utc).isoformat(timespec="seconds") + for alert in alerts: + await deliver(channels, format_alert(alert, node_peer_id, timestamp)) + return state + + +async def connect_with_retry(retries: int = 12, delay: float = 5.0) -> SamClient: + for attempt in range(1, retries + 1): + client = SamClient() + try: + await client.connect() + return client + except Exception as error: + if attempt == retries: + raise + print(f"[-] Failed to connect to SAM node (attempt {attempt}/{retries}): {error}. " + f"Retrying in {delay} seconds...", flush=True) + await asyncio.sleep(delay) + + +CONSECUTIVE_FAILURE_LIMIT = 3 +ABANDONED_CLIENT_LIMIT = 50 + +_shutdown = False + +# Dead clients are parked, never released: closing one raises from a foreign task, +# and letting the GC finalize it cancels a scope owned by ours. Reconnects are rare. +_abandoned: List[SamClient] = [] + + +def _request_shutdown() -> None: + global _shutdown + _shutdown = True + + +async def reconnect(client, state: SamCopState) -> Tuple[SamClient, str, SamCopState]: + """Recovers from a lost MCP session; the node's seq counter restarts too, so reset cursor.""" + _abandoned.append(client) + # Each parked client keeps a file descriptor; exit before they pile up - + # the supervisor restart clears the list. + if len(_abandoned) >= ABANDONED_CLIENT_LIMIT: + print(f"[-] {len(_abandoned)} stale clients parked, exiting for a clean restart", flush=True) + sys.exit(1) + client = await connect_with_retry() + mesh_info = parse_tool_json(await client.call_tool("get_mesh_info", {})) or {} + node_peer_id = mesh_info.get("peer_id", "unknown") + state.cursor = 0 + return client, node_peer_id, state + + +async def run_sam_cop(): + config = load_config(os.environ) + channels = build_channels(os.environ) + channel_names = ", ".join(channel.name for channel in channels) + print(f"[*] sam-cop starting: poll={config.poll_interval}s miss_threshold={config.miss_threshold} " + f"min_peers={config.min_peers} channels=[{channel_names}]", flush=True) + + loop = asyncio.get_running_loop() + for sig in (signal.SIGINT, signal.SIGTERM): + loop.add_signal_handler(sig, _request_shutdown) + + client = await connect_with_retry() + try: + mesh_info = parse_tool_json(await client.call_tool("get_mesh_info", {})) or {} + node_peer_id = mesh_info.get("peer_id", "unknown") + print(f"[+] connected to local node {node_peer_id}", flush=True) + + state = SamCopState() + consecutive_failures = 0 + while True: + try: + state = await run_cycle(client, state, channels, config, node_peer_id) + consecutive_failures = 0 + # A lost session surfaces as a bare CancelledError from anyio's cancel scope, + # indistinguishable by type from a real shutdown - hence the explicit flag. + except (Exception, asyncio.CancelledError) as error: + if _shutdown: + raise + consecutive_failures += 1 + print(f"[-] poll cycle failed ({consecutive_failures}/{CONSECUTIVE_FAILURE_LIMIT}): {error}", flush=True) + if consecutive_failures >= CONSECUTIVE_FAILURE_LIMIT: + print("[-] too many consecutive failures, reconnecting to SAM node", flush=True) + client, node_peer_id, state = await reconnect(client, state) + consecutive_failures = 0 + print(f"[+] reconnected to local node {node_peer_id}", flush=True) + try: + await asyncio.sleep(config.poll_interval) + except (Exception, asyncio.CancelledError): + # A discarded session can still cancel us here, outside the cycle. + if _shutdown: + raise + finally: + try: + await client.close() + except BaseException as error: + print(f"[-] error closing client on shutdown: {error}", flush=True) + + +if __name__ == "__main__": + try: + asyncio.run(run_sam_cop()) + except KeyboardInterrupt: + sys.exit(0) diff --git a/development/tools/sam-cop/test_sam_cop.py b/development/tools/sam-cop/test_sam_cop.py new file mode 100644 index 00000000..b1f27201 --- /dev/null +++ b/development/tools/sam-cop/test_sam_cop.py @@ -0,0 +1,495 @@ +import asyncio +import json + +import pytest + +from channels import ( + Channel, + SlackChannel, + StdoutChannel, + TelegramChannel, + build_channels, +) +from sam_cop import ( + Alert, + deliver, + format_alert, + load_config, + sanitize_alert_text, +) + + +def test_load_config_defaults(): + config = load_config({}) + assert config.poll_interval == 30.0 + assert config.miss_threshold == 3 + assert config.min_peers == 1 + + +def test_load_config_overrides(): + config = load_config({"POLL_INTERVAL": "5", "MISS_THRESHOLD": "2", "MIN_PEERS": "0"}) + assert config.poll_interval == 5.0 + assert config.miss_threshold == 2 + assert config.min_peers == 0 + + +def test_build_channels_stdout_fallback(): + channels = build_channels({}) + assert len(channels) == 1 + assert isinstance(channels[0], StdoutChannel) + + +def test_build_channels_slack_and_telegram(): + env = { + "SLACK_WEBHOOK_URL": "https://hooks.slack.example/x", + "TELEGRAM_BOT_TOKEN": "token", + "TELEGRAM_CHAT_ID": "42", + } + channels = build_channels(env) + assert {type(c) for c in channels} == {SlackChannel, TelegramChannel} + + +def test_build_channels_telegram_requires_chat_id(): + channels = build_channels({"TELEGRAM_BOT_TOKEN": "token"}) + assert len(channels) == 1 + assert isinstance(channels[0], StdoutChannel) + + +def test_build_channels_picks_up_registered_subclass(): + class EchoChannel(Channel): + name = "echo" + required_env = ("ECHO_TARGET",) + + def __init__(self, target): + self.target = target + + async def send(self, message): + pass + + try: + channels = build_channels({"ECHO_TARGET": "x"}) + assert len(channels) == 1 + assert isinstance(channels[0], EchoChannel) + assert channels[0].target == "x" + finally: + Channel.registry.remove(EchoChannel) + + +def test_format_alert(): + alert = Alert("critical", "node event: banned", "peer banned by hub", peer_id="12D3KooPeer") + message = format_alert(alert, "12D3KooSelf", "2026-08-06T10:00:00+00:00") + assert message.startswith("🚨 [CRITICAL] node event: banned") + assert "peer banned by hub" in message + assert "peer: 12D3KooPeer" in message + assert "reported by 12D3KooSelf at 2026-08-06T10:00:00+00:00" in message + + +class FlakyChannel(Channel): + name = "flaky" + + def __init__(self, failures): + self.failures = failures + self.attempts = 0 + self.delivered = [] + + async def send(self, message): + self.attempts += 1 + if self.attempts <= self.failures: + raise RuntimeError("boom") + self.delivered.append(message) + + +def test_deliver_retries_then_succeeds(): + channel = FlakyChannel(failures=2) + asyncio.run(deliver([channel], "hello")) + assert channel.delivered == ["hello"] + assert channel.attempts == 3 + + +def test_deliver_gives_up_without_raising(): + channel = FlakyChannel(failures=5) + asyncio.run(deliver([channel], "hello")) + assert channel.delivered == [] + assert channel.attempts == 3 + + +from sam_cop import SamCopConfig, SamCopState, evaluate_cycle + +CONFIG = SamCopConfig(poll_interval=30.0, miss_threshold=3, min_peers=1) +SERVICE_A = ("mcp", "service-a", "peer-1") +SERVICE_B = ("inference", "service-b", "peer-2") + + +def cycle(state, peers=2, events=None, latest_seq=None, snapshot=frozenset()): + events = events or [] + if latest_seq is None: + latest_seq = state.cursor + len(events) + return evaluate_cycle(state, peers, events, latest_seq, set(snapshot), CONFIG) + + +def test_first_snapshot_sets_baseline_without_alerts(): + state, alerts = cycle(SamCopState(), snapshot={SERVICE_A}) + assert alerts == [] + assert state.baseline == {SERVICE_A} + + +def test_service_appeared_alerts_immediately(): + state, _ = cycle(SamCopState(), snapshot={SERVICE_A}) + state, alerts = cycle(state, snapshot={SERVICE_A, SERVICE_B}) + assert [a.severity for a in alerts] == ["info"] + assert "appeared" in alerts[0].title + assert SERVICE_B in state.baseline + + +def test_service_disappeared_needs_consecutive_misses(): + state, _ = cycle(SamCopState(), snapshot={SERVICE_A}) + state, alerts = cycle(state, snapshot=set()) + assert alerts == [] + state, alerts = cycle(state, snapshot=set()) + assert alerts == [] + state, alerts = cycle(state, snapshot=set()) + assert [a.severity for a in alerts] == ["warning"] + assert "disappeared" in alerts[0].title + assert SERVICE_A not in state.baseline + + +def test_reappearing_service_resets_miss_count(): + state, _ = cycle(SamCopState(), snapshot={SERVICE_A}) + state, _ = cycle(state, snapshot=set()) + state, _ = cycle(state, snapshot={SERVICE_A}) + state, alerts = cycle(state, snapshot=set()) + assert alerts == [] + + +def test_partition_alerts_once_and_suppresses_churn(): + state, _ = cycle(SamCopState(), snapshot={SERVICE_A}) + state, alerts = cycle(state, peers=0, snapshot=set()) + assert [a.severity for a in alerts] == ["critical"] + state, alerts = cycle(state, peers=0, snapshot=set()) + assert alerts == [] + assert state.baseline == {SERVICE_A} + assert state.miss_counts == {} + + +def test_partition_recovery_resets_baseline(): + state, _ = cycle(SamCopState(), snapshot={SERVICE_A}) + state, _ = cycle(state, peers=0, snapshot=set()) + state, alerts = cycle(state, peers=2, snapshot={SERVICE_B}) + assert [a.severity for a in alerts] == ["info"] + assert "restored" in alerts[0].title + state, alerts = cycle(state, snapshot={SERVICE_B}) + assert alerts == [] + assert state.baseline == {SERVICE_B} + + +def test_node_event_severities(): + events = [ + {"seq": 1, "type": "banned", "peer_id": "p", "message": "m"}, + {"seq": 2, "type": "spoofing_attempt", "peer_id": "p", "message": "m"}, + {"seq": 3, "type": "policy_update", "peer_id": "", "message": "m"}, + {"seq": 4, "type": "key_rotation", "peer_id": "", "message": "m"}, + ] + state, alerts = cycle(SamCopState(cursor=0, first_snapshot=False), events=events, latest_seq=4) + # spoofing_attempt is aggregated per peer, so it surfaces after the one-alert-per-event types. + assert [a.severity for a in alerts] == ["critical", "info", "warning", "critical"] + assert state.cursor == 4 + + +def test_rate_limit_drops_aggregate_per_peer(): + events = [ + {"seq": 1, "type": "rate_limit_drop", "peer_id": "p1", "message": "m"}, + {"seq": 2, "type": "rate_limit_drop", "peer_id": "p1", "message": "m"}, + {"seq": 3, "type": "rate_limit_drop", "peer_id": "p2", "message": "m"}, + ] + _, alerts = cycle(SamCopState(first_snapshot=False), events=events, latest_seq=3) + assert len(alerts) == 2 + assert all(a.severity == "warning" for a in alerts) + descriptions = " | ".join(a.description for a in alerts) + assert "2 message(s)" in descriptions and "1 message(s)" in descriptions + + +def test_spoofing_attempts_aggregate_per_peer(): + events = [ + {"seq": 1, "type": "spoofing_attempt", "peer_id": "p1", "message": "m"}, + {"seq": 2, "type": "spoofing_attempt", "peer_id": "p1", "message": "m"}, + {"seq": 3, "type": "spoofing_attempt", "peer_id": "p1", "message": "m"}, + ] + _, alerts = cycle(SamCopState(first_snapshot=False), events=events, latest_seq=3) + assert len(alerts) == 1 + assert alerts[0].severity == "critical" + assert alerts[0].peer_id == "p1" + assert "3 spoofing attempt(s)" in alerts[0].description + + +def test_stale_events_aggregate_per_peer(): + events = [ + {"seq": 1, "type": "stale_event", "peer_id": "p2", "message": "m"}, + {"seq": 2, "type": "stale_event", "peer_id": "p2", "message": "m"}, + ] + _, alerts = cycle(SamCopState(first_snapshot=False), events=events, latest_seq=2) + assert len(alerts) == 1 + assert alerts[0].severity == "warning" + assert alerts[0].peer_id == "p2" + assert "2 stale event(s)" in alerts[0].description + + +def test_event_loss_detected_after_first_poll(): + state = SamCopState(cursor=5, first_snapshot=False) + state, alerts = cycle(state, events=[{"seq": 100, "type": "policy_update", "peer_id": "", "message": "m"}], latest_seq=100) + severities = [a.severity for a in alerts] + assert severities.count("warning") == 1 + assert any("buffer wrapped" in a.title for a in alerts) + assert state.cursor == 100 + + +def test_no_event_loss_alert_on_first_poll(): + _, alerts = cycle(SamCopState(first_snapshot=False), events=[], latest_seq=5000) + assert alerts == [] + + +def test_sanitize_alert_text_strips_control_chars_and_collapses(): + hostile = "svc\n🚨 [CRITICAL] fake alert\x07more" + sanitized = sanitize_alert_text(hostile) + assert "\n" not in sanitized + assert "\x07" not in sanitized + assert "svc" in sanitized and "fake alert" in sanitized + + +def test_sanitize_alert_text_truncates_with_ellipsis(): + truncated = sanitize_alert_text("a" * 500, limit=50) + assert len(truncated) == 53 + assert truncated.endswith("...") + + +def test_format_alert_sanitizes_hostile_service_name(): + hostile_service = "svc\n🚨 [CRITICAL] fake node event: banned" + alert = Alert("info", "service appeared", "desc", peer_id="peer-1", service=hostile_service) + message = format_alert(alert, "12D3KooSelf", "2026-08-06T10:00:00+00:00") + lines = message.split("\n") + assert len(lines) == 5 # no forged extra line from the injected newline + assert lines[0] == "ℹī¸ [INFO] service appeared" + assert "\n" not in lines[3] + assert "svc" in lines[3] and "CRITICAL" in lines[3] + + +from sam_cop import fetch_services, parse_tool_json + + +def test_parse_tool_json(): + result = {"content": [{"text": '{"events": [], "latest_seq": 7}'}]} + assert parse_tool_json(result) == {"events": [], "latest_seq": 7} + + +def test_parse_tool_json_empty_text(): + assert parse_tool_json({"content": [{"text": ""}]}) is None + + +def test_parse_tool_json_raises_on_tool_error(): + result = {"isError": True, "content": [{"text": "boom"}]} + with pytest.raises(RuntimeError): + parse_tool_json(result) + + +def test_parse_tool_json_empty_content_list_returns_none(): + assert parse_tool_json({"content": []}) is None + + +class FakeSamClient: + def __init__(self, pages): + self.pages = pages + self.calls = [] + + async def call_tool(self, name, arguments): + self.calls.append((name, arguments)) + service_type = arguments["type"] + offset = arguments["offset"] + providers = self.pages.get(service_type, [])[offset:offset + arguments["limit"]] + return {"content": [{"text": json.dumps(providers)}]} + + +def test_fetch_services_paginates_and_builds_tuples(): + mcp_providers = [{"peer_id": f"peer-{i}", "srv_name": f"service-{i}"} for i in range(250)] + client = FakeSamClient({"mcp": mcp_providers, "inference": [{"peer_id": "peer-x", "srv_name": "vllm"}]}) + snapshot = asyncio.run(fetch_services(client)) + assert ("inference", "vllm", "peer-x") in snapshot + assert len(snapshot) == 251 + mcp_calls = [c for c in client.calls if c[1]["type"] == "mcp"] + assert len(mcp_calls) == 2 # 250 providers, limit 200 → two pages + + +def test_connect_with_retry_returns_after_transient_failures(monkeypatch): + import sam_cop as sam_cop_module + + attempts = [] + + class FlakyConnectClient: + async def connect(self): + attempts.append(1) + if len(attempts) < 3: + raise RuntimeError("node not up yet") + + monkeypatch.setattr(sam_cop_module, "SamClient", FlakyConnectClient) + client = asyncio.run(sam_cop_module.connect_with_retry(retries=5, delay=0)) + assert isinstance(client, FlakyConnectClient) + assert len(attempts) == 3 + + +def test_reconnect_swallows_close_errors_refetches_peer_id_and_resets_cursor(monkeypatch): + import sam_cop as sam_cop_module + + class BrokenCloseClient: + async def close(self): + raise RuntimeError("close boom") + + class NewClient: + async def call_tool(self, name, arguments): + return {"content": [{"text": json.dumps({"peer_id": "peer-new"})}]} + + async def fake_connect_with_retry(retries=12, delay=5.0): + return NewClient() + + monkeypatch.setattr(sam_cop_module, "connect_with_retry", fake_connect_with_retry) + + state = SamCopState(cursor=42, first_snapshot=False) + new_client, node_peer_id, state = asyncio.run(sam_cop_module.reconnect(BrokenCloseClient(), state)) + assert isinstance(new_client, NewClient) + assert node_peer_id == "peer-new" + assert state.cursor == 0 + + +class StopLoop(Exception): + """Sentinel used to break run_sam_cop's infinite loop once the assertions are ready.""" + + +def test_run_sam_cop_reconnects_after_three_consecutive_failures(monkeypatch): + import sam_cop as sam_cop_module + + class FailsAfterStartupClient: + def __init__(self): + self.calls = 0 + self.closed = False + + async def call_tool(self, name, arguments): + self.calls += 1 + if self.calls == 1: + return {"content": [{"text": json.dumps({"peer_id": "peer-old"})}]} + raise RuntimeError("mesh unreachable") + + async def close(self): + self.closed = True + + class ReconnectedClient: + def __init__(self): + self.calls = 0 + + async def call_tool(self, name, arguments): + self.calls += 1 + return {"content": [{"text": json.dumps({"peer_id": "peer-new"})}]} + + async def close(self): + pass + + old_client = FailsAfterStartupClient() + new_client = ReconnectedClient() + connect_calls = [] + + async def fake_connect_with_retry(retries=12, delay=5.0): + connect_calls.append(1) + return old_client if len(connect_calls) == 1 else new_client + + sleep_calls = [] + + async def fake_sleep(delay): + sleep_calls.append(delay) + if len(sleep_calls) >= 3: # 2 failed cycles + the reconnecting cycle + raise StopLoop() + + monkeypatch.setattr(sam_cop_module, "connect_with_retry", fake_connect_with_retry) + monkeypatch.setattr(sam_cop_module.asyncio, "sleep", fake_sleep) + monkeypatch.setattr(sam_cop_module.os, "environ", {}) + + with pytest.raises(StopLoop): + asyncio.run(sam_cop_module.run_sam_cop()) + + assert old_client.closed is True + assert len(connect_calls) == 2 + assert new_client.calls >= 1 + + +from sam_cop import CONTROL_PLANE_UNPOLLED, detect_control_plane, router_peer_ids + +ROUTER_A_ADDRESSES = ["/dns4/router-a/tcp/4001/p2p/12D3KooRouterA", + "/dns4/router-a/udp/4001/quic-v1/p2p/12D3KooRouterA"] +ROUTER_B_ADDRESS = "/dns4/router-b/tcp/4001/p2p/12D3KooRouterB" +CONTROL_PLANE_INFO = {"router_addresses": ROUTER_A_ADDRESSES + [ROUTER_B_ADDRESS], + "signing_keys": ["a2V5LW9uZQ=="]} + + +def test_router_peer_ids_collapses_multiaddrs_per_router(): + assert router_peer_ids(ROUTER_A_ADDRESSES) == {"12D3KooRouterA"} + assert router_peer_ids(["/dns4/no-peer-component/tcp/4001"]) == {"/dns4/no-peer-component/tcp/4001"} + + +def test_control_plane_first_snapshot_sets_baseline_without_alerts(): + state = SamCopState() + alerts = detect_control_plane(state, CONTROL_PLANE_INFO, CONFIG.miss_threshold) + assert alerts == [] + assert state.routers == {"12D3KooRouterA", "12D3KooRouterB"} + assert state.signing_keys == {"a2V5LW9uZQ=="} + + +def test_router_disappearance_alerts_warning_without_debounce(): + state = SamCopState() + detect_control_plane(state, CONTROL_PLANE_INFO, CONFIG.miss_threshold) + alerts = detect_control_plane(state, {"router_addresses": ROUTER_A_ADDRESSES, + "signing_keys": ["a2V5LW9uZQ=="]}, CONFIG.miss_threshold) + assert [a.severity for a in alerts] == ["warning"] + assert "router disappeared" in alerts[0].title + assert alerts[0].peer_id == "12D3KooRouterB" + + +def test_router_appearance_alerts_info(): + state = SamCopState() + detect_control_plane(state, {"router_addresses": ROUTER_A_ADDRESSES, "signing_keys": []}, CONFIG.miss_threshold) + alerts = detect_control_plane(state, CONTROL_PLANE_INFO, CONFIG.miss_threshold) + assert any(a.severity == "info" and "router appeared" in a.title and a.peer_id == "12D3KooRouterB" + for a in alerts) + + +def test_new_signing_key_alerts_critical(): + state = SamCopState() + detect_control_plane(state, CONTROL_PLANE_INFO, CONFIG.miss_threshold) + alerts = detect_control_plane(state, {"router_addresses": CONTROL_PLANE_INFO["router_addresses"], + "signing_keys": ["a2V5LW9uZQ==", "a2V5LXR3bw=="]}, CONFIG.miss_threshold) + assert [a.severity for a in alerts] == ["critical"] + assert "new control-plane signing key" in alerts[0].title + + +def test_signing_key_retirement_alerts_warning(): + state = SamCopState() + detect_control_plane(state, CONTROL_PLANE_INFO, CONFIG.miss_threshold) + alerts = detect_control_plane(state, {"router_addresses": CONTROL_PLANE_INFO["router_addresses"], + "signing_keys": []}, CONFIG.miss_threshold) + assert [a.severity for a in alerts] == ["warning"] + assert "signing key retired" in alerts[0].title + + +def test_control_plane_unreachable_alerts_once_at_threshold_then_recovers(): + state = SamCopState() + detect_control_plane(state, CONTROL_PLANE_INFO, CONFIG.miss_threshold) + assert detect_control_plane(state, None, CONFIG.miss_threshold) == [] + assert detect_control_plane(state, None, CONFIG.miss_threshold) == [] + alerts = detect_control_plane(state, None, CONFIG.miss_threshold) + assert [a.severity for a in alerts] == ["warning"] + assert "control plane unreachable" in alerts[0].title + assert detect_control_plane(state, None, CONFIG.miss_threshold) == [] + alerts = detect_control_plane(state, CONTROL_PLANE_INFO, CONFIG.miss_threshold) + assert [a.severity for a in alerts] == ["info"] + assert "reachable again" in alerts[0].title + + +def test_unpolled_control_plane_is_noop(): + state = SamCopState() + assert detect_control_plane(state, CONTROL_PLANE_UNPOLLED, CONFIG.miss_threshold) == [] + assert state.control_plane_first_snapshot is True + assert state.control_plane_misses == 0 diff --git a/internal/node/controlplane.go b/internal/node/controlplane.go index 959fcb75..c7dce87e 100644 --- a/internal/node/controlplane.go +++ b/internal/node/controlplane.go @@ -28,14 +28,14 @@ import ( "google.golang.org/protobuf/proto" ) -// FetchControlPlaneInfo retrieves the latest configuration from the control plane's /info endpoint. -func FetchControlPlaneInfo(ctx context.Context, controlPlaneURL string) (*api.ControlPlaneInfoResponse, error) { +// fetchControlPlaneProto GETs a control plane endpoint and returns the raw protobuf body. +func fetchControlPlaneProto(ctx context.Context, controlPlaneURL, path string) ([]byte, error) { if !strings.HasPrefix(controlPlaneURL, "http://") && !strings.HasPrefix(controlPlaneURL, "https://") { controlPlaneURL = "https://" + controlPlaneURL } controlPlaneURL = strings.TrimSuffix(controlPlaneURL, "/") - urlStr := controlPlaneURL + "/info" + urlStr := controlPlaneURL + path req, err := http.NewRequestWithContext(ctx, "GET", urlStr, nil) if err != nil { return nil, fmt.Errorf("failed to create HTTP request: %w", err) @@ -59,14 +59,35 @@ func FetchControlPlaneInfo(ctx context.Context, controlPlaneURL string) (*api.Co return nil, fmt.Errorf("control plane returned status %s: %s", resp.Status, string(body)) } + return body, nil +} + +// FetchControlPlaneInfo retrieves the latest configuration from the control plane's /info endpoint. +func FetchControlPlaneInfo(ctx context.Context, controlPlaneURL string) (*api.ControlPlaneInfoResponse, error) { + body, err := fetchControlPlaneProto(ctx, controlPlaneURL, "/info") + if err != nil { + return nil, err + } var info api.ControlPlaneInfoResponse if err := proto.Unmarshal(body, &info); err != nil { return nil, fmt.Errorf("failed to decode /info response: %w", err) } - return &info, nil } +// FetchControlPlaneKeys retrieves the currently valid signing keys from the control plane's /keys endpoint. +func FetchControlPlaneKeys(ctx context.Context, controlPlaneURL string) (*api.KeysResponse, error) { + body, err := fetchControlPlaneProto(ctx, controlPlaneURL, "/keys") + if err != nil { + return nil, err + } + var keys api.KeysResponse + if err := proto.Unmarshal(body, &keys); err != nil { + return nil, fmt.Errorf("failed to decode /keys response: %w", err) + } + return &keys, nil +} + // SyncMeshConfig loads the mesh configuration from the store, attempts to refresh it // via HTTP from the control plane, and updates the store if successful. // It returns the control plane public key and the latest multiaddresses. diff --git a/internal/node/controlplane_test.go b/internal/node/controlplane_test.go index 1ef342ad..43070296 100644 --- a/internal/node/controlplane_test.go +++ b/internal/node/controlplane_test.go @@ -16,6 +16,8 @@ package node import ( "context" + "encoding/base64" + "encoding/json" "net/http" "net/http/httptest" "reflect" @@ -23,6 +25,7 @@ import ( "testing" "github.com/google/sam/api" + "github.com/modelcontextprotocol/go-sdk/mcp" "google.golang.org/protobuf/proto" ) @@ -94,6 +97,106 @@ func TestFetchControlPlaneInfo_InvalidProto(t *testing.T) { } } +func TestFetchControlPlaneKeys(t *testing.T) { + expectedKeys := &api.KeysResponse{ + PublicKeys: [][]byte{[]byte("key-one"), []byte("key-two")}, + } + + body, err := proto.Marshal(expectedKeys) + if err != nil { + t.Fatalf("Failed to marshal keys: %v", err) + } + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/keys" { + t.Errorf("Expected path /keys, got %s", r.URL.Path) + } + w.WriteHeader(http.StatusOK) + _, _ = w.Write(body) + })) + defer server.Close() + + keys, err := FetchControlPlaneKeys(context.Background(), server.URL) + if err != nil { + t.Fatalf("FetchControlPlaneKeys failed: %v", err) + } + + if !reflect.DeepEqual(keys.PublicKeys, expectedKeys.PublicKeys) { + t.Errorf("Expected PublicKeys %v, got %v", expectedKeys.PublicKeys, keys.PublicKeys) + } +} + +func TestHandleGetControlPlaneInfo(t *testing.T) { + infoBody, err := proto.Marshal(&api.ControlPlaneInfoResponse{ + RouterAddresses: []string{"/dns4/router-a/tcp/4001/p2p/12D3KooRouterA"}, + }) + if err != nil { + t.Fatalf("Failed to marshal info: %v", err) + } + keysBody, err := proto.Marshal(&api.KeysResponse{ + PublicKeys: [][]byte{[]byte("key-one")}, + }) + if err != nil { + t.Fatalf("Failed to marshal keys: %v", err) + } + + mux := http.NewServeMux() + mux.HandleFunc("/info", func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write(infoBody) }) + mux.HandleFunc("/keys", func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write(keysBody) }) + server := httptest.NewServer(mux) + defer server.Close() + + store, err := NewStore(t.TempDir()) + if err != nil { + t.Fatalf("Failed to create store: %v", err) + } + if err := store.SaveControlPlaneURL(server.URL); err != nil { + t.Fatalf("Failed to save control plane URL: %v", err) + } + + node := &SamNode{Store: store} + result, _, err := node.handleGetControlPlaneInfo(context.Background(), nil, GetControlPlaneInfoParams{}) + if err != nil { + t.Fatalf("handler error: %v", err) + } + + text := result.Content[0].(*mcp.TextContent).Text + var response struct { + ControlPlaneURL string `json:"control_plane_url"` + RouterAddresses []string `json:"router_addresses"` + SigningKeys []string `json:"signing_keys"` + } + if err := json.Unmarshal([]byte(text), &response); err != nil { + t.Fatalf("bad JSON: %v", err) + } + if response.ControlPlaneURL != server.URL { + t.Errorf("Expected control_plane_url %s, got %s", server.URL, response.ControlPlaneURL) + } + if !reflect.DeepEqual(response.RouterAddresses, []string{"/dns4/router-a/tcp/4001/p2p/12D3KooRouterA"}) { + t.Errorf("unexpected router_addresses: %v", response.RouterAddresses) + } + expectedKey := base64.StdEncoding.EncodeToString([]byte("key-one")) + if !reflect.DeepEqual(response.SigningKeys, []string{expectedKey}) { + t.Errorf("unexpected signing_keys: %v", response.SigningKeys) + } +} + +func TestHandleGetControlPlaneInfo_NoURL(t *testing.T) { + store, err := NewStore(t.TempDir()) + if err != nil { + t.Fatalf("Failed to create store: %v", err) + } + + node := &SamNode{Store: store} + _, _, err = node.handleGetControlPlaneInfo(context.Background(), nil, GetControlPlaneInfoParams{}) + if err == nil { + t.Fatal("Expected error for missing control plane URL, got nil") + } + if !strings.Contains(err.Error(), "no control plane URL stored") { + t.Errorf("Expected 'no control plane URL stored' error, got %v", err) + } +} + func TestSyncMeshConfig(t *testing.T) { expectedInfo := &api.ControlPlaneInfoResponse{ RouterAddresses: []string{"/ip4/127.0.0.1/tcp/4001"}, diff --git a/internal/node/events.go b/internal/node/events.go new file mode 100644 index 00000000..cf357bb8 --- /dev/null +++ b/internal/node/events.go @@ -0,0 +1,89 @@ +package node + +import ( + "sync" + "time" +) + +const nodeEventBufferSize = 1000 + +// Node event categories. +const ( + EventCategorySecurity = "security" + EventCategoryMesh = "mesh_event" +) + +// Node event types. +const ( + EventTypeRateLimitDrop = "rate_limit_drop" + EventTypeSpoofingAttempt = "spoofing_attempt" + EventTypeStaleEvent = "stale_event" + EventTypeBanned = "banned" + EventTypeKeyRotation = "key_rotation" + EventTypePolicyUpdate = "policy_update" +) + +// NodeEvent is one verified observation recorded by the node. +type NodeEvent struct { + Seq uint64 `json:"seq"` + Timestamp int64 `json:"timestamp"` + Category string `json:"category"` + Type string `json:"type"` + PeerID string `json:"peer_id,omitempty"` + Message string `json:"message,omitempty"` +} + +type nodeEventBuffer struct { + mu sync.Mutex + entries []NodeEvent + nextSeq uint64 +} + +func newNodeEventBuffer() *nodeEventBuffer { + // Seqs start at 1 so cursor 0 means "never polled": poll(0) returns everything and an empty buffer reports latest seq 0. + return &nodeEventBuffer{nextSeq: 1} +} + +var globalEventBuffer = newNodeEventBuffer() + +// RecordNodeEvent appends an event to the global node event buffer. +func RecordNodeEvent(category, eventType, peerID, message string) { + globalEventBuffer.record(category, eventType, peerID, message) +} + +func (b *nodeEventBuffer) record(category, eventType, peerID, message string) { + b.mu.Lock() + defer b.mu.Unlock() + + b.entries = append(b.entries, NodeEvent{ + Seq: b.nextSeq, + Timestamp: time.Now().UnixMilli(), + Category: category, + Type: eventType, + PeerID: peerID, + Message: message, + }) + b.nextSeq++ + if len(b.entries) > nodeEventBufferSize { + discarded := len(b.entries) - nodeEventBufferSize + // Zero the trimmed prefix so its strings don't linger until the next reallocation. + for i := range discarded { + b.entries[i] = NodeEvent{} + } + b.entries = b.entries[discarded:] + } +} + +// poll returns events with seq > sinceSeq, oldest first, plus the latest seq. +func (b *nodeEventBuffer) poll(sinceSeq uint64) ([]NodeEvent, uint64) { + b.mu.Lock() + defer b.mu.Unlock() + + events := []NodeEvent{} + for _, event := range b.entries { + if event.Seq > sinceSeq { + events = append(events, event) + } + } + return events, b.nextSeq - 1 +} diff --git a/internal/node/events_test.go b/internal/node/events_test.go new file mode 100644 index 00000000..fc812175 --- /dev/null +++ b/internal/node/events_test.go @@ -0,0 +1,86 @@ +package node + +import ( + "sync" + "testing" +) + +func TestNodeEventBufferRecordAndPoll(t *testing.T) { + buffer := newNodeEventBuffer() + buffer.record("security", "spoofing_attempt", "peer-a", "invalid signature") + buffer.record("mesh_event", "banned", "peer-b", "peer banned by hub") + + events, latestSeq := buffer.poll(0) + if latestSeq != 2 { + t.Fatalf("latestSeq = %d, want 2", latestSeq) + } + if len(events) != 2 { + t.Fatalf("len(events) = %d, want 2", len(events)) + } + if events[0].Seq != 1 || events[0].Type != "spoofing_attempt" || events[0].PeerID != "peer-a" { + t.Fatalf("unexpected first event: %+v", events[0]) + } + if events[0].Timestamp == 0 { + t.Fatal("timestamp not set") + } +} + +func TestNodeEventBufferCursor(t *testing.T) { + buffer := newNodeEventBuffer() + for i := 0; i < 5; i++ { + buffer.record("security", "stale_event", "peer-a", "stale") + } + + events, latestSeq := buffer.poll(3) + if latestSeq != 5 { + t.Fatalf("latestSeq = %d, want 5", latestSeq) + } + if len(events) != 2 || events[0].Seq != 4 || events[1].Seq != 5 { + t.Fatalf("unexpected events after cursor 3: %+v", events) + } + + // Re-reads are idempotent. + again, _ := buffer.poll(3) + if len(again) != 2 { + t.Fatalf("re-read returned %d events, want 2", len(again)) + } +} + +func TestNodeEventBufferWrap(t *testing.T) { + buffer := newNodeEventBuffer() + for i := 0; i < 1500; i++ { + buffer.record("security", "rate_limit_drop", "peer-a", "dropped") + } + + events, latestSeq := buffer.poll(0) + if latestSeq != 1500 { + t.Fatalf("latestSeq = %d, want 1500", latestSeq) + } + if len(events) != 1000 { + t.Fatalf("len(events) = %d, want 1000", len(events)) + } + if events[0].Seq != 501 || events[999].Seq != 1500 { + t.Fatalf("wrap kept wrong window: first=%d last=%d", events[0].Seq, events[999].Seq) + } +} + +func TestNodeEventBufferConcurrent(t *testing.T) { + buffer := newNodeEventBuffer() + var waitGroup sync.WaitGroup + for worker := 0; worker < 8; worker++ { + waitGroup.Add(1) + go func() { + defer waitGroup.Done() + for i := 0; i < 200; i++ { + buffer.record("security", "rate_limit_drop", "peer-a", "dropped") + buffer.poll(0) + } + }() + } + waitGroup.Wait() + + _, latestSeq := buffer.poll(0) + if latestSeq != 1600 { + t.Fatalf("latestSeq = %d, want 1600", latestSeq) + } +} diff --git a/internal/node/mcp.go b/internal/node/mcp.go index f2bd8d19..fed33a53 100644 --- a/internal/node/mcp.go +++ b/internal/node/mcp.go @@ -153,6 +153,18 @@ func NewMCPServer(node *SamNode) *mcp.Server { Description: "Returns the last few lines of the node's log output.", }, node.handleGetRecentLogs) + // Add the poll_node_events tool. + mcp.AddTool(mcpServer, &mcp.Tool{ + Name: "poll_node_events", + Description: "Poll typed node events (mesh events, security events) with cursor semantics. Pass since_seq from the previous response; 0 returns everything buffered.", + }, node.handlePollNodeEvents) + + // Add the get_control_plane_info tool. + mcp.AddTool(mcpServer, &mcp.Tool{ + Name: "get_control_plane_info", + Description: "Fetch the control plane's public /info and /keys endpoints; returns active router addresses and valid signing keys as JSON.", + }, node.handleGetControlPlaneInfo) + return mcpServer } diff --git a/internal/node/mcp_handlers.go b/internal/node/mcp_handlers.go index 030ce60e..54442021 100644 --- a/internal/node/mcp_handlers.go +++ b/internal/node/mcp_handlers.go @@ -2,6 +2,7 @@ package node import ( "context" + "encoding/base64" "encoding/json" "errors" "fmt" @@ -204,6 +205,7 @@ func (n *SamNode) handleGetMeshInfo(ctx context.Context, req *mcp.CallToolReques dhtSize := n.DHT.RoutingTable().Size() resData := map[string]any{ + "peer_id": n.Host.ID().String(), "connected_peers": connectedPeers, "dht_size": dhtSize, "router_peer_id": n.RouterPeerID.String(), @@ -896,3 +898,59 @@ func (n *SamNode) handleGetRecentLogs(ctx context.Context, req *mcp.CallToolRequ Content: []mcp.Content{&mcp.TextContent{Text: string(data)}}, }, nil, nil } + +// PollNodeEventsParams defines parameters for the poll_node_events tool. +type PollNodeEventsParams struct { + SinceSeq uint64 `json:"since_seq,omitempty" jsonschema:"Return events with seq greater than this. 0 or omitted returns everything buffered."` +} + +// handlePollNodeEvents implements the poll_node_events tool. +func (n *SamNode) handlePollNodeEvents(ctx context.Context, req *mcp.CallToolRequest, params PollNodeEventsParams) (*mcp.CallToolResult, any, error) { + events, latestSeq := globalEventBuffer.poll(params.SinceSeq) + data, err := json.Marshal(map[string]any{"events": events, "latest_seq": latestSeq}) + if err != nil { + return nil, nil, err + } + return &mcp.CallToolResult{ + Content: []mcp.Content{&mcp.TextContent{Text: string(data)}}, + }, nil, nil +} + +// GetControlPlaneInfoParams defines parameters for the get_control_plane_info tool. +type GetControlPlaneInfoParams struct{} + +// handleGetControlPlaneInfo implements the get_control_plane_info tool. +func (n *SamNode) handleGetControlPlaneInfo(ctx context.Context, req *mcp.CallToolRequest, params GetControlPlaneInfoParams) (*mcp.CallToolResult, any, error) { + controlPlaneURL, err := n.Store.LoadControlPlaneURL() + if err != nil { + return nil, nil, fmt.Errorf("failed to load control plane URL: %w", err) + } + if controlPlaneURL == "" { + return nil, nil, fmt.Errorf("no control plane URL stored; is the node enrolled?") + } + + info, err := FetchControlPlaneInfo(ctx, controlPlaneURL) + if err != nil { + return nil, nil, err + } + keys, err := FetchControlPlaneKeys(ctx, controlPlaneURL) + if err != nil { + return nil, nil, err + } + + signingKeys := []string{} + for _, key := range keys.PublicKeys { + signingKeys = append(signingKeys, base64.StdEncoding.EncodeToString(key)) + } + data, err := json.Marshal(map[string]any{ + "control_plane_url": controlPlaneURL, + "router_addresses": info.RouterAddresses, + "signing_keys": signingKeys, + }) + if err != nil { + return nil, nil, err + } + return &mcp.CallToolResult{ + Content: []mcp.Content{&mcp.TextContent{Text: string(data)}}, + }, nil, nil +} diff --git a/internal/node/mcp_handlers_test.go b/internal/node/mcp_handlers_test.go index 9b345725..fec3dfc1 100644 --- a/internal/node/mcp_handlers_test.go +++ b/internal/node/mcp_handlers_test.go @@ -1064,3 +1064,30 @@ func newPreDiscoverMCPHandler(t *testing.T, tools []*mcp.Tool) http.Handler { real.ServeHTTP(w, r) }) } + +func TestHandlePollNodeEvents(t *testing.T) { + globalEventBuffer = newNodeEventBuffer() + RecordNodeEvent("security", "spoofing_attempt", "peer-a", "invalid signature") + RecordNodeEvent("mesh_event", "banned", "peer-b", "peer banned by hub") + + node := &SamNode{} + result, _, err := node.handlePollNodeEvents(context.Background(), nil, PollNodeEventsParams{SinceSeq: 1}) + if err != nil { + t.Fatalf("handler error: %v", err) + } + + text := result.Content[0].(*mcp.TextContent).Text + var response struct { + Events []NodeEvent `json:"events"` + LatestSeq uint64 `json:"latest_seq"` + } + if err := json.Unmarshal([]byte(text), &response); err != nil { + t.Fatalf("bad JSON: %v", err) + } + if response.LatestSeq != 2 { + t.Fatalf("latest_seq = %d, want 2", response.LatestSeq) + } + if len(response.Events) != 1 || response.Events[0].Type != "banned" { + t.Fatalf("unexpected events: %+v", response.Events) + } +} diff --git a/internal/node/node.go b/internal/node/node.go index c079da7a..41502e26 100644 --- a/internal/node/node.go +++ b/internal/node/node.go @@ -1123,6 +1123,7 @@ func (n *SamNode) listenForControlPlaneEvents(ctx context.Context) { if !n.rateLimiter.Allow(msg.ReceivedFrom.String()) { logger.Warnf("[Mesh Event] Rate limit exceeded for %s, dropping message", msg.ReceivedFrom) + RecordNodeEvent(EventCategorySecurity, EventTypeRateLimitDrop, msg.ReceivedFrom.String(), "rate limit exceeded, message dropped") continue } @@ -1140,6 +1141,7 @@ func (n *SamNode) listenForControlPlaneEvents(ctx context.Context) { if !n.verifyEvent(&event) { logger.Warnf("[Mesh Event] Potential spoofing attempt: invalid signature on event from %s", msg.ReceivedFrom) + RecordNodeEvent(EventCategorySecurity, EventTypeSpoofingAttempt, msg.ReceivedFrom.String(), "invalid event signature") continue } @@ -1147,15 +1149,19 @@ func (n *SamNode) listenForControlPlaneEvents(ctx context.Context) { eventTime := time.UnixMilli(event.Timestamp) if time.Since(eventTime) > FreshnessThreshold || time.Until(eventTime) > FreshnessThreshold { logger.Warnf("[Mesh Event] Dropping stale or future event from %s (timestamp: %d)", msg.ReceivedFrom, event.Timestamp) + RecordNodeEvent(EventCategorySecurity, EventTypeStaleEvent, msg.ReceivedFrom.String(), "stale or future event dropped") continue } switch event.Type { case api.MeshEvent_BANNED: + RecordNodeEvent(EventCategoryMesh, EventTypeBanned, event.PeerId, "peer banned by hub") n.handleBannedEvent(&event) case api.MeshEvent_KEY_ROTATION: + RecordNodeEvent(EventCategoryMesh, EventTypeKeyRotation, event.PeerId, "hub key rotation") n.handleKeyRotationEvent(&event) case api.MeshEvent_POLICY_UPDATE: + RecordNodeEvent(EventCategoryMesh, EventTypePolicyUpdate, event.PeerId, "mesh policy update received") logger.Infof("[Mesh Event] Received POLICY_UPDATE event from %s, triggering sync", msg.ReceivedFrom) go func() { maxJitter := n.config.PolicySyncJitter diff --git a/sam-mcp-python/src/sam_mcp/client.py b/sam-mcp-python/src/sam_mcp/client.py index 7043d1bc..dfe0cb9b 100644 --- a/sam-mcp-python/src/sam_mcp/client.py +++ b/sam-mcp-python/src/sam_mcp/client.py @@ -3,6 +3,7 @@ from typing import Any, Dict, List, Optional from mcp import ClientSession from mcp.client.streamable_http import streamable_http_client +from mcp.shared._httpx_utils import create_mcp_http_client class SamClient: """High-level developer interface for SAM MCP using official SDK.""" @@ -19,12 +20,12 @@ def __init__(self, server_url: Optional[str] = None, token: Optional[str] = None async def connect(self): """Connects to the SAM node via Streamable HTTP.""" - import httpx headers = {"Accept": "application/json, text/event-stream"} if self.token: headers["X-Sam-Authentication"] = f"Bearer {self.token}" - - self._http_client = httpx.AsyncClient(headers=headers) + + # Applies MCP's SSE-friendly timeouts; a plain httpx client reads for 5s and drops the stream. + self._http_client = create_mcp_http_client(headers=headers) try: self._sh_cm = streamable_http_client(self.server_url, http_client=self._http_client) res = await self._sh_cm.__aenter__()