use std::collections::{BTreeMap, HashMap, VecDeque};
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex as StdMutex;
use std::time::{Duration, Instant};
use axum::http::StatusCode;
use serde::{Deserialize, Serialize};
use sha2::Sha256;
use tokio::sync::{broadcast, mpsc, watch, Mutex, RwLock};
use crate as x0x;
use crate::contacts::ContactStore;
use crate::{Agent, KvStoreHandle, TaskListHandle};
use super::auth::SessionStore;
use super::sse::SseEvent;
use super::ws::{SharedTopicState, WsOutboundStats, WsSession};
use super::{
ExpectedJoinResultInviter, FileChunkAckSlot, NamedGroupMetadataEvent, PendingJoinResult,
PendingTreeKemMetadataEvent, PendingWelcome, PendingWelcomeReceive, RestSubscription,
WelcomeFetchWaiter,
};
fn validate_instance_name_grammar(name: &str) -> anyhow::Result<()> {
if name.is_empty() || name.len() > 64 {
anyhow::bail!("instance name must be 1-64 characters");
}
let valid = name
.chars()
.next()
.is_some_and(|c| c.is_ascii_alphanumeric())
&& name.chars().all(|c| c.is_ascii_alphanumeric() || c == '-');
if !valid {
anyhow::bail!(
"instance name must start with alphanumeric and contain only alphanumeric or hyphens"
);
}
Ok(())
}
pub fn validate_instance_name(name: &str) -> anyhow::Result<()> {
validate_instance_name_grammar(name)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InstanceName(String);
impl InstanceName {
#[must_use]
pub fn as_str(&self) -> &str {
&self.0
}
#[must_use]
pub fn into_string(self) -> String {
self.0
}
}
impl TryFrom<String> for InstanceName {
type Error = anyhow::Error;
fn try_from(name: String) -> Result<Self, Self::Error> {
validate_instance_name_grammar(&name)?;
Ok(Self(name))
}
}
#[derive(Default)]
pub struct ServeOptions {
pub skip_update_check: bool,
pub cli_no_port_mapping: bool,
pub cli_disable_peer_cache: bool,
pub instance_name: Option<String>,
pub exec_policy: x0x::exec::ExecPolicy,
pub connect_policy: x0x::connect::ConnectPolicy,
pub self_update_enabled: bool,
}
pub struct ServerHandle {
pub(super) local_addr: SocketAddr,
pub(super) cancel: tokio_util::sync::CancellationToken,
pub(super) task: Option<tokio::task::JoinHandle<anyhow::Result<()>>>,
}
impl ServerHandle {
#[must_use]
pub fn local_addr(&self) -> SocketAddr {
self.local_addr
}
pub fn shutdown(&self) {
self.cancel.cancel();
}
pub async fn wait(mut self) -> anyhow::Result<()> {
let Some(task) = self.task.take() else {
return Err(anyhow::anyhow!("server handle already consumed"));
};
match task.await {
Ok(res) => res,
Err(e) => Err(anyhow::Error::new(e).context("server supervisor task failed")),
}
}
#[must_use]
pub fn cancellation_token(&self) -> tokio_util::sync::CancellationToken {
self.cancel.clone()
}
pub async fn shutdown_and_wait(self) -> anyhow::Result<()> {
self.cancel.cancel();
self.wait().await
}
}
impl Drop for ServerHandle {
fn drop(&mut self) {
self.cancel.cancel();
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DaemonConfig {
#[serde(default = "default_bind_address")]
pub bind_address: SocketAddr,
#[serde(default = "default_api_address")]
pub api_address: SocketAddr,
#[serde(default = "default_data_dir")]
pub data_dir: PathBuf,
#[serde(default = "default_log_level")]
pub log_level: String,
#[serde(default = "default_log_format")]
pub log_format: String,
#[serde(default = "default_bootstrap_peers")]
pub bootstrap_peers: Vec<SocketAddr>,
#[serde(default = "default_port_mapping_enabled")]
pub(super) port_mapping_enabled: bool,
#[serde(default)]
pub(super) peer_relay: x0x::network::PeerRelayConfig,
#[serde(default)]
pub(super) update: DaemonUpdateConfig,
#[serde(default)]
pub gossip: x0x::gossip::GossipConfig,
#[serde(default = "default_heartbeat_interval")]
pub(super) heartbeat_interval_secs: u64,
#[serde(default = "default_identity_ttl")]
pub(super) identity_ttl_secs: u64,
#[serde(default)]
pub(super) user_key_path: Option<PathBuf>,
#[serde(default = "default_rendezvous_enabled")]
pub(super) rendezvous_enabled: bool,
#[serde(default = "default_rendezvous_validity_ms")]
pub(super) rendezvous_validity_ms: u64,
#[serde(default)]
pub(super) presence_beacon_interval_secs: Option<u64>,
#[serde(default)]
pub(super) presence_event_poll_interval_secs: Option<u64>,
#[serde(default)]
pub(super) presence_offline_timeout_secs: Option<u64>,
#[serde(default)]
pub instance_name: Option<String>,
#[serde(default)]
pub identity_dir: Option<PathBuf>,
#[serde(default)]
pub(super) directory_digest_interval_secs: Option<u64>,
#[serde(default)]
pub(super) group_card_republish_interval_secs: Option<u64>,
#[serde(default)]
pub(super) directory_resubscribe_jitter_ms: Option<u64>,
#[serde(default)]
pub(super) forward: x0x::forward::ForwardConfig,
}
pub const DEFAULT_QUIC_PORT: u16 = 5483;
fn default_bootstrap_peers() -> Vec<SocketAddr> {
x0x::network::DEFAULT_BOOTSTRAP_PEERS
.iter()
.filter_map(|s| s.parse().ok())
.collect()
}
fn default_port_mapping_enabled() -> bool {
true
}
pub fn default_bind_address() -> SocketAddr {
SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 0], DEFAULT_QUIC_PORT))
}
pub fn default_api_address() -> SocketAddr {
SocketAddr::from(([127, 0, 0, 1], 12700))
}
pub fn default_data_dir() -> PathBuf {
dirs::data_dir()
.map(|d| d.join("x0x"))
.unwrap_or_else(|| PathBuf::from("/var/lib/x0x"))
}
pub(super) fn shared_cache_dir() -> PathBuf {
let dir = dirs::data_dir()
.map(|d| d.join("x0x"))
.unwrap_or_else(|| PathBuf::from("/var/lib/x0x"));
let _ = std::fs::create_dir_all(&dir);
dir
}
fn default_log_level() -> String {
"warn".to_string()
}
fn default_log_format() -> String {
"text".to_string()
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(super) struct DaemonUpdateConfig {
#[serde(default = "default_true")]
pub(super) enabled: bool,
#[serde(default = "default_rollout_window_minutes")]
pub(super) rollout_window_minutes: u64,
#[serde(default = "default_true")]
pub(super) stop_on_upgrade: bool,
#[serde(default = "default_fallback_check_interval_minutes")]
pub(super) fallback_check_interval_minutes: u64,
#[serde(default = "default_update_repo")]
pub(super) repo: String,
#[serde(default)]
pub(super) include_prereleases: bool,
#[serde(default = "default_true")]
pub(super) gossip_updates: bool,
}
impl Default for DaemonUpdateConfig {
fn default() -> Self {
Self {
enabled: true,
rollout_window_minutes: 0,
stop_on_upgrade: true,
fallback_check_interval_minutes: 2880,
repo: default_update_repo(),
include_prereleases: false,
gossip_updates: true,
}
}
}
fn default_true() -> bool {
true
}
fn default_rollout_window_minutes() -> u64 {
0
}
fn default_fallback_check_interval_minutes() -> u64 {
2880
}
fn default_update_repo() -> String {
"saorsa-labs/x0x".to_string()
}
fn default_heartbeat_interval() -> u64 {
x0x::IDENTITY_HEARTBEAT_INTERVAL_SECS
}
fn default_identity_ttl() -> u64 {
x0x::IDENTITY_TTL_SECS
}
fn default_rendezvous_enabled() -> bool {
true
}
fn default_rendezvous_validity_ms() -> u64 {
3_600_000 }
impl DaemonConfig {
#[must_use]
pub fn update_enabled(&self) -> bool {
self.update.enabled
}
}
impl Default for DaemonConfig {
fn default() -> Self {
Self {
bind_address: default_bind_address(),
api_address: default_api_address(),
data_dir: default_data_dir(),
log_level: default_log_level(),
log_format: default_log_format(),
bootstrap_peers: x0x::network::DEFAULT_BOOTSTRAP_PEERS
.iter()
.filter_map(|s| s.parse().ok())
.collect(),
port_mapping_enabled: default_port_mapping_enabled(),
peer_relay: x0x::network::PeerRelayConfig::default(),
update: DaemonUpdateConfig::default(),
gossip: x0x::gossip::GossipConfig::default(),
heartbeat_interval_secs: default_heartbeat_interval(),
identity_ttl_secs: default_identity_ttl(),
user_key_path: None,
rendezvous_enabled: default_rendezvous_enabled(),
rendezvous_validity_ms: default_rendezvous_validity_ms(),
presence_beacon_interval_secs: None,
presence_event_poll_interval_secs: None,
presence_offline_timeout_secs: None,
instance_name: None,
identity_dir: None,
directory_digest_interval_secs: None,
group_card_republish_interval_secs: None,
directory_resubscribe_jitter_ms: None,
forward: x0x::forward::ForwardConfig::default(),
}
}
}
pub(super) struct AppState {
pub(super) agent: Arc<Agent>,
pub(super) subscriptions: RwLock<HashMap<String, RestSubscription>>,
pub(super) task_lists: RwLock<HashMap<String, TaskListHandle>>,
pub(super) kv_stores: RwLock<HashMap<String, KvStoreHandle>>,
pub(super) named_groups: RwLock<HashMap<String, x0x::groups::GroupInfo>>,
pub(super) named_groups_path: PathBuf,
pub(super) named_groups_persistence_lock: Mutex<()>,
pub(super) group_metadata_tasks: RwLock<HashMap<String, tokio::task::JoinHandle<()>>>,
pub(super) group_card_cache: RwLock<HashMap<String, x0x::groups::GroupCard>>,
pub(super) directory_cache: RwLock<x0x::groups::DirectoryShardCache>,
pub(super) directory_subscriptions: RwLock<x0x::groups::SubscriptionSet>,
pub(super) directory_subscriptions_path: PathBuf,
pub(super) directory_tasks:
RwLock<HashMap<(x0x::groups::ShardKind, u32), tokio::task::JoinHandle<()>>>,
pub(super) directory_digest_interval_secs: u64,
pub(super) directory_resubscribe_jitter_ms: u64,
pub(super) public_messages: RwLock<HashMap<String, Vec<x0x::groups::GroupPublicMessage>>>,
pub(super) public_message_tasks: RwLock<HashMap<String, tokio::task::JoinHandle<()>>>,
pub(super) agent_kem_keypair: Arc<x0x::groups::kem_envelope::AgentKemKeypair>,
pub(super) contacts: Arc<RwLock<ContactStore>>,
pub(super) mls_groups: RwLock<HashMap<String, x0x::mls::MlsGroup>>,
#[allow(dead_code)]
pub(super) mls_groups_path: PathBuf,
pub(super) pending_join_results: RwLock<HashMap<String, PendingJoinResult>>,
pub(super) expected_join_result_inviters: StdMutex<HashMap<String, ExpectedJoinResultInviter>>,
pub(super) pending_welcomes: RwLock<HashMap<String, PendingWelcome>>,
pub(super) pending_welcome_receives: RwLock<HashMap<String, PendingWelcomeReceive>>,
pub(super) pending_welcome_waiters: RwLock<HashMap<String, Vec<WelcomeFetchWaiter>>>,
pub(super) pending_welcome_acks: RwLock<HashMap<String, Arc<FileChunkAckSlot>>>,
pub(super) treekem_pending_events:
RwLock<HashMap<String, VecDeque<PendingTreeKemMetadataEvent>>>,
pub(super) treekem_event_log: RwLock<HashMap<String, VecDeque<NamedGroupMetadataEvent>>>,
pub(super) treekem_member_key_packages: super::TreeKemMemberKeyPackageCache,
pub(super) treekem_catchup_throttle: RwLock<HashMap<String, Instant>>,
pub(super) group_membership_locks: RwLock<HashMap<String, Arc<Mutex<()>>>>,
pub(super) treekem_groups:
RwLock<HashMap<String, Arc<tokio::sync::Mutex<x0x::mls::TreeKemMlsGroup>>>>,
pub(super) treekem_dir: PathBuf,
pub(super) ws_sessions: RwLock<HashMap<String, WsSession>>,
pub(super) ws_topics: RwLock<HashMap<String, SharedTopicState>>,
pub(super) ws_outbound_stats: Arc<WsOutboundStats>,
pub(super) api_address: SocketAddr,
pub(super) start_time: Instant,
pub(super) broadcast_tx: broadcast::Sender<SseEvent>,
pub(super) file_transfers: RwLock<HashMap<String, x0x::files::TransferState>>,
pub(super) receive_hashers: RwLock<HashMap<String, Sha256>>,
pub(super) pending_file_chunks: RwLock<HashMap<String, BTreeMap<u64, Vec<u8>>>>,
pub(super) file_chunk_acks: RwLock<HashMap<String, Arc<FileChunkAckSlot>>>,
pub(super) transfers_dir: PathBuf,
pub(super) shutdown_tx: mpsc::Sender<()>,
pub(super) shutdown_notify: watch::Sender<bool>,
pub(super) update_config: DaemonUpdateConfig,
pub(super) self_update_enabled: bool,
pub(super) upgrade_check_cache: Mutex<Option<CachedUpgradeCheck>>,
pub(super) upgrade_apply_lock: Arc<Mutex<()>>,
pub(super) api_token: String,
pub(super) sessions: SessionStore,
pub(super) exec_service: Arc<x0x::exec::ExecService>,
pub(super) groups_diagnostics: Arc<x0x::groups::GroupsDiagnostics>,
pub(super) connect_diagnostics: Arc<x0x::connect::ConnectDiagnostics>,
pub(super) forward_service: Option<Arc<x0x::forward::ForwardService>>,
}
#[derive(Clone)]
pub(super) struct CachedUpgradeCheck {
pub(super) checked_at: Instant,
pub(super) status: StatusCode,
pub(super) body: serde_json::Value,
pub(super) ttl: Duration,
}
#[cfg(test)]
mod tests {
use super::*;
const GRAMMAR_ERR: &str =
"instance name must start with alphanumeric and contain only alphanumeric or hyphens";
const LENGTH_ERR: &str = "instance name must be 1-64 characters";
#[test]
fn try_from_rejects_path_traversal_separators_and_invalid_grammar() {
let overlength = "a".repeat(65); let cases: &[(&str, &str)] = &[
("/", GRAMMAR_ERR),
("\\", GRAMMAR_ERR),
("a/b", GRAMMAR_ERR),
("a\\b", GRAMMAR_ERR),
("..", GRAMMAR_ERR),
("../etc/passwd", GRAMMAR_ERR),
("/etc/passwd", GRAMMAR_ERR),
("", LENGTH_ERR),
(" ", GRAMMAR_ERR),
("ab cd", GRAMMAR_ERR),
("-lead", GRAMMAR_ERR),
("café", GRAMMAR_ERR),
(overlength.as_str(), LENGTH_ERR),
];
for &(raw, expected) in cases {
let err = InstanceName::try_from(raw.to_owned())
.expect_err("invalid instance name must be rejected");
let msg = err.to_string();
assert!(
msg.contains(expected),
"{raw:?}: expected error containing {expected:?}, got {msg:?}"
);
}
}
#[test]
fn try_from_accepts_valid_names_and_preserves_bytes() {
let max_len = "a".repeat(64);
let cases: &[&str] = &[
"a", "testnet", "x0x-443", "ProdNode1", "build-", max_len.as_str(), ];
for &raw in cases {
let name = InstanceName::try_from(raw.to_owned())
.unwrap_or_else(|e| panic!("{raw:?}: expected valid, got {e}"));
assert_eq!(
name.as_str(),
raw,
"an accepted name must be preserved byte-for-byte (no trim/lowercase)"
);
}
}
#[test]
fn borrowed_validator_matches_typed_constructor_outcomes_and_messages() {
let validate: fn(&str) -> anyhow::Result<()> = x0x::server::validate_instance_name;
let max = "a".repeat(64); let over = "a".repeat(65);
let cases: &[&str] = &[
"a", "testnet", "x0x-443", max.as_str(), "",
" ", "ab cd", "\t", "/", "\\", "a/b", "a\\b", "..", "../etc/passwd", "/etc/passwd", "C:\\Users", "\\\\server\\share", "-lead",
"café", "测试", over.as_str(),
];
for &raw in cases {
let borrowed = validate(raw);
let typed = InstanceName::try_from(raw.to_owned());
match (&borrowed, &typed) {
(Ok(()), Ok(_)) => {}
(Err(borrowed_err), Err(typed_err)) => assert_eq!(
borrowed_err.to_string(),
typed_err.to_string(),
"borrowed and typed validators disagree on error message for {raw:?}"
),
(Ok(()), Err(typed_err)) => panic!(
"parity break: borrowed validator ACCEPTED {raw:?} \
but typed constructor rejected it: {typed_err}"
),
(Err(borrowed_err), Ok(_)) => panic!(
"parity break: borrowed validator REJECTED {raw:?} \
but typed constructor accepted it: {borrowed_err}"
),
}
}
}
}