popov/neuron-tai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

validator edits

Browse Source
This commit is contained in:
Dobromir Popov
2026-07-04 23:03:03 +02:00
parent 69b0e726b8
commit 9bd15de65b
6 changed files with 425 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

171
packages/tracker/meshnet_tracker/server.py
View File

@@ -37,6 +37,7 @@ from importlib.resources import files
from typing import Any
from .accounts import DEFAULT_ACCOUNTS_DB_PATH, AccountStore
from .auth import is_validator_token, sign_hive_request, verify_hive_request
from .billing import DEFAULT_BILLING_DB_PATH, BillingLedger
from .gossip import NodeGossip
from .raft import RaftNode
@@ -1264,6 +1265,8 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
billing: "BillingLedger | None" = None,
accounts: "AccountStore | None" = None,
benchmark_results_path: str | None = None,
validator_service_token: str | None = None,
hive_secret: str | None = None,
) -> None:
super().__init__(addr, handler)
self.registry = registry
@@ -1282,6 +1285,8 @@ class _TrackerHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
os.getcwd(), "benchmark_results.json"
)
self.benchmark_lock = threading.Lock()
self.validator_service_token = validator_service_token
self.hive_secret = hive_secret
class _TrackerHandler(http.server.BaseHTTPRequestHandler):
@@ -1299,6 +1304,65 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
except BrokenPipeError:
pass
# ---- unified auth boundary (ADR-0017) ----
def _resolve_identity(self) -> tuple[str | None, dict | None]:
"""Resolve the caller to (role, account).
Roles: "validator" (service token), "admin"/"user" (session token).
Client API keys resolve to no privileged role — they authorize
inference and wallet binding only, never operator endpoints.
"""
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
token = _api_key_from_headers(self.headers)
if not token:
return None, None
if is_validator_token(token, server.validator_service_token):
return "validator", None
if server.accounts is not None:
account = server.accounts.session_account(token)
if account is not None:
return account.get("role", "user"), account
return None, None
def _require_role(self, *allowed: str) -> bool:
"""Gate a privileged handler; sends 401/403 and returns False on failure.
401 when no credential was presented; 403 when a credential was
presented but does not resolve to an allowed role — this covers
client API keys and garbage bearer strings on operator endpoints.
"""
role, _account = self._resolve_identity()
if role in allowed:
return True
if _api_key_from_headers(self.headers) is None:
self._send_json(401, {"error": "authentication required (admin session or service token)"})
else:
self._send_json(403, {"error": "this endpoint requires an admin session or service token"})
return False
def _read_hive_authenticated_body(self) -> dict | None:
"""Read + verify a hive gossip body (HMAC per ADR-0017 §3).
Fails closed: without a configured --hive-secret no gossip is
accepted. Sends the error response itself and returns None on failure.
"""
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
length = int(self.headers.get("Content-Length", 0))
raw = self.rfile.read(length) if length else b"{}"
if not verify_hive_request(server.hive_secret, self.headers, raw):
self._send_json(401, {"error": "valid hive signature required"})
return None
try:
body = json.loads(raw or b"{}")
except json.JSONDecodeError:
self._send_json(400, {"error": "invalid JSON body"})
return None
if not isinstance(body, dict):
self._send_json(400, {"error": "JSON body must be an object"})
return None
return body
def _read_json_body(self) -> dict | None:
length = int(self.headers.get("Content-Length", 0))
try:
@@ -2354,7 +2418,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
def _handle_stats_gossip(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
body = self._read_json_body()
body = self._read_hive_authenticated_body()
if body is None:
return
tracker_url = body.get("tracker_url", "")
@@ -2365,6 +2429,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
def _handle_billing_summary(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if not self._require_role("admin"):
return
if server.billing is None:
self._send_json(404, {"error": "billing is not enabled on this tracker"})
return
@@ -2390,6 +2456,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
def _handle_registry_wallets(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if not self._require_role("admin"):
return
if server.contracts is None:
self._send_json(200, {"wallets": {}})
return
@@ -2406,6 +2474,8 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
def _handle_billing_settlements(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if not self._require_role("admin"):
return
if server.billing is None:
self._send_json(404, {"error": "billing is not enabled on this tracker"})
return
@@ -2413,7 +2483,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
def _handle_billing_gossip(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
body = self._read_json_body()
body = self._read_hive_authenticated_body()
if body is None:
return
if server.billing is None:
@@ -2429,12 +2499,11 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
def _handle_billing_forfeit(self):
"""Privileged: forfeit a node's pending balance + record a strike (US-034).
Auth is a header-presence stub (non-empty Authorization), matching the
benchmark endpoints — real validator auth arrives with on-chain keys.
ADR-0017 §4: validator service token or admin session only. Client
API keys — valid or not — are rejected.
"""
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if not self.headers.get("Authorization"):
self._send_json(401, {"error": "Authorization header required"})
if not self._require_role("validator", "admin"):
return
if server.billing is None:
self._send_json(404, {"error": "billing is not enabled on this tracker"})
@@ -2609,7 +2678,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
def _handle_accounts_gossip(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
body = self._read_json_body()
body = self._read_hive_authenticated_body()
if body is None:
return
if server.accounts is None:
@@ -2655,10 +2724,9 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
is its total minus the (k-1)-node route's total.
"""
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
auth = self.headers.get("Authorization")
if not auth:
self._send_json(401, {"error": "Authorization header required"})
if not self._require_role("admin", "validator"):
return
auth = self.headers.get("Authorization")
body = self._read_json_body()
if body is None:
return
@@ -2744,8 +2812,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler):
def _handle_benchmark_results(self):
server: _TrackerHTTPServer = self.server # type: ignore[assignment]
if not self.headers.get("Authorization"):
self._send_json(401, {"error": "Authorization header required"})
if not self._require_role("admin", "validator"):
return
results: list = []
with server.benchmark_lock:
@@ -3198,6 +3265,8 @@ class TrackerServer:
payout_threshold: float = 5.0,
payout_dust_floor: float = 0.01,
settlement_check_interval: float | None = None,
validator_service_token: str | None = None,
hive_secret: str | None = None,
) -> None:
self._host = host
self._requested_port = port
@@ -3256,6 +3325,16 @@ class TrackerServer:
)
self._settlement_stop = threading.Event()
self._settlement_thread: threading.Thread | None = None
self._validator_service_token = (
validator_service_token
if validator_service_token is not None
else os.environ.get("MESHNET_VALIDATOR_SERVICE_TOKEN") or None
)
self._hive_secret = (
hive_secret
if hive_secret is not None
else os.environ.get("MESHNET_HIVE_SECRET") or None
)
self.port: int | None = None
def start(self) -> int:
@@ -3282,6 +3361,8 @@ class TrackerServer:
billing=self._billing,
accounts=self._accounts,
benchmark_results_path=self._benchmark_results_path,
validator_service_token=self._validator_service_token,
hive_secret=self._hive_secret,
)
self.port = self._server.server_address[1]
@@ -3450,6 +3531,23 @@ class TrackerServer:
_purge_expired_nodes_locked(server)
_rebalance_all_locked(server)
def _push_to_peers(self, path: str, body: bytes) -> bool:
"""POST a hive-signed payload to every cluster peer; True if all succeeded."""
headers = {"Content-Type": "application/json"}
if self._hive_secret:
headers.update(sign_hive_request(self._hive_secret, body))
delivered_all = True
for peer in self._cluster_peers:
try:
req = urllib.request.Request(
f"{peer}{path}", data=body, headers=headers, method="POST",
)
with urllib.request.urlopen(req, timeout=2.0) as r:
r.read()
except Exception:
delivered_all = False
return delivered_all
def _stats_loop(self) -> None:
"""Periodically save stats/billing to DB and push local slices to cluster peers."""
while not self._stats_stop.wait(_StatsCollector.SAVE_INTERVAL):
@@ -3459,62 +3557,31 @@ class TrackerServer:
self._billing.save_to_db()
if self._accounts is not None:
self._accounts.save_to_db()
if self._cluster_peers and not self._hive_secret:
print(
"[tracker] WARNING: cluster peers configured without --hive-secret — "
"gossip pushes will be rejected (ADR-0017)",
flush=True,
)
if self._stats is not None and self._cluster_peers:
self_url = self._cluster_self_url or f"http://{self._host}:{self.port}"
local_rpms = self._stats.get_local_rpms()
body = json.dumps({"tracker_url": self_url, "stats": local_rpms}).encode()
for peer in self._cluster_peers:
try:
req = urllib.request.Request(
f"{peer}/v1/stats/gossip",
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=2.0) as r:
r.read()
except Exception:
pass
self._push_to_peers("/v1/stats/gossip", body)
if self._billing is not None and self._cluster_peers:
events, cursor = self._billing.events_since(self._billing_gossip_cursor)
if events:
delivered_all = True
body = json.dumps({"events": events}).encode()
for peer in self._cluster_peers:
try:
req = urllib.request.Request(
f"{peer}/v1/billing/gossip",
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=2.0) as r:
r.read()
except Exception:
delivered_all = False
# Only advance past events every peer has seen; unreachable
# peers get the full backlog on the next tick (idempotent
# via event-id dedupe on the receiving side).
if delivered_all:
if self._push_to_peers("/v1/billing/gossip", body):
self._billing_gossip_cursor = cursor
if self._accounts is not None and self._cluster_peers:
events, cursor = self._accounts.events_since(self._accounts_gossip_cursor)
if events:
delivered_all = True
body = json.dumps({"events": events}).encode()
for peer in self._cluster_peers:
try:
req = urllib.request.Request(
f"{peer}/v1/accounts/gossip",
data=body,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=2.0) as r:
r.read()
except Exception:
delivered_all = False
if delivered_all:
if self._push_to_peers("/v1/accounts/gossip", body):
self._accounts_gossip_cursor = cursor
def stop(self) -> None: