| 1 |
|
| 2 |
|
| 3 |
|
| 4 |
|
| 5 |
use super::{ui, Ctx}; |
| 6 |
use anyhow::{bail, Context, Result}; |
| 7 |
use spectra_core::daemon::Daemon; |
| 8 |
use spectra_mcp::client::Client; |
| 9 |
use spectra_mcp::http::{AppState, PeerInfo}; |
| 10 |
use std::net::SocketAddr; |
| 11 |
use std::sync::Arc; |
| 12 |
|
| 13 |
|
| 14 |
struct ServerCfg { |
| 15 |
bind: String, |
| 16 |
tls: String, |
| 17 |
files: Option<(std::path::PathBuf, std::path::PathBuf)>, |
| 18 |
names: Vec<String>, |
| 19 |
} |
| 20 |
|
| 21 |
fn server_cfg(ctx: &Ctx, bind: Option<String>, tls: Option<String>) -> Result<ServerCfg> { |
| 22 |
let text = std::fs::read_to_string(ctx.data.join("spectra.toml")).unwrap_or_default(); |
| 23 |
let t: toml::Table = text.parse().with_context(|| format!("reading {}", ctx.data.join("spectra.toml").display()))?; |
| 24 |
let server = t.get("server").and_then(|s| s.as_table()).cloned().unwrap_or_default(); |
| 25 |
let (mut mode, mut files) = (String::from("auto"), None); |
| 26 |
match server.get("tls") { |
| 27 |
Some(toml::Value::String(m)) => mode = m.clone(), |
| 28 |
Some(toml::Value::Table(f)) => { |
| 29 |
let path = |k: &str| f.get(k).and_then(|v| v.as_str()).map(std::path::PathBuf::from); |
| 30 |
files = Some((path("cert").context("[server] tls needs cert")?, path("key").context("[server] tls needs key")?)); |
| 31 |
mode = "on".into(); |
| 32 |
} |
| 33 |
_ => {} |
| 34 |
} |
| 35 |
if let Some(m) = tls { |
| 36 |
mode = m; |
| 37 |
} |
| 38 |
if !matches!(mode.as_str(), "auto" | "on" | "off") { |
| 39 |
bail!("TLS is auto, on or off, not {mode}"); |
| 40 |
} |
| 41 |
let names = match server.get("names").and_then(|n| n.as_array()) { |
| 42 |
Some(a) => a.iter().filter_map(|v| v.as_str().map(String::from)).collect(), |
| 43 |
None => default_names(), |
| 44 |
}; |
| 45 |
let bind = bind.or_else(|| server.get("bind").and_then(|b| b.as_str()).map(String::from)).unwrap_or_else(|| "127.0.0.1:7350".into()); |
| 46 |
Ok(ServerCfg { bind, tls: mode, files, names }) |
| 47 |
} |
| 48 |
|
| 49 |
|
| 50 |
fn default_names() -> Vec<String> { |
| 51 |
let mut names = Vec::new(); |
| 52 |
let short = std::fs::read_to_string("/etc/hostname").ok().map(|h| h.trim().to_string()).filter(|h| !h.is_empty()); |
| 53 |
let fqdn = std::process::Command::new("hostname") |
| 54 |
.arg("--fqdn") |
| 55 |
.output() |
| 56 |
.ok() |
| 57 |
.filter(|o| o.status.success()) |
| 58 |
.map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string()) |
| 59 |
.filter(|h| !h.is_empty()); |
| 60 |
for n in [short, fqdn, Some("localhost".to_string())].into_iter().flatten() { |
| 61 |
if !names.contains(&n) { |
| 62 |
names.push(n); |
| 63 |
} |
| 64 |
} |
| 65 |
names |
| 66 |
} |
| 67 |
|
| 68 |
type Served = std::pin::Pin<Box<dyn std::future::Future<Output = std::io::Result<()>> + Send>>; |
| 69 |
|
| 70 |
pub async fn serve(ctx: &Ctx, bind: Option<String>, tls: Option<String>, insecure: bool) -> Result<i32> { |
| 71 |
let cfg = server_cfg(ctx, bind, tls)?; |
| 72 |
let addr: SocketAddr = cfg.bind.parse().with_context(|| format!("{} is not an address and port", cfg.bind))?; |
| 73 |
let use_tls = match cfg.tls.as_str() { |
| 74 |
"on" => true, |
| 75 |
"off" => false, |
| 76 |
_ => !addr.ip().is_loopback(), |
| 77 |
}; |
| 78 |
if !addr.ip().is_loopback() && !use_tls && !insecure { |
| 79 |
bail!("Spectra serves other machines over TLS only (spec §21.1); drop --tls off, or pass --insecure to bind {addr} in plain HTTP anyway"); |
| 80 |
} |
| 81 |
let socket = ctx.socket(); |
| 82 |
if socket.exists() { |
| 83 |
if Client::alive(&socket).await { |
| 84 |
bail!("Spectra is already serving {} (its socket answers)", ctx.data.display()); |
| 85 |
} |
| 86 |
std::fs::remove_file(&socket).ok(); |
| 87 |
} |
| 88 |
let daemon = Daemon::start(&ctx.data).await?; |
| 89 |
let unix = tokio::net::UnixListener::bind(&socket).with_context(|| format!("binding {}", socket.display()))?; |
| 90 |
#[cfg(unix)] |
| 91 |
{ |
| 92 |
use std::os::unix::fs::PermissionsExt; |
| 93 |
std::fs::set_permissions(&socket, std::fs::Permissions::from_mode(0o660))?; |
| 94 |
} |
| 95 |
let tcp = tokio::net::TcpListener::bind(addr).await.with_context(|| format!("binding {addr}"))?; |
| 96 |
let port = tcp.local_addr()?.port(); |
| 97 |
let mut origins = vec![format!("http://127.0.0.1:{port}"), format!("http://localhost:{port}"), format!("http://{addr}")]; |
| 98 |
let tls = if use_tls { |
| 99 |
let source = match &cfg.files { |
| 100 |
Some((cert, key)) => spectra_mcp::tls::CertSource::Files { cert: cert.clone(), key: key.clone() }, |
| 101 |
None => spectra_mcp::tls::CertSource::Own { names: cfg.names.clone() }, |
| 102 |
}; |
| 103 |
let t = spectra_mcp::tls::ServerTls::load(&ctx.data, &source)?; |
| 104 |
let instance = spectra_core::keys::KeyPair::instance(&ctx.data, "spectra")?; |
| 105 |
spectra_mcp::instance::record_tls(&daemon.ledger, &instance, &t.spki_sha256, &t.names).await?; |
| 106 |
for n in &cfg.names { |
| 107 |
origins.push(format!("https://{n}:{port}")); |
| 108 |
} |
| 109 |
origins.push(format!("https://{addr}")); |
| 110 |
Some(t) |
| 111 |
} else { |
| 112 |
None |
| 113 |
}; |
| 114 |
let state = AppState { daemon: daemon.clone(), origins: Arc::new(origins) }; |
| 115 |
let app = spectra_mcp::http::router(state.clone()).merge(spectra_mcp::api::router(state)); |
| 116 |
|
| 117 |
let (head, _) = daemon.ledger.head().await; |
| 118 |
let agents = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM agents WHERE disabled_ms IS NULL") |
| 119 |
.fetch_one(daemon.ledger.pool()) |
| 120 |
.await |
| 121 |
.unwrap_or(0); |
| 122 |
println!("{} {}", ui::done("Serving"), ui::subject("Spectra")); |
| 123 |
println!("{}", ui::kv("socket", ui::path(socket.display()))); |
| 124 |
match &tls { |
| 125 |
Some(t) => { |
| 126 |
let host = if addr.ip().is_unspecified() { |
| 127 |
cfg.names.first().cloned().unwrap_or_else(|| "localhost".into()) |
| 128 |
} else { |
| 129 |
addr.ip().to_string() |
| 130 |
}; |
| 131 |
println!("{}", ui::kv("https", format!("https://{host}:{port}"))); |
| 132 |
println!( |
| 133 |
"{}", |
| 134 |
ui::kv( |
| 135 |
"certificate", |
| 136 |
format!("{}{}", ui::hash(hex::encode(&t.spki_sha256[..8])), ui::dim(" (signed by the instance key)")) |
| 137 |
) |
| 138 |
); |
| 139 |
} |
| 140 |
None => println!("{}", ui::kv("http", format!("http://{}:{port}", addr.ip()))), |
| 141 |
} |
| 142 |
println!("{}", ui::kv("ledger", format!("{head} events{}{agents} agent(s)", ui::sep()))); |
| 143 |
println!(" {} {}", ui::note(), ui::dim("agents connect with: spectra mcp --agent NAME")); |
| 144 |
|
| 145 |
let (stop_tx, stop_rx) = tokio::sync::watch::channel(false); |
| 146 |
let stop = |mut rx: tokio::sync::watch::Receiver<bool>| async move { |
| 147 |
let _ = rx.wait_for(|s| *s).await; |
| 148 |
}; |
| 149 |
let unix_srv = |
| 150 |
axum::serve(unix, app.clone().into_make_service_with_connect_info::<PeerInfo>()).with_graceful_shutdown(stop(stop_rx.clone())); |
| 151 |
let tcp_srv: Served = match &tls { |
| 152 |
Some(t) => { |
| 153 |
let listener = spectra_mcp::tls::TlsListener::new(tcp, t.config.clone())?; |
| 154 |
let srv = axum::serve(listener, app.into_make_service_with_connect_info::<PeerInfo>()).with_graceful_shutdown(stop(stop_rx)); |
| 155 |
Box::pin(async move { srv.await }) |
| 156 |
} |
| 157 |
None => { |
| 158 |
let srv = axum::serve(tcp, app.into_make_service_with_connect_info::<PeerInfo>()).with_graceful_shutdown(stop(stop_rx)); |
| 159 |
Box::pin(async move { srv.await }) |
| 160 |
} |
| 161 |
}; |
| 162 |
let signal = async move { |
| 163 |
shutdown_signal().await; |
| 164 |
let _ = stop_tx.send(true); |
| 165 |
}; |
| 166 |
|
| 167 |
|
| 168 |
let (stopping_tx, mut stopping) = tokio::sync::watch::channel(false); |
| 169 |
tokio::spawn(async move { |
| 170 |
signal.await; |
| 171 |
let _ = stopping_tx.send(true); |
| 172 |
}); |
| 173 |
let servers = async { tokio::join!(unix_srv, tcp_srv) }; |
| 174 |
tokio::select! { |
| 175 |
(u, t) = servers => { |
| 176 |
u?; |
| 177 |
t?; |
| 178 |
} |
| 179 |
_ = async { |
| 180 |
let _ = stopping.wait_for(|s| *s).await; |
| 181 |
tokio::time::sleep(std::time::Duration::from_secs(5)).await; |
| 182 |
} => { |
| 183 |
tracing::info!("connections still open five seconds after the signal; stopping anyway"); |
| 184 |
} |
| 185 |
} |
| 186 |
daemon.shutdown().await; |
| 187 |
std::fs::remove_file(&socket).ok(); |
| 188 |
println!("{} {}", ui::done("Stopped"), ui::dim(format!("ledger checkpointed at {} events", daemon.ledger.head().await.0))); |
| 189 |
Ok(0) |
| 190 |
} |
| 191 |
|
| 192 |
async fn shutdown_signal() { |
| 193 |
#[cfg(unix)] |
| 194 |
{ |
| 195 |
let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()).expect("a SIGTERM handler"); |
| 196 |
tokio::select! { |
| 197 |
_ = tokio::signal::ctrl_c() => {} |
| 198 |
_ = term.recv() => {} |
| 199 |
} |
| 200 |
} |
| 201 |
#[cfg(not(unix))] |
| 202 |
let _ = tokio::signal::ctrl_c().await; |
| 203 |
} |