From 39f6f23c839c61c1b06ced89d313bc22cdc75dd6 Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Mon, 29 Jun 2026 09:46:22 +0300 Subject: [PATCH] feat: add fraud detection validator --- .../distributed-inference-network/prd.json | 5 +- conftest.py | 1 + .../contracts/meshnet_contracts/__init__.py | 147 ++++++++++++++++ packages/gateway/meshnet_gateway/server.py | 10 +- packages/node/meshnet_node/server.py | 18 +- .../validator/meshnet_validator/__init__.py | 159 ++++++++++++++++++ packages/validator/pyproject.toml | 12 ++ tests/test_contracts_settlement.py | 20 +++ tests/test_fraud_detection.py | 145 ++++++++++++++++ 9 files changed, 511 insertions(+), 6 deletions(-) create mode 100644 packages/validator/meshnet_validator/__init__.py create mode 100644 packages/validator/pyproject.toml create mode 100644 tests/test_fraud_detection.py diff --git a/.scratch/distributed-inference-network/prd.json b/.scratch/distributed-inference-network/prd.json index 240e520..1535ec2 100644 --- a/.scratch/distributed-inference-network/prd.json +++ b/.scratch/distributed-inference-network/prd.json @@ -171,12 +171,13 @@ "Commit only this story's code changes with a focused conventional commit message" ], "priority": 7, - "passes": false, + "passes": true, "notes": "Source issue: .scratch/distributed-inference-network/issues/07-fraud-detection-slash.md", "dependsOn": [ "US-005", "US-006" - ] + ], + "completionNotes": "Completed by fresh Ralph/Codex session 04475912 with controller fix for duplicate slash suppression; verified by pytest, compileall, and diff check." }, { "id": "US-008", diff --git a/conftest.py b/conftest.py index ffb9a89..617c147 100644 --- a/conftest.py +++ b/conftest.py @@ -11,6 +11,7 @@ _packages = [ "packages/sdk", "packages/contracts", "packages/p2p", + "packages/validator", ] for _pkg in _packages: diff --git a/packages/contracts/meshnet_contracts/__init__.py b/packages/contracts/meshnet_contracts/__init__.py index 4570183..23e12e2 100644 --- a/packages/contracts/meshnet_contracts/__init__.py +++ b/packages/contracts/meshnet_contracts/__init__.py @@ -6,6 +6,8 @@ replace this adapter later without changing gateway or tracker call sites. """ from dataclasses import dataclass, field +import json +import urllib.request __version__ = "0.1.0" @@ -49,6 +51,33 @@ class ComputeAttribution: return self.layer_count * self.tokens +@dataclass(frozen=True) +class ValidationEvent: + """Completed inference data consumed by fraud validators.""" + + index: int + session_id: str + model: str + messages: list[dict] + observed_output: str + route_nodes: list[dict] + + +@dataclass(frozen=True) +class SlashReceipt: + """Local slash transaction receipt.""" + + signature: str + cluster: str + wallet_address: str + slash_amount: int + stake_before: int + stake_after: int + strike_count: int + banned: bool + reason: str + + @dataclass(frozen=True) class SettlementResult: """Summary returned by an epoch settlement transaction.""" @@ -64,6 +93,8 @@ class _LocalContractState: registry: dict[str, RegistryWallet] = field(default_factory=dict) api_keys: dict[str, ApiKeyBalance] = field(default_factory=dict) attributions: list[ComputeAttribution] = field(default_factory=list) + validation_events: list[ValidationEvent] = field(default_factory=list) + slash_receipts: list[SlashReceipt] = field(default_factory=list) token_balances: dict[str, int] = field(default_factory=dict) @@ -98,6 +129,46 @@ class RegistryContract: ) return {"signature": f"local-strike-{wallet_address}", "cluster": self._cluster} + def submit_slash_proof( + self, + *, + wallet_address: str, + slash_amount: int, + strike_threshold: int = 3, + reason: str = "fraudulent inference output", + webhook_url: str | None = None, + ) -> SlashReceipt: + if not wallet_address: + raise ValueError("wallet_address is required") + if slash_amount <= 0: + raise ValueError("slash_amount must be positive") + if strike_threshold <= 0: + raise ValueError("strike_threshold must be positive") + + current = self.get_wallet(wallet_address) + stake_after = max(0, current.stake_balance - slash_amount) + strike_count = current.strike_count + 1 + banned = current.banned or strike_count >= strike_threshold + self._state.registry[wallet_address] = RegistryWallet( + stake_balance=stake_after, + strike_count=strike_count, + banned=banned, + ) + receipt = SlashReceipt( + signature=f"local-slash-{wallet_address}-{len(self._state.slash_receipts) + 1}", + cluster=self._cluster, + wallet_address=wallet_address, + slash_amount=slash_amount, + stake_before=current.stake_balance, + stake_after=stake_after, + strike_count=strike_count, + banned=banned, + reason=reason, + ) + self._state.slash_receipts.append(receipt) + _notify_slash(receipt, webhook_url) + return receipt + def ban_wallet(self, wallet_address: str) -> dict: current = self.get_wallet(wallet_address) self._state.registry[wallet_address] = RegistryWallet( @@ -114,6 +185,49 @@ class RegistryContract: return wallet.stake_balance >= minimum_stake +class ValidationContract: + """Validation event log consumed by the optimistic fraud detector.""" + + def __init__(self, state: _LocalContractState) -> None: + self._state = state + + def record_completed_inference( + self, + *, + session_id: str, + model: str, + messages: list[dict], + observed_output: str, + route_nodes: list[dict], + ) -> ValidationEvent: + if not session_id: + raise ValueError("session_id is required") + if not model: + raise ValueError("model is required") + if not isinstance(messages, list): + raise ValueError("messages must be a list") + if not isinstance(observed_output, str): + raise ValueError("observed_output must be a string") + if not isinstance(route_nodes, list): + raise ValueError("route_nodes must be a list") + event = ValidationEvent( + index=len(self._state.validation_events), + session_id=session_id, + model=model, + messages=[dict(message) for message in messages], + observed_output=observed_output, + route_nodes=[dict(node) for node in route_nodes], + ) + self._state.validation_events.append(event) + return event + + def list_completed_inferences(self, *, after_index: int = -1) -> list[ValidationEvent]: + return [ + event for event in self._state.validation_events + if event.index > after_index + ] + + class PaymentContract: """Payment wrapper for funded API keys and compute attribution.""" @@ -297,6 +411,7 @@ class LocalSolanaContracts: self.cluster = cluster self._state = _LocalContractState() self.registry = RegistryContract(self._state, cluster) + self.validation = ValidationContract(self._state) self.payment = PaymentContract(self._state, cluster, starting_credit_lamports) self.settlement = SettlementContract( self._state, @@ -305,6 +420,35 @@ class LocalSolanaContracts: ) +def _notify_slash(receipt: SlashReceipt, webhook_url: str | None) -> None: + message = ( + f"WARNING: node {receipt.wallet_address} slashed " + f"by {receipt.slash_amount} lamports; strikes={receipt.strike_count}; " + f"banned={receipt.banned}; reason={receipt.reason}" + ) + print(message, flush=True) + if not webhook_url: + return + payload = json.dumps({ + "event": "node_slashed", + "wallet_address": receipt.wallet_address, + "slash_amount": receipt.slash_amount, + "strike_count": receipt.strike_count, + "banned": receipt.banned, + "reason": receipt.reason, + }).encode() + req = urllib.request.Request( + webhook_url, + data=payload, + headers={"Content-Type": "application/json"}, + method="POST", + ) + try: + urllib.request.urlopen(req, timeout=2).read() + except OSError: + pass + + __all__ = [ "ApiKeyBalance", "ComputeAttribution", @@ -314,4 +458,7 @@ __all__ = [ "RegistryWallet", "SettlementContract", "SettlementResult", + "SlashReceipt", + "ValidationContract", + "ValidationEvent", ] diff --git a/packages/gateway/meshnet_gateway/server.py b/packages/gateway/meshnet_gateway/server.py index 79750a3..070d11a 100644 --- a/packages/gateway/meshnet_gateway/server.py +++ b/packages/gateway/meshnet_gateway/server.py @@ -147,10 +147,10 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler): text = node_resp["text"] model = body.get("model", "stub-model") + session_id = str(uuid.uuid4()) if server.contracts is not None and route_nodes: api_key = _api_key_from_authorization(self.headers.get("Authorization")) if api_key: - session_id = str(uuid.uuid4()) tokens = max(1, _estimate_token_count(messages, text)) for node in route_nodes: wallet_address = node.get("wallet_address") @@ -165,6 +165,14 @@ class _GatewayHandler(http.server.BaseHTTPRequestHandler): tokens=tokens, speed_score=float(node.get("score", 1.0)), ) + if server.contracts is not None: + server.contracts.validation.record_completed_inference( + session_id=session_id, + model=model, + messages=messages if isinstance(messages, list) else [], + observed_output=text, + route_nodes=route_nodes, + ) if streaming: self._send_sse_response(text, model) diff --git a/packages/node/meshnet_node/server.py b/packages/node/meshnet_node/server.py index d01512c..49d1346 100644 --- a/packages/node/meshnet_node/server.py +++ b/packages/node/meshnet_node/server.py @@ -28,11 +28,20 @@ def _make_stub_activations(prompt: str) -> dict: class _StubHTTPServer(http.server.HTTPServer): - def __init__(self, addr, handler, shard_start: int, shard_end: int, is_last_shard: bool): + def __init__( + self, + addr, + handler, + shard_start: int, + shard_end: int, + is_last_shard: bool, + response_prefix: str, + ): super().__init__(addr, handler) self.shard_start = shard_start self.shard_end = shard_end self.is_last_shard = is_last_shard + self.response_prefix = response_prefix self.received_activations: bool = False @@ -57,7 +66,7 @@ class _StubHandler(http.server.BaseHTTPRequestHandler): server.received_activations = True prompt = body["activations"].get("context", {}).get("prompt", "") if server.is_last_shard: - payload = json.dumps({"text": f"stub response to: {prompt}"}).encode() + payload = json.dumps({"text": f"{server.response_prefix} {prompt}"}).encode() else: payload = json.dumps({"activations": _make_stub_activations(prompt)}).encode() else: @@ -66,7 +75,7 @@ class _StubHandler(http.server.BaseHTTPRequestHandler): last_content = messages[-1]["content"] if messages else "" if server.is_last_shard: # Single-node shortcut: no activation hop needed. - payload = json.dumps({"text": f"stub response to: {last_content}"}).encode() + payload = json.dumps({"text": f"{server.response_prefix} {last_content}"}).encode() else: payload = json.dumps({"activations": _make_stub_activations(last_content)}).encode() @@ -92,6 +101,7 @@ class StubNodeServer: shard_start: int = 0, shard_end: int = 31, is_last_shard: bool = True, + response_prefix: str = "stub response to:", ): self._host = host self._requested_port = port @@ -100,6 +110,7 @@ class StubNodeServer: self._shard_start = shard_start self._shard_end = shard_end self._is_last_shard = is_last_shard + self._response_prefix = response_prefix self._server: _StubHTTPServer | None = None self._thread: threading.Thread | None = None self.port: int | None = None @@ -119,6 +130,7 @@ class StubNodeServer: self._shard_start, self._shard_end, self._is_last_shard, + self._response_prefix, ) self.port = self._server.server_address[1] self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) diff --git a/packages/validator/meshnet_validator/__init__.py b/packages/validator/meshnet_validator/__init__.py new file mode 100644 index 0000000..48b00f6 --- /dev/null +++ b/packages/validator/meshnet_validator/__init__.py @@ -0,0 +1,159 @@ +"""Optimistic fraud validator for completed inference requests.""" + +import json +import math +import random +import threading +import time +import urllib.request +from typing import Any + +__version__ = "0.1.0" + + +class ValidatorProcess: + """Separate validator loop that samples completed requests and submits slashes.""" + + def __init__( + self, + *, + contracts: Any, + reference_node_url: str, + sample_rate: float = 0.05, + tolerance: float = 1e-6, + slash_amount: int = 100, + strike_threshold: int = 3, + random_seed: int | None = None, + webhook_url: str | None = None, + interval_seconds: float = 1.0, + ) -> None: + if not 0.0 <= sample_rate <= 1.0: + raise ValueError("sample_rate must be between 0 and 1") + if tolerance < 0: + raise ValueError("tolerance must be non-negative") + if slash_amount <= 0: + raise ValueError("slash_amount must be positive") + if strike_threshold <= 0: + raise ValueError("strike_threshold must be positive") + if interval_seconds <= 0: + raise ValueError("interval_seconds must be positive") + + self._contracts = contracts + self._reference_node_url = reference_node_url.rstrip("/") + self._sample_rate = sample_rate + self._tolerance = tolerance + self._slash_amount = slash_amount + self._strike_threshold = strike_threshold + self._webhook_url = webhook_url + self._interval_seconds = interval_seconds + self._random = random.Random(random_seed) + self._last_event_index = -1 + self._running = False + self._thread: threading.Thread | None = None + self.sampled_count = 0 + + def validate_once(self) -> list[Any]: + """Run one validation cycle and return slash receipts submitted this cycle.""" + receipts: list[Any] = [] + events = self._contracts.validation.list_completed_inferences( + after_index=self._last_event_index, + ) + for event in events: + self._last_event_index = max(self._last_event_index, event.index) + if self._random.random() >= self._sample_rate: + continue + self.sampled_count += 1 + reference_output = self._run_reference(event.messages) + if _outputs_match(event.observed_output, reference_output, self._tolerance): + continue + receipts.extend(self._slash_route(event.route_nodes, event.observed_output, reference_output)) + return receipts + + def start(self) -> None: + if self._running: + raise RuntimeError("ValidatorProcess is already running") + self._running = True + self._thread = threading.Thread(target=self._run_loop, daemon=True) + self._thread.start() + + def stop(self) -> None: + self._running = False + if self._thread is not None: + self._thread.join(timeout=2) + self._thread = None + + def _run_loop(self) -> None: + while self._running: + self.validate_once() + time.sleep(self._interval_seconds) + + def _run_reference(self, messages: list[dict]) -> str: + response = _post_json( + f"{self._reference_node_url}/v1/infer", + {"messages": messages}, + ) + text = response.get("text") + if not isinstance(text, str): + raise ValueError("reference node response did not contain text") + return text + + def _slash_route( + self, + route_nodes: list[dict], + observed_output: str, + reference_output: str, + ) -> list[Any]: + receipts: list[Any] = [] + node = _final_text_node(route_nodes) + wallet_address = node.get("wallet_address") if node else None + if not wallet_address: + return receipts + if self._contracts.registry.get_wallet(wallet_address).banned: + return receipts + receipts.append(self._contracts.registry.submit_slash_proof( + wallet_address=wallet_address, + slash_amount=self._slash_amount, + strike_threshold=self._strike_threshold, + reason=( + "reference output diverged " + f"(observed={observed_output!r}, reference={reference_output!r})" + ), + webhook_url=self._webhook_url, + )) + return receipts + + +def _final_text_node(route_nodes: list[dict]) -> dict | None: + if not route_nodes: + return None + return max(route_nodes, key=lambda node: int(node.get("shard_end", 0))) + + +def _outputs_match(observed: str, reference: str, tolerance: float) -> bool: + observed_float = _parse_float(observed) + reference_float = _parse_float(reference) + if observed_float is not None and reference_float is not None: + return math.isclose(observed_float, reference_float, rel_tol=tolerance, abs_tol=tolerance) + return observed == reference + + +def _parse_float(value: str) -> float | None: + try: + return float(value) + except ValueError: + return None + + +def _post_json(url: str, payload: dict, timeout: float = 5.0) -> dict: + data = json.dumps(payload).encode() + req = urllib.request.Request( + url, + data=data, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=timeout) as response: + return json.loads(response.read()) + + +__all__ = ["ValidatorProcess"] diff --git a/packages/validator/pyproject.toml b/packages/validator/pyproject.toml new file mode 100644 index 0000000..be68e4e --- /dev/null +++ b/packages/validator/pyproject.toml @@ -0,0 +1,12 @@ +[build-system] +requires = ["setuptools>=64"] +build-backend = "setuptools.build_meta" + +[project] +name = "meshnet-validator" +version = "0.1.0" +description = "Optimistic fraud validator for the Distributed Inference Network" +requires-python = ">=3.10" + +[tool.setuptools.packages.find] +include = ["meshnet_validator*"] diff --git a/tests/test_contracts_settlement.py b/tests/test_contracts_settlement.py index 7af0a2c..a3d7547 100644 --- a/tests/test_contracts_settlement.py +++ b/tests/test_contracts_settlement.py @@ -59,3 +59,23 @@ def test_epoch_settlement_debits_clients_and_rewards_nodes_and_validator(): assert contracts.settlement.get_token_balance("node-wallet-a") == 96 assert contracts.settlement.get_token_balance("node-wallet-b") == 192 assert contracts.settlement.get_token_balance("validator-wallet") == 32 + + +def test_slash_proof_reduces_stake_increments_strikes_and_warns(capsys): + """Submitting a slash proof updates registry state and notifies the operator.""" + contracts = LocalSolanaContracts() + contracts.registry.submit_stake("node-wallet-a", 500) + + receipt = contracts.registry.submit_slash_proof( + wallet_address="node-wallet-a", + slash_amount=125, + strike_threshold=2, + reason="test divergence", + ) + + wallet = contracts.registry.get_wallet("node-wallet-a") + assert receipt.signature.startswith("local-slash-node-wallet-a") + assert wallet.stake_balance == 375 + assert wallet.strike_count == 1 + assert wallet.banned is False + assert "WARNING: node node-wallet-a slashed" in capsys.readouterr().out diff --git a/tests/test_fraud_detection.py b/tests/test_fraud_detection.py new file mode 100644 index 0000000..8dc00cc --- /dev/null +++ b/tests/test_fraud_detection.py @@ -0,0 +1,145 @@ +"""US-007 integration tests: optimistic fraud detection and slashing.""" + +import json +import urllib.error +import urllib.request + +from meshnet_contracts import LocalSolanaContracts +from meshnet_gateway.server import GatewayServer +from meshnet_node.server import StubNodeServer +from meshnet_tracker.server import TrackerServer +from meshnet_validator import ValidatorProcess + + +def _post_json(url: str, payload: dict, headers: dict | None = None) -> dict: + req = urllib.request.Request( + url, + data=json.dumps(payload).encode(), + headers={"Content-Type": "application/json", **(headers or {})}, + method="POST", + ) + with urllib.request.urlopen(req) as resp: + return json.loads(resp.read()) + + +def _register_node( + tracker_url: str, + endpoint: str, + shard_start: int, + shard_end: int, + wallet_address: str, +) -> None: + _post_json( + f"{tracker_url}/v1/nodes/register", + { + "endpoint": endpoint, + "shard_start": shard_start, + "shard_end": shard_end, + "hardware_profile": {}, + "wallet_address": wallet_address, + "score": 1.0, + }, + ) + + +def _send_completion(gateway_url: str, prompt: str) -> str: + body = _post_json( + f"{gateway_url}/v1/chat/completions", + { + "model": "stub-model", + "messages": [{"role": "user", "content": prompt}], + }, + {"Authorization": "Bearer fraud-test-key"}, + ) + return body["choices"][0]["message"]["content"] + + +def test_bad_node_is_slashed_and_excluded_from_gateway_routes(capsys): + """A bad final shard is slashed by the validator and then excluded by routing.""" + contracts = LocalSolanaContracts() + contracts.registry.submit_stake("wallet-good", 500) + contracts.registry.submit_stake("wallet-bad", 500) + tracker = TrackerServer(contracts=contracts) + tracker_port = tracker.start() + tracker_url = f"http://127.0.0.1:{tracker_port}" + + good_node = StubNodeServer(shard_start=0, shard_end=15, is_last_shard=False) + bad_node = StubNodeServer( + shard_start=16, + shard_end=31, + is_last_shard=True, + response_prefix="fraudulent response to:", + ) + reference_node = StubNodeServer(shard_start=0, shard_end=31, is_last_shard=True) + good_port = good_node.start() + bad_port = bad_node.start() + reference_port = reference_node.start() + + _register_node(tracker_url, f"http://127.0.0.1:{good_port}", 0, 15, "wallet-good") + _register_node(tracker_url, f"http://127.0.0.1:{bad_port}", 16, 31, "wallet-bad") + + gateway = GatewayServer(tracker_url=tracker_url, contracts=contracts) + gateway_port = gateway.start() + gateway_url = f"http://127.0.0.1:{gateway_port}" + validator = ValidatorProcess( + contracts=contracts, + reference_node_url=f"http://127.0.0.1:{reference_port}", + sample_rate=1.0, + slash_amount=125, + strike_threshold=1, + random_seed=7, + ) + + try: + for i in range(20): + assert _send_completion(gateway_url, f"prompt {i}").startswith("fraudulent response to:") + + receipts = validator.validate_once() + + bad_wallet = contracts.registry.get_wallet("wallet-bad") + assert receipts + assert bad_wallet.stake_balance == 375 + assert bad_wallet.strike_count == 1 + assert bad_wallet.banned is True + assert "WARNING: node wallet-bad slashed" in capsys.readouterr().out + + try: + _send_completion(gateway_url, "after slash") + raise AssertionError("Expected no route after bad wallet is excluded") + except urllib.error.HTTPError as exc: + assert exc.code == 503 + finally: + gateway.stop() + reference_node.stop() + bad_node.stop() + good_node.stop() + tracker.stop() + + +def test_validator_sampling_rate_is_configurable(): + """The validator only reruns requests selected by its configured sample rate.""" + contracts = LocalSolanaContracts() + + class InProcessReferenceValidator(ValidatorProcess): + def _run_reference(self, messages: list[dict]) -> str: + return f"stub response to: {messages[-1]['content']}" + + validator = InProcessReferenceValidator( + contracts=contracts, + reference_node_url="http://reference-node.invalid", + sample_rate=0.05, + random_seed=1, + ) + + for i in range(1_000): + contracts.validation.record_completed_inference( + session_id=f"session-{i}", + model="stub-model", + messages=[{"role": "user", "content": f"prompt {i}"}], + observed_output=f"stub response to: prompt {i}", + route_nodes=[{"wallet_address": "wallet-good"}], + ) + + validator.validate_once() + + assert 30 <= validator.sampled_count <= 70