5c634baa6d
All checks were successful
Build Archipelago ISO (dev) / build-iso (push) Successful in 34m25s
Re-adds the TCP transport (`0.0.0.0:8443`) to the rendered fips.yaml alongside UDP. Upstream factory default enables both; we had inadvertently narrowed to UDP-only when the yaml rewriter was last touched, which left nodes unable to reach fips.v0l.io (the public anchor only answers on TCP right now) or talk across networks that block UDP. Backend startup now compares the installed yaml against the current rendered schema and restarts whichever fips unit is active when they differ — so OTA-upgrading nodes pick up the new transport without anyone having to click Reconnect. Dropped the earlier plan to auto-add federated peers as seed anchors: invites don't carry a FIPS-reachable IP:port, and once TCP reconnects the public mesh, federated peers become npub-routable without needing a seed entry. Seed Anchors modal cleanup: replaced malformed header icon with a three-arc broadcast glyph, and the close button now matches the What's New modal (embedded in the card header, same icon + hover style) instead of the earlier floating off-design placeholder. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
936 lines
40 KiB
Rust
936 lines
40 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;
|
||
}
|
||
// Check if the installed fips.yaml matches what we'd
|
||
// render now. If not, we need to restart the daemon after
|
||
// reinstalling so it picks up schema changes (e.g. the
|
||
// v1.7.25 re-addition of the TCP transport). Without this,
|
||
// OTA'd nodes would be stuck on the old UDP-only config
|
||
// until someone manually clicked Reconnect.
|
||
let expected = crate::fips::config::render_config_yaml();
|
||
let installed = tokio::fs::read_to_string("/etc/fips/fips.yaml")
|
||
.await
|
||
.ok();
|
||
let config_changed = installed.as_deref() != Some(expected.as_str());
|
||
|
||
if let Err(e) = crate::fips::config::install(&identity_dir).await {
|
||
tracing::warn!("FIPS config install failed on startup: {}", e);
|
||
return;
|
||
}
|
||
if config_changed {
|
||
tracing::info!(
|
||
"FIPS config schema changed on disk — restarting daemon to pick up new transports"
|
||
);
|
||
// Restart whichever unit is actually supervising
|
||
// the daemon (archipelago-fips vs upstream fips).
|
||
let unit = crate::fips::service::active_unit().await;
|
||
if let Err(e) = crate::fips::service::restart(unit).await {
|
||
tracing::warn!(
|
||
"FIPS restart after config migration failed on {}: {} — user can retry via fips.reconnect",
|
||
unit,
|
||
e
|
||
);
|
||
}
|
||
}
|
||
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(())
|
||
}
|