isb_server/server/
mod.rs

1//! `isb serve`: the MCP server layer, with the tools supplied by the embedder.
2//!
3//! Security model:
4//! - TCP listeners bind loopback only. Remote clients arrive through a
5//!   cloudflared tunnel, behind a Cloudflare Access application with Managed
6//!   OAuth; Cloudflare runs the OAuth flow and isb stays the resource origin.
7//! - With Access configured, every `/mcp` request must carry a valid
8//!   `Cf-Access-Jwt-Assertion` for the application's audience, so a request
9//!   reaching the port by another route is still refused.
10//! - A TCP listener without Access is refused unless the embedder opts in, and
11//!   then only accepts browser requests whose `Origin` is localhost, which
12//!   blocks DNS rebinding.
13//! - The unix socket (0600, in a 0700 directory) is the trusted local path:
14//!   filesystem permissions are the gate, and its callers are
15//!   [`Caller::Local`], which [`Caller::is_trusted`] reports.
16//! - `/healthz` never requires auth and reveals only what the embedder puts in it.
17//! - A listener can carry extra [`Routes`] (`isb serve` mounts the identity
18//!   endpoints, `/api/v1/auth/*`, this way). They authenticate their own
19//!   callers; with Access configured they sit behind it, as `/mcp` does.
20//! - [`Listener::public_routes`] are served ahead of Access, for requests
21//!   that carry their own credential (app webhooks, signed by the sender).
22//! - [`Listener::preview`] sees every request first, and takes the ones
23//!   addressed to a preview host (a workspace port's own origin), which it
24//!   authenticates itself; nothing of isb's (UI, API, headers) is served
25//!   on those hosts.
26
27pub mod access;
28pub mod aliases;
29pub use isb_core::serve_client as client;
30#[cfg(test)]
31mod client_tests;
32pub mod http;
33pub mod mcp;
34pub mod openapi;
35pub mod service;
36pub mod ssh;
37pub mod ssh_config;
38pub mod tailnet;
39pub mod terminal;
40
41use std::path::PathBuf;
42use std::sync::Arc;
43use std::time::Duration;
44
45use serde_json::Value;
46
47pub use access::{AccessValidator, Identity};
48pub use http::Shutdown;
49pub use mcp::{Authenticated, Caller, Hooks, Registry, Tool, ToolHandler, ToolPolicy};
50
51use crate::error::{Error, Result};
52use http::{Handler, HttpListener, HttpServer, Limits};
53use mcp::Endpoint;
54
55/// Health for `GET /healthz`: `(ok, body)`. 200 when ok, else 503. Called on
56/// every probe, so it must be cheap.
57pub type Healthz = Arc<dyn Fn() -> (bool, Value) + Send + Sync>;
58
59/// Extra routes on a listener: `Some` answers the request, `None` leaves it
60/// to the server (a 404).
61pub type Routes = Arc<dyn Fn(&http::Request) -> Option<http::Response> + Send + Sync>;
62
63#[derive(Clone)]
64pub enum ListenerKind {
65    /// `host:port`, which must resolve to loopback only.
66    Tcp(String),
67    Unix(PathBuf),
68}
69
70impl std::fmt::Debug for ListenerKind {
71    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
72        match self {
73            ListenerKind::Tcp(a) => write!(f, "Tcp({a:?})"),
74            ListenerKind::Unix(p) => write!(f, "Unix({p:?})"),
75        }
76    }
77}
78
79/// One address the server answers on, with its own gate and tool policy.
80#[derive(Clone)]
81pub struct Listener {
82    pub kind: ListenerKind,
83    /// Required on TCP unless `allow_unauthenticated`; refused on unix.
84    pub access: Option<Arc<AccessValidator>>,
85    pub policy: ToolPolicy,
86    /// Serve TCP with no Access validation, trusting whatever reaches the port.
87    pub allow_unauthenticated: bool,
88    /// A TCP listener on a tailnet address rather than loopback.
89    pub tailnet: bool,
90    /// Paths other than `/healthz` and `/mcp`.
91    pub routes: Option<Routes>,
92    /// Routes that authenticate every request themselves and are served
93    /// even with Access configured (webhooks, signed by their sender).
94    pub public_routes: Option<Routes>,
95    /// Requests for preview hosts, by `Host`, ahead of everything else.
96    pub preview: Option<Routes>,
97    /// Authentication and authorization the embedder supplies.
98    pub hooks: mcp::Hooks,
99}
100
101impl std::fmt::Debug for Listener {
102    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
103        f.debug_struct("Listener")
104            .field("kind", &self.kind)
105            .field("access", &self.access)
106            .field("policy", &self.policy)
107            .field("allow_unauthenticated", &self.allow_unauthenticated)
108            .field("routes", &self.routes.is_some())
109            .finish()
110    }
111}
112
113impl Listener {
114    pub fn tcp(addr: impl Into<String>) -> Self {
115        Self::new(ListenerKind::Tcp(addr.into()))
116    }
117
118    pub fn unix(path: impl Into<PathBuf>) -> Self {
119        Self::new(ListenerKind::Unix(path.into()))
120    }
121
122    fn new(kind: ListenerKind) -> Self {
123        Listener {
124            kind,
125            access: None,
126            policy: ToolPolicy::default(),
127            allow_unauthenticated: false,
128            tailnet: false,
129            routes: None,
130            public_routes: None,
131            preview: None,
132            hooks: mcp::Hooks::default(),
133        }
134    }
135
136    /// Serve `routes` on this listener too.
137    pub fn hooks(mut self, h: mcp::Hooks) -> Self {
138        self.hooks = h;
139        self
140    }
141
142    pub fn routes(mut self, r: Routes) -> Self {
143        self.routes = Some(r);
144        self
145    }
146
147    /// Serve `r` ahead of Access: only for routes whose every request
148    /// carries its own credential.
149    pub fn public_routes(mut self, r: Routes) -> Self {
150        self.public_routes = Some(r);
151        self
152    }
153
154    /// Let `r` take requests by their `Host` before anything else: it
155    /// answers only for hosts that are its own.
156    pub fn preview(mut self, r: Routes) -> Self {
157        self.preview = Some(r);
158        self
159    }
160
161    pub fn access(mut self, v: AccessValidator) -> Self {
162        self.access = Some(Arc::new(v));
163        self
164    }
165
166    /// [`Self::access`], sharing a validator (and its key cache).
167    pub fn access_shared(mut self, v: Arc<AccessValidator>) -> Self {
168        self.access = Some(v);
169        self
170    }
171
172    /// Bind a tailnet address instead of loopback (no Access: only tailnet
173    /// peers reach it).
174    pub fn tailnet(mut self, yes: bool) -> Self {
175        self.tailnet = yes;
176        self
177    }
178
179    pub fn policy(mut self, p: ToolPolicy) -> Self {
180        self.policy = p;
181        self
182    }
183
184    pub fn allow_unauthenticated(mut self, yes: bool) -> Self {
185        self.allow_unauthenticated = yes;
186        self
187    }
188
189    /// The unix socket is trusted as itself; a TCP caller is trusted only
190    /// as a superadmin, per request.
191    pub fn is_trusted(&self) -> bool {
192        matches!(self.kind, ListenerKind::Unix(_))
193    }
194
195    fn check(&self) -> Result<()> {
196        match (&self.kind, &self.access) {
197            (ListenerKind::Unix(p), Some(_)) => Err(Error::invalid(format!(
198                "unix socket {}: Cloudflare Access applies to TCP listeners only",
199                p.display()
200            ))),
201            (ListenerKind::Tcp(a), Some(_)) if self.tailnet => Err(Error::invalid(format!(
202                "{a}: Cloudflare Access applies to loopback listeners (behind the tunnel), not a tailnet one"
203            ))),
204            (ListenerKind::Tcp(a), None) if !self.allow_unauthenticated => {
205                Err(Error::invalid(format!(
206                    "TCP listener {a} needs Cloudflare Access (team domain and audience), \
207                     or an explicit opt-in to serve it unauthenticated"
208                )))
209            }
210            _ => Ok(()),
211        }
212    }
213
214    fn describe(&self, tools: usize) -> String {
215        match (&self.kind, &self.access) {
216            (ListenerKind::Unix(p), _) => {
217                format!("unix:{} (trusted local, {tools} tools)", p.display())
218            }
219            (ListenerKind::Tcp(a), Some(v)) => format!(
220                "http://{a}/mcp (Cloudflare Access: {}, {tools} tools)",
221                v.issuer()
222            ),
223            (ListenerKind::Tcp(a), None) if self.tailnet => format!(
224                "http://{a}/mcp on the tailnet ({tools} tools): callers sign in with isb API tokens \
225                 or sessions, or are superadmins by tailnet identity"
226            ),
227            (ListenerKind::Tcp(a), None) if self.hooks.authorize.is_some() => format!(
228                "http://{a}/mcp ({tools} tools) without Cloudflare Access: callers sign in \
229                 with isb API tokens or sessions"
230            ),
231            (ListenerKind::Tcp(a), None) => format!(
232                "http://{a}/mcp ({tools} tools) WITHOUT Cloudflare Access: anything that \
233                 reaches this port can call these tools"
234            ),
235        }
236    }
237}
238
239/// Where the CLI and the server meet: `$ISB_SERVE_SOCKET`, else
240/// `$XDG_RUNTIME_DIR/isb/serve.sock`, else a per-uid directory under /tmp.
241/// On macOS, where the daemon runs inside the `isb machine`, it is that
242/// machine's forwarded socket, `~/.isb/machine/isb/serve.sock`.
243pub fn default_socket_path() -> PathBuf {
244    if let Some(s) = std::env::var_os("ISB_SERVE_SOCKET").filter(|s| !s.is_empty()) {
245        return PathBuf::from(s);
246    }
247    #[cfg(target_os = "macos")]
248    if let Ok(s) = crate::machine::serve_socket(crate::machine::DEFAULT_NAME) {
249        return s;
250    }
251    if let Some(d) = std::env::var_os("XDG_RUNTIME_DIR").filter(|s| !s.is_empty()) {
252        return PathBuf::from(d).join("isb/serve.sock");
253    }
254    std::env::temp_dir()
255        .join(format!("isb-{}", rustix::process::getuid().as_raw()))
256        .join("serve.sock")
257}
258
259/// Serve until SIGINT or SIGTERM.
260pub fn serve(listeners: Vec<Listener>, registry: Registry, healthz: Healthz) -> Result<()> {
261    serve_until(listeners, registry, healthz, Shutdown::on_signals()?)
262}
263
264/// [`serve`], with a registry the embedder also keeps (to serve it on
265/// listeners it adds later, [`spawn_private`]).
266pub fn serve_shared(
267    listeners: Vec<Listener>,
268    registry: Arc<Registry>,
269    healthz: Healthz,
270) -> Result<()> {
271    serve_until_shared(listeners, registry, healthz, Shutdown::on_signals()?)
272}
273
274/// What a listener's requests go to: `/healthz`, `/mcp`, the REST surface
275/// and its routes, through its hooks and policy.
276pub fn handler(l: &Listener, registry: Arc<Registry>, healthz: Healthz) -> Handler {
277    let ep = Endpoint {
278        registry,
279        policy: l.policy.clone(),
280        access: l.access.clone(),
281        healthz,
282        routes: l.routes.clone(),
283        public_routes: l.public_routes.clone(),
284        hooks: l.hooks.clone(),
285    };
286    let preview = l.preview.clone();
287    Arc::new(move |r: &http::Request| {
288        if let Some(resp) = preview.as_ref().and_then(|p| p(r)) {
289            return resp;
290        }
291        ep.handle(r)
292    })
293}
294
295/// Serve `handler` on a private (RFC 1918) address that is not loopback,
296/// such as an org bridge's, until `stop` (or a bind failure, returned at
297/// once). The caller's `handler` decides who gets in: nothing about such
298/// an address keeps anyone out.
299pub fn spawn_private(
300    addr: std::net::SocketAddr,
301    handler: Handler,
302    stop: Shutdown,
303) -> Result<std::thread::JoinHandle<()>> {
304    let sock = HttpListener::bind_tcp_private(addr)?;
305    let server = HttpServer::new(Limits::default(), stop);
306    std::thread::Builder::new()
307        .name(format!("isb-listen-{addr}"))
308        .spawn(move || {
309            if let Err(e) = server.run(sock, handler) {
310                eprintln!("isb serve: listener {addr}: {e}");
311            }
312            server.drain(Duration::from_secs(5));
313        })
314        .map_err(|e| Error::Protocol(format!("cannot start a listener thread: {e}")))
315}
316
317/// Serve until `shutdown` is triggered, then give in-flight requests up to 10s.
318/// Every listener is bound before any is served, so a bad one fails startup.
319pub fn serve_until(
320    listeners: Vec<Listener>,
321    registry: Registry,
322    healthz: Healthz,
323    shutdown: Shutdown,
324) -> Result<()> {
325    serve_until_shared(listeners, Arc::new(registry), healthz, shutdown)
326}
327
328/// [`serve_until`] with a shared registry.
329pub fn serve_until_shared(
330    listeners: Vec<Listener>,
331    registry: Arc<Registry>,
332    healthz: Healthz,
333    shutdown: Shutdown,
334) -> Result<()> {
335    if listeners.is_empty() {
336        return Err(Error::invalid("isb serve needs at least one listener"));
337    }
338    let mut bound: Vec<(HttpListener, Handler)> = Vec::new();
339    for l in &listeners {
340        l.check()?;
341        let sock = match &l.kind {
342            ListenerKind::Tcp(a) if l.tailnet => HttpListener::bind_tcp_tailnet(a)?,
343            ListenerKind::Tcp(a) => HttpListener::bind_tcp(a)?,
344            ListenerKind::Unix(p) => HttpListener::bind_unix(p)?,
345        };
346        let tools = registry
347            .tools()
348            .iter()
349            .filter(|t| l.policy.allows(&t.name))
350            .count();
351        let line = l.describe(tools);
352        if l.access.is_none() && !l.is_trusted() && l.hooks.authorize.is_none() {
353            eprintln!("isb serve: WARNING: {line}");
354        } else {
355            eprintln!("isb serve: listening on {line}");
356        }
357        bound.push((sock, handler(l, registry.clone(), healthz.clone())));
358    }
359    let server = HttpServer::new(Limits::default(), shutdown.clone());
360    let threads: Vec<_> = bound
361        .into_iter()
362        .map(|(sock, h)| {
363            let (srv, stop) = (server.clone(), shutdown.clone());
364            std::thread::spawn(move || {
365                let r = srv.run(sock, h);
366                // One listener failing takes the others down with it.
367                stop.trigger();
368                r
369            })
370        })
371        .collect();
372    let mut first = Ok(());
373    for t in threads {
374        let r = t
375            .join()
376            .unwrap_or_else(|_| Err(Error::Protocol("listener thread panicked".into())));
377        if first.is_ok() {
378            first = r;
379        }
380    }
381    server.drain(Duration::from_secs(10));
382    if server.active() > 0 {
383        eprintln!(
384            "isb serve: exiting with {} request(s) still running",
385            server.active()
386        );
387    }
388    first
389}
390
391#[cfg(test)]
392mod tests {
393    use super::*;
394    use serde_json::json;
395
396    #[test]
397    fn listener_config_is_checked() {
398        assert!(Listener::tcp("127.0.0.1:0").check().is_err());
399        assert!(
400            Listener::tcp("127.0.0.1:0")
401                .allow_unauthenticated(true)
402                .check()
403                .is_ok()
404        );
405        let v = || AccessValidator::new("team.cloudflareaccess.com", "aud").unwrap();
406        assert!(Listener::unix("/x").access(v()).check().is_err());
407        assert!(Listener::tcp("127.0.0.1:0").access(v()).check().is_ok());
408        assert!(Listener::unix("/x").is_trusted());
409        let r = serve_until(
410            vec![Listener::tcp("0.0.0.0:0").allow_unauthenticated(true)],
411            Registry::new(),
412            Arc::new(|| (true, json!({}))),
413            Shutdown::new(),
414        );
415        assert!(r.unwrap_err().to_string().contains("not loopback"));
416    }
417}
isb_server/server/mod.rs

isb_server/server/
mod.rs
