"""hive — the swarm address book, and the fleet's config store.
A tiny standalone service for the main server: Navi instances across the
network announce themselves here (name, uuid, address, machine info), and
agents ask this book who is currently alive and where. It also stores the
central config: one versioned document whose layers are applied over each
instance's own config.
Not part of the Navi server — it never starts by default. Run it manually on
the main server (see hive/README.md):
.venv/bin/uvicorn hive.app:app --host 0.0.0.0 --port 8087
Two credentials:
- the shared swarm key (PSK) — announce/peers/fetch/ack. It is also the
ability to *read* config secrets, which is why the hive belongs on the
LAN, not on a public port;
- the admin key (``HIVE_ADMIN_KEY_FILE``) or a loopback connection —
publishing, history, rollback and status. Deliberately not the swarm key,
so a compromised instance cannot rewrite what the fleet runs.
"""
from __future__ import annotations
import asyncio
import hmac
import logging
import os
import time
from contextlib import asynccontextmanager
from pathlib import Path
import httpx
from fastapi import FastAPI, Header, HTTPException, Request
from pydantic import BaseModel, Field
from hive.config_store import (
ConfigError,
content_hash,
effective_hash,
effective_layers,
validate_layers,
)
from hive.db import HiveDB
log = logging.getLogger("hive")
_KEY_FILE = Path(os.environ.get("HIVE_KEY_FILE", ".swarm-key"))
# Rotation window: the previous key keeps working until it is deleted.
_PREVIOUS_KEY_FILE = _KEY_FILE.with_name(_KEY_FILE.name + ".previous")
_ADMIN_KEY_FILE = Path(os.environ.get("HIVE_ADMIN_KEY_FILE", ".hive-admin-key"))
_DB_PATH = os.environ.get("HIVE_DB", "hive.db")
_TTL_SEC = int(os.environ.get("HIVE_TTL_SEC", "300"))
_NOTIFY_TIMEOUT = float(os.environ.get("HIVE_NOTIFY_TIMEOUT_SEC", "3"))
_NOTIFY_RETRY_SEC = float(os.environ.get("HIVE_NOTIFY_RETRY_SEC", "300"))
_NOTIFY_SCHEME = os.environ.get("HIVE_NOTIFY_SCHEME", "http")
# ── credentials ─────────────────────────────────────────────────────────────
def _load_keys() -> list[str]:
keys: list[str] = []
for p in (_KEY_FILE, _PREVIOUS_KEY_FILE):
try:
value = p.read_text(encoding="utf-8").strip()
except OSError:
continue
if value:
keys.append(value)
return keys
def _check_key(x_swarm_key: str | None) -> None:
keys = _load_keys()
if not keys:
raise HTTPException(status_code=503, detail="hive has no swarm key configured")
if not x_swarm_key:
raise HTTPException(status_code=401, detail="X-Swarm-Key header required")
if not any(hmac.compare_digest(x_swarm_key, k) for k in keys):
raise HTTPException(status_code=403, detail="invalid swarm key")
def _admin_key() -> str:
# Read on every call, not at import: the key can be installed while the
# hive runs, and reading it here keeps tests honest.
value = os.environ.get("HIVE_ADMIN_KEY", "")
if value.strip():
return value.strip()
try:
return _ADMIN_KEY_FILE.read_text(encoding="utf-8").strip()
except OSError:
return ""
def _check_admin(request: Request, x_admin_key: str | None) -> str:
"""Return who is acting, for the version's ``created_by``."""
expected = _admin_key()
if x_admin_key:
if expected and hmac.compare_digest(x_admin_key, expected):
return "admin-key"
raise HTTPException(status_code=403, detail="invalid admin key")
peer = request.client.host if request.client else ""
if (
not expected
and os.environ.get("HIVE_ALLOW_LOCAL_ADMIN", "1") != "0"
and peer in {"127.0.0.1", "::1", "localhost"}
):
return f"local:{peer}"
raise HTTPException(
status_code=403,
detail=(
"publishing requires X-Admin-Key (HIVE_ADMIN_KEY_FILE) "
"or a loopback connection"
),
)
# ── app ─────────────────────────────────────────────────────────────────────
_db = HiveDB(_DB_PATH)
def _reported_config(meta: dict | None) -> dict | None:
config = (meta or {}).get("config")
return config if isinstance(config, dict) else None
def _is_behind(reported: dict | None, effective: str) -> bool:
"""Is this instance running something other than the current layers?
``False`` when the instance does not report config at all (an older
build) — that is "unknown", and re-notifying it forever helps nobody.
"""
if reported is None:
return False
if reported.get("status") == "disabled":
return False
return reported.get("hash") != effective
def _identity_of(peer: dict) -> tuple[str, str, list[str]]:
meta = peer.get("meta") or {}
tags = meta.get("tags") or []
return peer.get("name") or "", peer.get("instance_id") or "", list(tags)
async def _notify_one(peer: dict, version: int, doc_hash: str) -> bool:
host, port = peer.get("host"), peer.get("port")
if not host or not port:
return False
keys = _load_keys()
if not keys:
return False
url = f"{_NOTIFY_SCHEME}://{host}:{port}/config/notify"
try:
async with httpx.AsyncClient(timeout=_NOTIFY_TIMEOUT) as client:
resp = await client.post(
url,
json={"version": version, "hash": doc_hash},
headers={"X-Swarm-Key": keys[0]},
)
return resp.status_code < 300
except Exception as exc:
log.info("config.notify_failed instance=%s error=%s", peer.get("name"), exc)
return False
async def _notify_all(version: int, doc_hash: str) -> None:
"""Push a 'config changed' ping to every registered instance.
Best effort by design: the announce reply carries the same hint, so an
instance that is offline (or behind a proxy the hive cannot reach) learns
about the change on its next heartbeat. The hive never needs this to
succeed for the config to be correct.
"""
peers = _db.list_peers(_TTL_SEC)
results = await asyncio.gather(
*(_notify_one(peer, version, doc_hash) for peer in peers),
return_exceptions=True,
)
reached = sum(1 for r in results if r is True)
log.info(
"config.notified version=%s reached=%s total=%s", version, reached, len(peers)
)
async def _notify_behind() -> None:
"""Periodic pass: re-ping instances that still report an older hash."""
latest = _db.latest_config()
if latest is None:
return
acks = _db.get_acks(latest["id"])
peers = _db.list_peers(_TTL_SEC)
behind = []
for peer in peers:
name, instance_id, tags = _identity_of(peer)
eff = effective_hash(
latest["layers"], name=name, instance_id=instance_id, tags=tags
)
# A fresh ack for the current effective hash is proof enough: the
# announce payload may simply not have caught up yet.
ack = acks.get(instance_id)
if ack and ack.get("hash") == eff:
continue
if _is_behind(_reported_config(peer.get("meta")), eff):
behind.append(peer)
if not behind:
return
log.info(
"config.retry_notify version=%s count=%s names=%s",
latest["id"],
len(behind),
[p.get("name") for p in behind],
)
await asyncio.gather(
*(_notify_one(peer, latest["id"], latest["hash"]) for peer in behind),
return_exceptions=True,
)
async def _retry_loop() -> None:
while True:
await asyncio.sleep(_NOTIFY_RETRY_SEC)
try:
await _notify_behind()
except asyncio.CancelledError:
raise
except Exception: # a failed sweep must not kill the loop
log.exception("config.retry_loop_error")
@asynccontextmanager
async def lifespan(_: FastAPI):
task = asyncio.create_task(_retry_loop())
try:
yield
finally:
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
app = FastAPI(
title="navi hive",
description="Swarm address book and central config store for Navi instances.",
lifespan=lifespan,
)
# ── address book ────────────────────────────────────────────────────────────
class AnnouncePayload(BaseModel):
name: str = Field(min_length=1, max_length=64)
instance_id: str = Field(min_length=8, max_length=64)
version: str = Field(default="", max_length=32)
port: int = Field(ge=1, le=65535)
meta: dict = Field(default_factory=dict)
@app.get("/health")
async def health() -> dict:
latest = _db.latest_config()
return {
"status": "ok",
"hive": True,
"machines": len(_db.list_peers(_TTL_SEC)),
"config_version": latest["id"] if latest else 0,
}
@app.post("/announce")
async def announce(
payload: AnnouncePayload,
request: Request,
x_swarm_key: str | None = Header(default=None),
) -> dict:
_check_key(x_swarm_key)
# The source IP is the reliable address for a floating-IP network; the
# announced payload only supplies the port (and could not be trusted for
# the host anyway — anyone with the key may announce from anywhere).
host = request.client.host if request.client else None
_db.upsert(
instance_id=payload.instance_id,
name=payload.name,
host=host,
port=payload.port,
version=payload.version,
meta=payload.meta,
)
# The heartbeat is the fallback delivery channel: tell an instance that it
# is behind, so a missed notification costs one announce interval, not a
# manual push.
hint = None
latest = _db.latest_config()
if latest is not None:
tags = list(payload.meta.get("tags") or [])
eff = effective_hash(
latest["layers"],
name=payload.name,
instance_id=payload.instance_id,
tags=tags,
)
if _is_behind(_reported_config(payload.meta), eff):
hint = {"fetch": True, "version": latest["id"], "hash": eff}
return {"ok": True, "host": host, "config": hint}
@app.get("/peers")
async def peers(x_swarm_key: str | None = Header(default=None)) -> dict:
_check_key(x_swarm_key)
return {"peers": _db.list_peers(_TTL_SEC), "ttl_sec": _TTL_SEC}
# ── config: read side (swarm key) ───────────────────────────────────────────
def _tags_param(value: str | None) -> list[str]:
return [t.strip() for t in (value or "").split(",") if t.strip()]
@app.get("/config")
async def get_config(
name: str = "",
instance_id: str = "",
tags: str = "",
x_swarm_key: str | None = Header(default=None),
) -> dict:
"""The layers that apply to the asking instance.
``404`` when nothing was ever published, so an instance can tell "the
hive has no config channel for me" from "the hive says: no overlay".
An empty ``layers`` list is the latter and means "use your local config".
"""
_check_key(x_swarm_key)
latest = _db.latest_config()
if latest is None:
raise HTTPException(status_code=404, detail="no config published yet")
layers = effective_layers(
latest["layers"],
name=name,
instance_id=instance_id,
tags=_tags_param(tags),
)
return {
"version": latest["id"],
"hash": content_hash(layers),
"layers": layers,
"note": latest["note"],
"updated_at": latest["created_at"],
}
class AckPayload(BaseModel):
instance_id: str = Field(min_length=8, max_length=64)
hash: str = Field(default="", max_length=128)
version: int = 0
status: str = Field(default="applied", max_length=32)
restart_required: list[str] = Field(default_factory=list)
error: str | None = None
@app.post("/config/ack")
async def ack(
payload: AckPayload,
x_swarm_key: str | None = Header(default=None),
) -> dict:
_check_key(x_swarm_key)
latest = _db.latest_config()
version = payload.version or (latest["id"] if latest else 0)
if not version:
return {"ok": True, "ignored": "no config published"}
_db.record_ack(
instance_id=payload.instance_id,
version=version,
hash=payload.hash,
status=payload.status,
restart_required=payload.restart_required,
last_error=payload.error,
)
return {"ok": True, "version": version}
# ── config: write and administration (admin key / loopback) ─────────────────
class PublishPayload(BaseModel):
layers: list[dict]
note: str = Field(default="", max_length=500)
force: bool = False
@app.put("/config")
async def publish(
payload: PublishPayload,
request: Request,
x_admin_key: str | None = Header(default=None),
) -> dict:
"""Publish a new document and ping the fleet.
An identical document is a no-op unless ``force`` is set — republishing
the same config would otherwise churn the whole fleet for nothing.
"""
actor = _check_admin(request, x_admin_key)
try:
layers = validate_layers(payload.layers)
except ConfigError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
doc_hash = content_hash(layers)
latest = _db.latest_config()
if latest is not None and latest["hash"] == doc_hash and not payload.force:
return {
"status": "unchanged",
"version": latest["id"],
"hash": doc_hash,
"layers": len(layers),
}
if latest is not None and latest["hash"] == doc_hash:
asyncio.create_task(_notify_all(latest["id"], doc_hash))
return {
"status": "renotified",
"version": latest["id"],
"hash": doc_hash,
"layers": len(layers),
}
row = _db.add_config(
layers=layers, doc_hash=doc_hash, note=payload.note, created_by=actor
)
asyncio.create_task(_notify_all(row["id"], doc_hash))
return {
"status": "published",
"version": row["id"],
"hash": doc_hash,
"layers": len(layers),
"notify": "scheduled",
}
@app.get("/config/info")
async def config_info(
request: Request, x_admin_key: str | None = Header(default=None)
) -> dict:
_check_admin(request, x_admin_key)
latest = _db.latest_config() or {"id": 0, "hash": "", "layers": [], "note": ""}
return {
"version": latest["id"],
"hash": latest["hash"],
"note": latest.get("note") or "",
"layers": latest["layers"],
}
@app.get("/config/history")
async def config_history(
request: Request,
limit: int = 50,
x_admin_key: str | None = Header(default=None),
) -> dict:
_check_admin(request, x_admin_key)
return {"versions": _db.config_history(limit)}
@app.get("/config/version/{version_id}")
async def config_version(
version_id: int,
request: Request,
x_admin_key: str | None = Header(default=None),
) -> dict:
_check_admin(request, x_admin_key)
row = _db.get_config(version_id)
if row is None:
raise HTTPException(status_code=404, detail="no such version")
return row
@app.post("/config/rollback/{version_id}")
async def config_rollback(
version_id: int,
request: Request,
x_admin_key: str | None = Header(default=None),
) -> dict:
"""Roll back by publishing an old document as a new version.
History is never rewritten: the fleet's next fetch is a normal publish,
and the bad version stays visible in the history with its note intact.
"""
actor = _check_admin(request, x_admin_key)
row = _db.get_config(version_id)
if row is None:
raise HTTPException(status_code=404, detail="no such version")
new = _db.add_config(
layers=row["layers"],
doc_hash=row["hash"],
note=f"rollback to #{version_id}",
created_by=actor,
)
asyncio.create_task(_notify_all(new["id"], row["hash"]))
return {"status": "rolled_back", "version": new["id"], "from": version_id}
@app.get("/config/status")
async def config_status(
request: Request, x_admin_key: str | None = Header(default=None)
) -> dict:
"""Who runs the current config, who is behind, who rejected it.
This is the tracking the owner asked for: after a publish, look here to
see which machines took it and which still need a nudge (the hive pings
those on its own every ``HIVE_NOTIFY_RETRY_SEC``).
"""
_check_admin(request, x_admin_key)
latest = _db.latest_config()
acks = _db.get_acks(latest["id"]) if latest else {}
instances = []
behind = []
for peer in _db.list_peers(_TTL_SEC):
name, instance_id, tags = _identity_of(peer)
reported = _reported_config(peer.get("meta"))
eff = (
effective_hash(
latest["layers"], name=name, instance_id=instance_id, tags=tags
)
if latest
else ""
)
is_behind = bool(latest) and _is_behind(reported, eff)
ack = acks.get(instance_id)
if ack and ack.get("hash") == eff:
# The ack is fresher than the announce payload: it applied the
# current effective layers, whatever the last heartbeat said.
is_behind = False
if is_behind:
behind.append(name)
instances.append(
{
"name": name,
"instance_id": instance_id,
"tags": tags,
"online": peer["online"],
"last_seen": peer["last_seen"],
"reported": reported,
"effective_hash": eff,
"behind": is_behind,
"ack": acks.get(instance_id),
}
)
return {
"version": latest["id"] if latest else 0,
"hash": latest["hash"] if latest else "",
"generated_at": time.time(),
"behind": behind,
"instances": instances,
}
