DorianandClaude Opus 4.7
f8304aed90
Adds a local seed-anchor list at <data_dir>/seed-anchors.json. Each
entry is {npub, address, transport, label}. On archipelago startup
and every 5 minutes the list is pushed into the running fips daemon
via `fipsctl connect <npub> <addr> <transport>`, so a cluster can
anchor itself independently of the global fips.v0l.io. A flaky or
unreachable public anchor no longer strands a fresh install.
New RPCs:
- fips.list-seed-anchors
- fips.add-seed-anchor (validates npub1… + host:port)
- fips.remove-seed-anchor
- fips.apply-seed-anchors (on-demand re-dial)
New standalone UI card at views/server/FipsSeedAnchorsCard.vue. Not
wired into Home.vue / Server.vue — operator places it per the
entry-point convention.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
909 lines
38 KiB
Rust
909 lines
38 KiB
Rust
use crate::api::ApiHandler;
|
||
use crate::config::{Config, ContainerRuntime};
|
||
use crate::container::{docker_packages, DockerPackageScanner};
|
||
use crate::identity::{self, NodeIdentity};
|
||
use crate::monitoring::MetricsStore;
|
||
use crate::node_message;
|
||
use crate::nostr_discovery;
|
||
use crate::nostr_handshake;
|
||
use crate::peers;
|
||
use crate::state::StateManager;
|
||
use anyhow::Result;
|
||
use hyper::server::conn::Http;
|
||
use hyper::service::service_fn;
|
||
use std::collections::HashMap;
|
||
use std::net::SocketAddr;
|
||
use std::sync::Arc;
|
||
use std::time::Duration;
|
||
use tokio::net::TcpListener;
|
||
use tracing::{debug, error, info, warn};
|
||
|
||
pub struct Server {
|
||
_config: Config,
|
||
_identity: Arc<NodeIdentity>,
|
||
api_handler: Arc<ApiHandler>,
|
||
_state_manager: Arc<StateManager>,
|
||
}
|
||
|
||
impl Server {
|
||
pub async fn new(config: Config) -> Result<Self> {
|
||
let state_manager = Arc::new(StateManager::new());
|
||
|
||
// Load node identity and set stable server_info.
|
||
// Detect seed-backed vs legacy vs fresh install.
|
||
let identity_dir = config.data_dir.join("identity");
|
||
let has_seed = crate::seed::seed_exists(&config.data_dir);
|
||
let has_node_key = NodeIdentity::key_exists(&identity_dir);
|
||
|
||
let identity = if has_node_key {
|
||
// Existing keys on disk (seed-derived or legacy random) — load them.
|
||
NodeIdentity::load_or_create(&identity_dir).await?
|
||
} else {
|
||
// Fresh install — create a temporary identity.
|
||
// Onboarding will overwrite this with seed-derived keys.
|
||
NodeIdentity::load_or_create(&identity_dir).await?
|
||
};
|
||
|
||
let (mut data, _) = state_manager.get_snapshot().await;
|
||
data.server_info.id = identity.node_id();
|
||
data.server_info.pubkey = identity.pubkey_hex();
|
||
data.server_info.seed_backed = has_seed;
|
||
// Load persisted server name
|
||
let name_file = config.data_dir.join("server-name");
|
||
if let Ok(name) = tokio::fs::read_to_string(&name_file).await {
|
||
let name = name.trim().to_string();
|
||
if !name.is_empty() {
|
||
data.server_info.name = Some(name);
|
||
}
|
||
}
|
||
data.server_info.tor_address = docker_packages::read_tor_address("archipelago").await;
|
||
if let Some(ref tor) = data.server_info.tor_address {
|
||
data.server_info.node_address = Some(identity.node_address(tor));
|
||
}
|
||
state_manager.update_data(data.clone()).await;
|
||
|
||
// Retry Tor address in background — Tor may not be ready at startup
|
||
if data.server_info.tor_address.is_none() {
|
||
let sm = state_manager.clone();
|
||
let pubkey = identity.pubkey_hex();
|
||
tokio::spawn(async move {
|
||
for delay in [5, 10, 20, 30, 60] {
|
||
tokio::time::sleep(std::time::Duration::from_secs(delay)).await;
|
||
if let Some(tor) = docker_packages::read_tor_address("archipelago").await {
|
||
let (mut d, _) = sm.get_snapshot().await;
|
||
let addr =
|
||
format!("archipelago://{}#{}", tor.trim_end_matches('/'), pubkey);
|
||
d.server_info.tor_address = Some(tor.clone());
|
||
d.server_info.node_address = Some(addr);
|
||
sm.update_data(d).await;
|
||
tracing::info!(
|
||
"Tor address discovered after startup: {}",
|
||
&tor[..20.min(tor.len())]
|
||
);
|
||
break;
|
||
}
|
||
}
|
||
});
|
||
}
|
||
|
||
// Load persisted messages (Archipelago channel)
|
||
node_message::init(&config.data_dir).await;
|
||
|
||
// Auto-create default identity if none exist (fresh boot or factory reset)
|
||
{
|
||
let im = crate::identity_manager::IdentityManager::new(&config.data_dir).await;
|
||
if let Ok(mgr) = im {
|
||
if let Ok((list, _)) = mgr.list().await {
|
||
if list.is_empty() {
|
||
match mgr
|
||
.create(
|
||
"Default".to_string(),
|
||
crate::identity_manager::IdentityPurpose::Personal,
|
||
)
|
||
.await
|
||
{
|
||
Ok(record) => {
|
||
let _ = mgr.create_nostr_key(&record.id).await;
|
||
tracing::info!(did = %record.did, "Auto-created default identity with Nostr key");
|
||
}
|
||
Err(e) => tracing::debug!("Auto-identity creation (non-fatal): {}", e),
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// Revoke any previously published Nostr data (runs before publish so revocation is not overwritten)
|
||
let identity_dir = config.data_dir.join("identity");
|
||
let tor_proxy_revoke = config.nostr_tor_proxy.clone();
|
||
if let Err(e) =
|
||
nostr_discovery::revoke_if_needed(&identity_dir, tor_proxy_revoke.as_deref()).await
|
||
{
|
||
tracing::debug!("Nostr revoke (non-fatal): {}", e);
|
||
}
|
||
|
||
// Publish presence-only to Nostr (DID + Nostr pubkey, NO onion address).
|
||
// Onion addresses are exchanged privately via NIP-44 encrypted DMs.
|
||
if config.nostr_discovery_enabled && !config.nostr_relays.is_empty() {
|
||
let identity_dir = config.data_dir.join("identity");
|
||
let did =
|
||
identity::did_key_from_pubkey_hex(&data.server_info.pubkey).unwrap_or_default();
|
||
let version = data.server_info.version.clone();
|
||
let relays = config.nostr_relays.clone();
|
||
let tor_proxy = config.nostr_tor_proxy.clone();
|
||
tokio::spawn(async move {
|
||
if let Err(e) = nostr_handshake::publish_presence(
|
||
&identity_dir,
|
||
&did,
|
||
&version,
|
||
&relays,
|
||
tor_proxy.as_deref(),
|
||
)
|
||
.await
|
||
{
|
||
tracing::debug!("Nostr presence publish (non-fatal): {}", e);
|
||
}
|
||
});
|
||
}
|
||
info!(
|
||
"🔑 Node identity: {} (pubkey: {}...)",
|
||
identity.node_id(),
|
||
&identity.pubkey_hex()[..16.min(identity.pubkey_hex().len())]
|
||
);
|
||
|
||
let identity = Arc::new(identity);
|
||
|
||
// Create metrics store and spawn background collector
|
||
let metrics_store = Arc::new(MetricsStore::with_data_dir(config.data_dir.clone()).await);
|
||
let metrics_for_telemetry = metrics_store.clone();
|
||
crate::monitoring::spawn_metrics_collector(
|
||
metrics_store.clone(),
|
||
Some(state_manager.clone()),
|
||
Some(config.data_dir.clone()),
|
||
);
|
||
|
||
let api_handler =
|
||
Arc::new(ApiHandler::new(config.clone(), state_manager.clone(), metrics_store).await?);
|
||
|
||
// Initialize mesh networking service (if config has enabled: true)
|
||
{
|
||
let data_dir = config.data_dir.clone();
|
||
let did =
|
||
identity::did_key_from_pubkey_hex(&data.server_info.pubkey).unwrap_or_default();
|
||
let pubkey_hex = identity.pubkey_hex();
|
||
let signing_key = identity.signing_key();
|
||
match crate::mesh::MeshService::new(&data_dir, signing_key, &did, &pubkey_hex).await {
|
||
Ok(mut mesh_service) => {
|
||
// Pass the human-readable server name for mesh adverts
|
||
mesh_service.set_server_name(data.server_info.name.clone());
|
||
let mut mesh_config = crate::mesh::load_config(&data_dir)
|
||
.await
|
||
.unwrap_or_default();
|
||
|
||
// Auto-enable mesh if a radio is detected and no config exists yet
|
||
if !mesh_config.enabled {
|
||
let devices = crate::mesh::detect_devices().await;
|
||
if !devices.is_empty() {
|
||
info!("📡 Auto-detected mesh radio: {:?} — enabling mesh", devices);
|
||
mesh_config.enabled = true;
|
||
mesh_config.device_path = Some(devices[0].clone());
|
||
let _ = crate::mesh::save_config(&data_dir, &mesh_config).await;
|
||
}
|
||
}
|
||
|
||
if mesh_config.enabled {
|
||
if let Err(e) = mesh_service.start() {
|
||
warn!("Mesh service start failed (non-fatal): {}", e);
|
||
} else {
|
||
info!("📡 Mesh networking started");
|
||
}
|
||
}
|
||
api_handler
|
||
.rpc_handler()
|
||
.set_mesh_service(mesh_service)
|
||
.await;
|
||
info!("📡 Mesh service initialized");
|
||
}
|
||
Err(e) => {
|
||
warn!("Mesh service init failed (non-fatal): {}", e);
|
||
}
|
||
}
|
||
}
|
||
|
||
// Initialize transport router (unified routing: mesh > lan > tor)
|
||
{
|
||
let data_dir = config.data_dir.clone();
|
||
let did =
|
||
identity::did_key_from_pubkey_hex(&data.server_info.pubkey).unwrap_or_default();
|
||
let pubkey_hex = identity.pubkey_hex();
|
||
let mesh_config = crate::mesh::load_config(&data_dir)
|
||
.await
|
||
.unwrap_or_default();
|
||
let mesh_only = mesh_config.mesh_only_mode.unwrap_or(false);
|
||
|
||
match crate::transport::PeerRegistry::load(&data_dir).await {
|
||
Ok(registry) => {
|
||
let registry = std::sync::Arc::new(registry);
|
||
let mut transports: Vec<Box<dyn crate::transport::NodeTransport>> = Vec::new();
|
||
|
||
// Tor transport (always register — availability checked dynamically)
|
||
transports.push(Box::new(crate::transport::tor::TorTransport::new(
|
||
&pubkey_hex,
|
||
)));
|
||
|
||
// Mesh transport (wraps the mesh service)
|
||
transports.push(Box::new(
|
||
crate::transport::mesh_transport::MeshTransport::new(
|
||
api_handler.rpc_handler().mesh_service_arc(),
|
||
),
|
||
));
|
||
|
||
// LAN transport (mDNS discovery)
|
||
let mut lan = crate::transport::lan::LanTransport::new(&did, &pubkey_hex, 5678);
|
||
match lan.start(registry.clone()) {
|
||
Ok(()) => info!("📡 LAN transport (mDNS) started"),
|
||
Err(e) => debug!("LAN transport init (non-fatal): {}", e),
|
||
}
|
||
transports.push(Box::new(lan));
|
||
|
||
let router = std::sync::Arc::new(crate::transport::TransportRouter::new(
|
||
transports, registry, mesh_only,
|
||
));
|
||
api_handler.rpc_handler().set_transport_router(router).await;
|
||
info!("📡 Transport router initialized (mesh_only={})", mesh_only);
|
||
}
|
||
Err(e) => {
|
||
warn!("Transport router init failed (non-fatal): {}", e);
|
||
}
|
||
}
|
||
}
|
||
|
||
// Register Archipelago DWN protocols (background, non-blocking)
|
||
{
|
||
let data_dir = config.data_dir.clone();
|
||
tokio::spawn(async move {
|
||
if let Err(e) = register_dwn_protocols(&data_dir).await {
|
||
debug!("DWN protocol registration (non-fatal): {}", e);
|
||
}
|
||
});
|
||
}
|
||
|
||
// Periodic Tor address refresh (runs regardless of dev_mode)
|
||
// Picks up hostname when Tor creates it after startup/rotation (30-60s delay)
|
||
{
|
||
let state = state_manager.clone();
|
||
let identity_clone = identity.clone();
|
||
tokio::spawn(async move {
|
||
let mut interval = tokio::time::interval(Duration::from_secs(30));
|
||
loop {
|
||
interval.tick().await;
|
||
if let Err(e) = refresh_tor_address(&state, identity_clone.as_ref()).await {
|
||
debug!("Tor address refresh (non-fatal): {}", e);
|
||
}
|
||
}
|
||
});
|
||
}
|
||
|
||
// Initialize container scanner — discovers installed apps from Podman/Docker
|
||
{
|
||
let scanner = create_docker_scanner(&config).await?;
|
||
let state = state_manager.clone();
|
||
let identity_clone = identity.clone();
|
||
|
||
// Initial scan (delayed to let crash recovery finish first)
|
||
tokio::spawn(async move {
|
||
// Brief delay for containers to stabilize after boot
|
||
tokio::time::sleep(Duration::from_secs(3)).await;
|
||
info!("🐳 Scanning containers...");
|
||
// Tracks how many consecutive scans each container has been absent from.
|
||
// Prevents UI flapping when podman intermittently returns incomplete results.
|
||
let mut absence_tracker: HashMap<String, u32> = HashMap::new();
|
||
if let Err(e) = scan_and_update_packages(
|
||
&scanner,
|
||
&state,
|
||
identity_clone.as_ref(),
|
||
&mut absence_tracker,
|
||
)
|
||
.await
|
||
{
|
||
error!("Failed to scan containers: {}", e);
|
||
}
|
||
|
||
// Periodic scan every 60 seconds (only broadcasts if state changed)
|
||
// Uses an in-flight guard to skip scans when a previous one is still running
|
||
let mut interval = tokio::time::interval(Duration::from_secs(60));
|
||
// Skip missed ticks instead of catching up — prevents burst of scans
|
||
// after a slow podman response (which causes DB lock storms)
|
||
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
||
let scanning = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
|
||
loop {
|
||
interval.tick().await;
|
||
if scanning.load(std::sync::atomic::Ordering::Relaxed) {
|
||
debug!("Skipping container scan — previous scan still in progress");
|
||
continue;
|
||
}
|
||
scanning.store(true, std::sync::atomic::Ordering::Relaxed);
|
||
if let Err(e) = scan_and_update_packages(
|
||
&scanner,
|
||
&state,
|
||
identity_clone.as_ref(),
|
||
&mut absence_tracker,
|
||
)
|
||
.await
|
||
{
|
||
error!("Failed to update containers: {}", e);
|
||
}
|
||
scanning.store(false, std::sync::atomic::Ordering::Relaxed);
|
||
}
|
||
});
|
||
}
|
||
|
||
// Peer health monitoring — check every 5 minutes
|
||
{
|
||
let state = state_manager.clone();
|
||
let data_dir = config.data_dir.clone();
|
||
tokio::spawn(async move {
|
||
let mut interval = tokio::time::interval(Duration::from_secs(300));
|
||
loop {
|
||
interval.tick().await;
|
||
if let Err(e) = check_peer_health(&state, &data_dir).await {
|
||
debug!("Peer health check (non-fatal): {}", e);
|
||
}
|
||
}
|
||
});
|
||
}
|
||
|
||
// FIPS seed-anchor apply loop — every 5 minutes we re-push the
|
||
// configured seed anchors into the running fips daemon via
|
||
// `fipsctl connect`. This keeps the mesh bootstrap resilient:
|
||
// operators add cluster-local anchors in the UI, and a daemon
|
||
// restart or a flaky public anchor can't strand the node.
|
||
// First run is delayed 30s so fips has time to come up after
|
||
// onboarding before we start dialing.
|
||
{
|
||
let data_dir = config.data_dir.clone();
|
||
tokio::spawn(async move {
|
||
tokio::time::sleep(Duration::from_secs(30)).await;
|
||
let mut interval = tokio::time::interval(Duration::from_secs(300));
|
||
loop {
|
||
interval.tick().await;
|
||
match crate::fips::anchors::load(&data_dir).await {
|
||
Ok(list) if !list.is_empty() => {
|
||
let _ = crate::fips::anchors::apply(&list).await;
|
||
}
|
||
Ok(_) => { /* no seed anchors configured yet */ }
|
||
Err(e) => tracing::debug!(
|
||
"Seed-anchor apply: load failed (non-fatal): {}",
|
||
e
|
||
),
|
||
}
|
||
}
|
||
});
|
||
}
|
||
|
||
// did:dht auto-refresh — re-publish DHT records every 2 hours
|
||
if config.nostr_discovery_enabled {
|
||
let data_dir = config.data_dir.clone();
|
||
tokio::spawn(async move {
|
||
let mut interval = tokio::time::interval(Duration::from_secs(7200));
|
||
loop {
|
||
interval.tick().await;
|
||
let identity_dir = data_dir.join("identity");
|
||
let node_key_path = identity_dir.join("node_key");
|
||
if !node_key_path.exists() {
|
||
continue;
|
||
}
|
||
match tokio::fs::read(&node_key_path).await {
|
||
Ok(key_bytes) if key_bytes.len() == 32 => {
|
||
let mut seed = [0u8; 32];
|
||
seed.copy_from_slice(&key_bytes);
|
||
let signing_key = ed25519_dalek::SigningKey::from_bytes(&seed);
|
||
match crate::network::did_dht::create_and_publish(&signing_key, &[])
|
||
.await
|
||
{
|
||
Ok(did) => tracing::info!(did = %did, "did:dht record refreshed"),
|
||
Err(e) => tracing::debug!("did:dht refresh (non-fatal): {}", e),
|
||
}
|
||
}
|
||
_ => {
|
||
tracing::debug!("did:dht refresh skipped: no valid node key");
|
||
}
|
||
}
|
||
}
|
||
});
|
||
}
|
||
|
||
// Periodic federation state sync — every 30 min we call
|
||
// federation::sync_with_peer on each Trusted peer. Without this
|
||
// users had to manually click Sync for `fips_npub`/transport
|
||
// badge/state updates to propagate; now it happens in the
|
||
// background. Staggers peers with a 5s delay so we don't thunder
|
||
// the Tor SOCKS proxy. Sync itself already prefers FIPS.
|
||
{
|
||
let data_dir = config.data_dir.clone();
|
||
let state = state_manager.clone();
|
||
tokio::spawn(async move {
|
||
// First run 60s after boot to let onboarding settle.
|
||
tokio::time::sleep(Duration::from_secs(60)).await;
|
||
let mut interval = tokio::time::interval(Duration::from_secs(1800));
|
||
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
|
||
loop {
|
||
interval.tick().await;
|
||
let Ok(nodes) = crate::federation::load_nodes(&data_dir).await else {
|
||
continue;
|
||
};
|
||
if nodes.is_empty() {
|
||
continue;
|
||
}
|
||
let (data, _) = state.get_snapshot().await;
|
||
let Ok(local_did) =
|
||
crate::identity::did_key_from_pubkey_hex(&data.server_info.pubkey)
|
||
else {
|
||
continue;
|
||
};
|
||
let identity_dir = data_dir.join("identity");
|
||
let Ok(node_identity) =
|
||
crate::identity::NodeIdentity::load_or_create(&identity_dir).await
|
||
else {
|
||
continue;
|
||
};
|
||
|
||
for node in &nodes {
|
||
if node.trust_level == crate::federation::TrustLevel::Untrusted {
|
||
continue;
|
||
}
|
||
match crate::federation::sync_with_peer(
|
||
&data_dir,
|
||
node,
|
||
&local_did,
|
||
|bytes| node_identity.sign(bytes),
|
||
)
|
||
.await
|
||
{
|
||
Ok(_) => debug!(
|
||
"Periodic federation sync ok: {}",
|
||
node.did.chars().take(20).collect::<String>()
|
||
),
|
||
Err(e) => debug!(
|
||
"Periodic federation sync with {}: {}",
|
||
node.did.chars().take(20).collect::<String>(),
|
||
e
|
||
),
|
||
}
|
||
tokio::time::sleep(Duration::from_secs(5)).await;
|
||
}
|
||
}
|
||
});
|
||
}
|
||
|
||
// Container health monitoring — auto-restart unhealthy containers
|
||
// Respects webhook config: skips when disabled or ContainerCrash not subscribed
|
||
crate::health_monitor::spawn_health_monitor(state_manager.clone(), config.data_dir.clone());
|
||
|
||
// Periodic telemetry reporter (every 15 min when opted in)
|
||
crate::monitoring::spawn_telemetry_reporter(
|
||
metrics_for_telemetry,
|
||
Some(state_manager.clone()),
|
||
config.data_dir.clone(),
|
||
);
|
||
|
||
// Post-onboarding auto-activation for archipelago-fips. Runs once
|
||
// at startup: if fips_key is on disk, install /etc/fips/fips.yaml
|
||
// (schema-refreshed) and start the service. This removes the
|
||
// need for a user-facing "Activate" button — the node comes up
|
||
// with FIPS running whenever the seed has been onboarded. Also
|
||
// self-heals legacy raw-byte fips.key files (load_fips_keys
|
||
// rewrites them as bech32 nsec the first time they're read).
|
||
// Pre-onboarding nodes: ConditionPathExists on the service unit
|
||
// + the `fips_key_exists` guard here keep this quiet.
|
||
{
|
||
let data_dir = config.data_dir.clone();
|
||
tokio::spawn(async move {
|
||
let identity_dir = data_dir.join("identity");
|
||
if !crate::identity::fips_key_exists(&identity_dir) {
|
||
tracing::debug!("FIPS auto-activate skipped: fips_key not on disk");
|
||
return;
|
||
}
|
||
// Trigger the migration path in load_fips_keys so old raw-byte
|
||
// key files are rewritten as bech32 before fips.yaml install.
|
||
if let Err(e) = crate::identity::load_fips_keys(&identity_dir).await {
|
||
tracing::warn!("FIPS key load/migrate failed: {}", e);
|
||
return;
|
||
}
|
||
if let Err(e) = crate::fips::config::install(&identity_dir).await {
|
||
tracing::warn!("FIPS config install failed on startup: {}", e);
|
||
return;
|
||
}
|
||
if let Err(e) = crate::fips::service::activate(crate::fips::SERVICE_UNIT).await {
|
||
tracing::warn!(
|
||
"archipelago-fips activate failed on startup: {} — user can retry via fips.install RPC",
|
||
e
|
||
);
|
||
return;
|
||
}
|
||
tracing::info!("archipelago-fips auto-activated on startup");
|
||
});
|
||
}
|
||
|
||
Ok(Self {
|
||
_config: config,
|
||
_identity: identity,
|
||
api_handler,
|
||
_state_manager: state_manager,
|
||
})
|
||
}
|
||
|
||
/// Serve with a graceful shutdown signal.
|
||
///
|
||
/// `main_addr` is the primary listener (historically `127.0.0.1:5678`).
|
||
/// The main listener always comes up on `main_addr`. The FIPS peer
|
||
/// listener (path-filtered, bound to `fips0`'s ULA) is managed by a
|
||
/// late-binding task that polls every 30s: if fips0 isn't up at
|
||
/// startup (pre-onboarding install, legacy node pre-fips.install),
|
||
/// it keeps trying until the interface appears — no archipelago
|
||
/// restart required after the user activates FIPS.
|
||
///
|
||
/// When `shutdown` completes, both listeners stop accepting and drain
|
||
/// in-flight requests (bounded by `DRAIN_TIMEOUT`).
|
||
pub async fn serve_with_shutdown(
|
||
&self,
|
||
main_addr: SocketAddr,
|
||
shutdown: impl std::future::Future<Output = ()>,
|
||
) -> Result<()> {
|
||
let active_connections = Arc::new(tokio::sync::Semaphore::new(1024));
|
||
let (tx, rx_main) = tokio::sync::watch::channel(false);
|
||
|
||
let main_task = tokio::spawn(accept_loop(
|
||
self.api_handler.clone(),
|
||
TcpListener::bind(main_addr).await?,
|
||
active_connections.clone(),
|
||
false, // main listener: no path filter
|
||
rx_main,
|
||
main_addr,
|
||
));
|
||
|
||
// Peer listener: late-binding so we don't need an archipelago
|
||
// restart when fips0 comes up after onboarding.
|
||
let peer_task = tokio::spawn(peer_late_bind_loop(
|
||
self.api_handler.clone(),
|
||
active_connections.clone(),
|
||
tx.subscribe(),
|
||
));
|
||
|
||
shutdown.await;
|
||
info!("Shutdown signal received, draining connections...");
|
||
let _ = tx.send(true);
|
||
|
||
// Wait up to 5s for in-flight requests.
|
||
let drain_start = std::time::Instant::now();
|
||
let drain_timeout = std::time::Duration::from_secs(5);
|
||
while active_connections.available_permits() < 1024 {
|
||
if drain_start.elapsed() > drain_timeout {
|
||
warn!("Drain timeout reached, forcing shutdown");
|
||
break;
|
||
}
|
||
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
||
}
|
||
|
||
let _ = main_task.await;
|
||
let _ = peer_task.await;
|
||
|
||
info!("Shutdown complete");
|
||
Ok(())
|
||
}
|
||
}
|
||
|
||
/// Poll every 30s for `fips0`'s ULA; when it appears, bind the peer
|
||
/// listener and run the normal accept loop. If the bind fails (port
|
||
/// already taken, permissions), log and keep retrying. Returns on
|
||
/// shutdown. First tick fires immediately so the hot path for
|
||
/// already-up fips0 is still zero-cost.
|
||
async fn peer_late_bind_loop(
|
||
handler: Arc<ApiHandler>,
|
||
active_connections: Arc<tokio::sync::Semaphore>,
|
||
mut shutdown_rx: tokio::sync::watch::Receiver<bool>,
|
||
) {
|
||
let mut interval = tokio::time::interval(std::time::Duration::from_secs(30));
|
||
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
|
||
loop {
|
||
tokio::select! {
|
||
_ = interval.tick() => {
|
||
let Some(ip) = crate::fips::iface::fips0_ula() else { continue };
|
||
let addr = SocketAddr::new(
|
||
std::net::IpAddr::V6(ip),
|
||
crate::fips::dial::PEER_PORT,
|
||
);
|
||
let listener = match TcpListener::bind(addr).await {
|
||
Ok(l) => l,
|
||
Err(e) => {
|
||
warn!("FIPS peer listener bind {} failed: {} — retrying in 30s", addr, e);
|
||
continue;
|
||
}
|
||
};
|
||
info!("FIPS peer listener bound {}", addr);
|
||
// Once bound, serve until shutdown fires. accept_loop
|
||
// returns on shutdown, which also ends this outer loop.
|
||
accept_loop(
|
||
handler,
|
||
listener,
|
||
active_connections,
|
||
true, // peer listener: apply path filter
|
||
shutdown_rx,
|
||
addr,
|
||
)
|
||
.await;
|
||
return;
|
||
}
|
||
_ = shutdown_rx.changed() => {
|
||
if *shutdown_rx.borrow() { return; }
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Whitelist of HTTP paths reachable via the peer-facing (FIPS) listener.
|
||
/// Every entry is an endpoint already protected by cryptographic auth
|
||
/// (ed25519 signature verification inside the handler, federation DID
|
||
/// headers checked by the content server, or JSON-RPC methods whose
|
||
/// handlers verify per-message signatures).
|
||
///
|
||
/// Anything not on this list returns 404 on the peer listener.
|
||
pub fn is_peer_allowed_path(path: &str) -> bool {
|
||
// Exact matches
|
||
matches!(
|
||
path,
|
||
"/health"
|
||
| "/rpc/v1"
|
||
| "/archipelago/node-message"
|
||
| "/archipelago/mesh-typed"
|
||
| "/dwn"
|
||
| "/transport/inbox"
|
||
)
|
||
// Prefix-matched content endpoints (peer file browse + fetch)
|
||
|| path.starts_with("/content/")
|
||
}
|
||
|
||
async fn accept_loop(
|
||
handler: Arc<ApiHandler>,
|
||
listener: TcpListener,
|
||
active_connections: Arc<tokio::sync::Semaphore>,
|
||
peer_only: bool,
|
||
mut shutdown_rx: tokio::sync::watch::Receiver<bool>,
|
||
local_addr: SocketAddr,
|
||
) {
|
||
loop {
|
||
tokio::select! {
|
||
result = listener.accept() => {
|
||
let (stream, peer_addr) = match result {
|
||
Ok(c) => c,
|
||
Err(e) => {
|
||
error!("{} accept error: {}", local_addr, e);
|
||
continue;
|
||
}
|
||
};
|
||
let handler = handler.clone();
|
||
let permit = active_connections.clone().acquire_owned().await;
|
||
tokio::spawn(async move {
|
||
let _permit = permit;
|
||
let service = service_fn(move |req: hyper::Request<hyper::Body>| {
|
||
let handler = handler.clone();
|
||
async move {
|
||
if peer_only && !is_peer_allowed_path(req.uri().path()) {
|
||
let resp = hyper::Response::builder()
|
||
.status(hyper::StatusCode::NOT_FOUND)
|
||
.body(hyper::Body::empty())
|
||
.expect("static response builds");
|
||
return Ok::<_, std::io::Error>(resp);
|
||
}
|
||
handler
|
||
.handle_request(req)
|
||
.await
|
||
.map_err(|e| std::io::Error::other(format!("{}", e)))
|
||
}
|
||
});
|
||
if let Err(e) = Http::new()
|
||
.http1_keep_alive(false)
|
||
.serve_connection(stream, service)
|
||
.with_upgrades()
|
||
.await
|
||
{
|
||
error!("Error serving connection from {}: {}", peer_addr, e);
|
||
}
|
||
});
|
||
}
|
||
_ = shutdown_rx.changed() => {
|
||
if *shutdown_rx.borrow() {
|
||
return;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
async fn create_docker_scanner(config: &Config) -> Result<DockerPackageScanner> {
|
||
let user = std::env::var("USER").unwrap_or_else(|_| "archipelago".to_string());
|
||
|
||
let runtime: Arc<dyn archipelago_container::ContainerRuntime> = match &config.container_runtime
|
||
{
|
||
ContainerRuntime::Podman => {
|
||
Arc::new(archipelago_container::PodmanRuntime::new(user.clone()))
|
||
}
|
||
ContainerRuntime::Docker => {
|
||
Arc::new(archipelago_container::DockerRuntime::new(user.clone()))
|
||
}
|
||
ContainerRuntime::Auto => {
|
||
Arc::new(archipelago_container::AutoRuntime::new(user.clone()).await?)
|
||
}
|
||
};
|
||
|
||
Ok(DockerPackageScanner::new(runtime))
|
||
}
|
||
|
||
async fn refresh_tor_address(state: &StateManager, identity: &NodeIdentity) -> Result<()> {
|
||
let tor_addr = docker_packages::read_tor_address("archipelago").await;
|
||
let (current_data, _) = state.get_snapshot().await;
|
||
if tor_addr != current_data.server_info.tor_address {
|
||
let mut data = current_data;
|
||
data.server_info.tor_address = tor_addr.clone();
|
||
data.server_info.node_address = tor_addr.as_ref().map(|t| identity.node_address(t));
|
||
state.update_data(data).await;
|
||
if let Some(ref addr) = tor_addr {
|
||
info!("🔒 Tor address updated: {}", addr);
|
||
}
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// Number of consecutive absent scans before removing a container from state.
|
||
/// 3 scans × 30s = 90 seconds of absence before removal.
|
||
const CONTAINER_ABSENCE_THRESHOLD: u32 = 3;
|
||
|
||
async fn scan_and_update_packages(
|
||
scanner: &DockerPackageScanner,
|
||
state: &StateManager,
|
||
identity: &NodeIdentity,
|
||
absence_tracker: &mut HashMap<String, u32>,
|
||
) -> Result<()> {
|
||
let packages = scanner.scan_containers().await?;
|
||
|
||
let (current_data, _) = state.get_snapshot().await;
|
||
let tor_addr = docker_packages::read_tor_address("archipelago").await;
|
||
let tor_changed = tor_addr != current_data.server_info.tor_address;
|
||
let first_scan = !current_data.server_info.status_info.containers_scanned;
|
||
|
||
// Check if update scheduler has found an available update
|
||
let update_available = crate::update::load_state(std::path::Path::new("/var/lib/archipelago"))
|
||
.await
|
||
.map(|s| s.available_update.is_some())
|
||
.unwrap_or(false);
|
||
let update_changed = update_available != current_data.server_info.status_info.updated;
|
||
|
||
// Empty scan result = podman failure or timeout, preserve existing state
|
||
if packages.is_empty() && !first_scan {
|
||
if tor_changed || update_changed {
|
||
let mut data = current_data;
|
||
data.server_info.tor_address = tor_addr.clone();
|
||
data.server_info.node_address = tor_addr.as_ref().map(|t| identity.node_address(t));
|
||
data.server_info.status_info.updated = update_available;
|
||
state.update_data(data).await;
|
||
}
|
||
return Ok(());
|
||
}
|
||
|
||
// Merge scan results with current state instead of full replacement.
|
||
// This prevents containers from vanishing when podman intermittently
|
||
// returns incomplete results under heavy load.
|
||
let mut merged = current_data.package_data.clone();
|
||
let mut changed = false;
|
||
|
||
// Update/add containers found in this scan
|
||
for (id, pkg) in &packages {
|
||
absence_tracker.remove(id);
|
||
if merged.get(id) != Some(pkg) {
|
||
merged.insert(id.clone(), pkg.clone());
|
||
changed = true;
|
||
}
|
||
}
|
||
|
||
// Track containers in state but missing from this scan.
|
||
// Only remove after CONTAINER_ABSENCE_THRESHOLD consecutive absent scans.
|
||
let current_ids: Vec<String> = merged.keys().cloned().collect();
|
||
for id in current_ids {
|
||
if !packages.contains_key(&id) {
|
||
let count = absence_tracker.entry(id.clone()).or_insert(0);
|
||
*count += 1;
|
||
if *count >= CONTAINER_ABSENCE_THRESHOLD {
|
||
debug!(
|
||
"Removing {} from state after {} consecutive absent scans",
|
||
id, count
|
||
);
|
||
merged.remove(&id);
|
||
absence_tracker.remove(&id);
|
||
changed = true;
|
||
}
|
||
}
|
||
}
|
||
|
||
if changed || tor_changed || first_scan || update_changed {
|
||
let mut data = current_data;
|
||
data.package_data = merged;
|
||
data.server_info.tor_address = tor_addr.clone();
|
||
data.server_info.node_address = tor_addr.as_ref().map(|t| identity.node_address(t));
|
||
data.server_info.status_info.containers_scanned = true;
|
||
data.server_info.status_info.updated = update_available;
|
||
state.update_data(data).await;
|
||
debug!(
|
||
"📦 State changed (packages={}, tor={}, first_scan={}, update={}), broadcasting update",
|
||
changed, tor_changed, first_scan, update_changed
|
||
);
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
|
||
/// Register Archipelago DWN protocols on startup.
|
||
async fn register_dwn_protocols(data_dir: &std::path::Path) -> Result<()> {
|
||
use crate::network::dwn_store::{DwnStore, ProtocolDefinition};
|
||
|
||
let protocols = [
|
||
("https://archipelago.dev/protocols/node-identity/v1", true),
|
||
("https://archipelago.dev/protocols/file-catalog/v1", true),
|
||
("https://archipelago.dev/protocols/federation/v1", false),
|
||
("https://archipelago.dev/protocols/app-deploy/v1", false),
|
||
];
|
||
|
||
let store = DwnStore::new(data_dir).await?;
|
||
let existing = store.list_protocols().await?;
|
||
let existing_uris: std::collections::HashSet<String> =
|
||
existing.iter().map(|p| p.protocol.clone()).collect();
|
||
|
||
let mut registered = 0;
|
||
for (uri, published) in &protocols {
|
||
if existing_uris.contains(*uri) {
|
||
continue;
|
||
}
|
||
let def = ProtocolDefinition {
|
||
protocol: uri.to_string(),
|
||
published: *published,
|
||
types: std::collections::HashMap::new(),
|
||
structure: std::collections::HashMap::new(),
|
||
date_registered: chrono::Utc::now().to_rfc3339(),
|
||
};
|
||
store.register_protocol(&def).await?;
|
||
registered += 1;
|
||
}
|
||
|
||
if registered > 0 {
|
||
info!("📋 Registered {registered} DWN protocols");
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// Periodically check peer reachability and broadcast status changes.
|
||
async fn check_peer_health(state: &StateManager, data_dir: &std::path::Path) -> Result<()> {
|
||
let known_peers = peers::load_peers(data_dir).await.unwrap_or_default();
|
||
if known_peers.is_empty() {
|
||
return Ok(());
|
||
}
|
||
|
||
let mut new_health = std::collections::HashMap::new();
|
||
for peer in &known_peers {
|
||
let fips_npub = crate::federation::fips_npub_for_onion(data_dir, &peer.onion).await;
|
||
let reachable =
|
||
node_message::check_peer_reachable(&peer.onion, fips_npub.as_deref())
|
||
.await
|
||
.unwrap_or(false);
|
||
new_health.insert(peer.onion.clone(), reachable);
|
||
}
|
||
|
||
let (current_data, _) = state.get_snapshot().await;
|
||
if current_data.peer_health != new_health {
|
||
let mut data = current_data;
|
||
data.peer_health = new_health;
|
||
state.update_data(data).await;
|
||
debug!("🔗 Peer health updated, broadcasting changes");
|
||
}
|
||
|
||
Ok(())
|
||
}
|