OXYGENHUB
# 67bfa06a894f spectra / src / cli / serve.rs
9.1 KiB · Rust · last changed Serve other machines over TLS, trusted through the instance key (#20) 1 day ago source blame history raw
1
//! `spectra serve`: the daemon (spec §7.1, §21.1). It holds the ledger,
2
//! serves Tuner, the API and Radar on the Unix socket and on a TCP port,
3
//! over TLS beyond loopback (§8.6), and writes a checkpoint when it stops.
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
/// `[server]` of spectra.toml, with the command line's overrides.
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
/// The host name, its fully qualified name, and localhost.
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, open: 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), sessions: Default::default() };
115
    let app = spectra_mcp::http::router(state.clone()).merge(spectra_mcp::api::router(state.clone())).merge(radar::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
    println!("  {} {}", ui::note(), ui::dim("open Radar with: spectra open"));
145
    if open {
146
        let ctx2 = Ctx { data: ctx.data.clone() };
147
        tokio::spawn(async move {
148
            // Once the socket answers, sign the link as the owner and open it.
149
            tokio::time::sleep(std::time::Duration::from_millis(200)).await;
150
            if let Err(e) = super::open::open(&ctx2, None, true).await {
151
                eprintln!("{} {e:#}", ui::caution());
152
            }
153
        });
154
    }
155
156
    let (stop_tx, stop_rx) = tokio::sync::watch::channel(false);
157
    let stop = |mut rx: tokio::sync::watch::Receiver<bool>| async move {
158
        let _ = rx.wait_for(|s| *s).await;
159
    };
160
    let unix_srv =
161
        axum::serve(unix, app.clone().into_make_service_with_connect_info::<PeerInfo>()).with_graceful_shutdown(stop(stop_rx.clone()));
162
    let tcp_srv: Served = match &tls {
163
        Some(t) => {
164
            let listener = spectra_mcp::tls::TlsListener::new(tcp, t.config.clone())?;
165
            let srv = axum::serve(listener, app.into_make_service_with_connect_info::<PeerInfo>()).with_graceful_shutdown(stop(stop_rx));
166
            Box::pin(async move { srv.await })
167
        }
168
        None => {
169
            let srv = axum::serve(tcp, app.into_make_service_with_connect_info::<PeerInfo>()).with_graceful_shutdown(stop(stop_rx));
170
            Box::pin(async move { srv.await })
171
        }
172
    };
173
    let signal = async move {
174
        shutdown_signal().await;
175
        let _ = stop_tx.send(true);
176
    };
177
    // After the signal, open connections get five seconds to finish: a
178
    // `subscriptions/listen` stream never does by itself.
179
    let (stopping_tx, mut stopping) = tokio::sync::watch::channel(false);
180
    tokio::spawn(async move {
181
        signal.await;
182
        let _ = stopping_tx.send(true);
183
    });
184
    let servers = async { tokio::join!(unix_srv, tcp_srv) };
185
    tokio::select! {
186
        (u, t) = servers => {
187
            u?;
188
            t?;
189
        }
190
        _ = async {
191
            let _ = stopping.wait_for(|s| *s).await;
192
            tokio::time::sleep(std::time::Duration::from_secs(5)).await;
193
        } => {
194
            tracing::info!("connections still open five seconds after the signal; stopping anyway");
195
        }
196
    }
197
    daemon.shutdown().await;
198
    std::fs::remove_file(&socket).ok();
199
    println!("{} {}", ui::done("Stopped"), ui::dim(format!("ledger checkpointed at {} events", daemon.ledger.head().await.0)));
200
    Ok(0)
201
}
202
203
async fn shutdown_signal() {
204
    #[cfg(unix)]
205
    {
206
        let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()).expect("a SIGTERM handler");
207
        tokio::select! {
208
            _ = tokio::signal::ctrl_c() => {}
209
            _ = term.recv() => {}
210
        }
211
    }
212
    #[cfg(not(unix))]
213
    let _ = tokio::signal::ctrl_c().await;
214
}