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    /// `host:port` on any address, TLS with required client certificates:
69    /// an agent's listener for its control plane (docs/guides/servers.md).
70    Mtls(String, http::TlsConfig),
71}
72
73impl std::fmt::Debug for ListenerKind {
74    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
75        match self {
76            ListenerKind::Tcp(a) => write!(f, "Tcp({a:?})"),
77            ListenerKind::Unix(p) => write!(f, "Unix({p:?})"),
78            ListenerKind::Mtls(a, _) => write!(f, "Mtls({a:?})"),
79        }
80    }
81}
82
83/// One address the server answers on, with its own gate and tool policy.
84#[derive(Clone)]
85pub struct Listener {
86    pub kind: ListenerKind,
87    /// Required on TCP unless `allow_unauthenticated`; refused on unix.
88    pub access: Option<Arc<AccessValidator>>,
89    pub policy: ToolPolicy,
90    /// Serve TCP with no Access validation, trusting whatever reaches the port.
91    pub allow_unauthenticated: bool,
92    /// A TCP listener on a tailnet address rather than loopback.
93    pub tailnet: bool,
94    /// Paths other than `/healthz` and `/mcp`.
95    pub routes: Option<Routes>,
96    /// Routes that authenticate every request themselves and are served
97    /// even with Access configured (webhooks, signed by their sender).
98    pub public_routes: Option<Routes>,
99    /// Requests for preview hosts, by `Host`, ahead of everything else.
100    pub preview: Option<Routes>,
101    /// Authentication and authorization the embedder supplies.
102    pub hooks: mcp::Hooks,
103}
104
105impl std::fmt::Debug for Listener {
106    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
107        f.debug_struct("Listener")
108            .field("kind", &self.kind)
109            .field("access", &self.access)
110            .field("policy", &self.policy)
111            .field("allow_unauthenticated", &self.allow_unauthenticated)
112            .field("routes", &self.routes.is_some())
113            .finish()
114    }
115}
116
117impl Listener {
118    pub fn tcp(addr: impl Into<String>) -> Self {
119        Self::new(ListenerKind::Tcp(addr.into()))
120    }
121
122    pub fn unix(path: impl Into<PathBuf>) -> Self {
123        Self::new(ListenerKind::Unix(path.into()))
124    }
125
126    /// TLS on any address, admitting only clients whose certificate `tls`
127    /// verifies.
128    pub fn mtls(addr: impl Into<String>, tls: http::TlsConfig) -> Self {
129        Self::new(ListenerKind::Mtls(addr.into(), tls))
130    }
131
132    fn new(kind: ListenerKind) -> Self {
133        Listener {
134            kind,
135            access: None,
136            policy: ToolPolicy::default(),
137            allow_unauthenticated: false,
138            tailnet: false,
139            routes: None,
140            public_routes: None,
141            preview: None,
142            hooks: mcp::Hooks::default(),
143        }
144    }
145
146    /// Serve `routes` on this listener too.
147    pub fn hooks(mut self, h: mcp::Hooks) -> Self {
148        self.hooks = h;
149        self
150    }
151
152    pub fn routes(mut self, r: Routes) -> Self {
153        self.routes = Some(r);
154        self
155    }
156
157    /// Serve `r` ahead of Access: only for routes whose every request
158    /// carries its own credential.
159    pub fn public_routes(mut self, r: Routes) -> Self {
160        self.public_routes = Some(r);
161        self
162    }
163
164    /// Let `r` take requests by their `Host` before anything else: it
165    /// answers only for hosts that are its own.
166    pub fn preview(mut self, r: Routes) -> Self {
167        self.preview = Some(r);
168        self
169    }
170
171    pub fn access(mut self, v: AccessValidator) -> Self {
172        self.access = Some(Arc::new(v));
173        self
174    }
175
176    /// [`Self::access`], sharing a validator (and its key cache).
177    pub fn access_shared(mut self, v: Arc<AccessValidator>) -> Self {
178        self.access = Some(v);
179        self
180    }
181
182    /// Bind a tailnet address instead of loopback (no Access: only tailnet
183    /// peers reach it).
184    pub fn tailnet(mut self, yes: bool) -> Self {
185        self.tailnet = yes;
186        self
187    }
188
189    pub fn policy(mut self, p: ToolPolicy) -> Self {
190        self.policy = p;
191        self
192    }
193
194    pub fn allow_unauthenticated(mut self, yes: bool) -> Self {
195        self.allow_unauthenticated = yes;
196        self
197    }
198
199    /// The unix socket is trusted as itself; a TCP caller is trusted only
200    /// as a superadmin, per request.
201    pub fn is_trusted(&self) -> bool {
202        matches!(self.kind, ListenerKind::Unix(_))
203    }
204
205    fn check(&self) -> Result<()> {
206        match (&self.kind, &self.access) {
207            (ListenerKind::Unix(p), Some(_)) => Err(Error::invalid(format!(
208                "unix socket {}: Cloudflare Access applies to TCP listeners only",
209                p.display()
210            ))),
211            (ListenerKind::Mtls(a, _), Some(_)) => Err(Error::invalid(format!(
212                "{a}: Cloudflare Access does not apply to an mTLS listener"
213            ))),
214            (ListenerKind::Tcp(a), Some(_)) if self.tailnet => Err(Error::invalid(format!(
215                "{a}: Cloudflare Access applies to loopback listeners (behind the tunnel), not a tailnet one"
216            ))),
217            (ListenerKind::Tcp(a), None) if !self.allow_unauthenticated => {
218                Err(Error::invalid(format!(
219                    "TCP listener {a} needs Cloudflare Access (team domain and audience), \
220                     or an explicit opt-in to serve it unauthenticated"
221                )))
222            }
223            _ => Ok(()),
224        }
225    }
226
227    fn describe(&self, tools: usize) -> String {
228        match (&self.kind, &self.access) {
229            (ListenerKind::Unix(p), _) => {
230                format!("unix:{} (trusted local, {tools} tools)", p.display())
231            }
232            (ListenerKind::Mtls(a, _), _) => {
233                format!("https://{a} (control plane mTLS only, {tools} tools)")
234            }
235            (ListenerKind::Tcp(a), Some(v)) => format!(
236                "http://{a}/mcp (Cloudflare Access: {}, {tools} tools)",
237                v.issuer()
238            ),
239            (ListenerKind::Tcp(a), None) if self.tailnet => format!(
240                "http://{a}/mcp on the tailnet ({tools} tools): callers sign in with isb API tokens \
241                 or sessions, or are superadmins by tailnet identity"
242            ),
243            (ListenerKind::Tcp(a), None) if self.hooks.authorize.is_some() => format!(
244                "http://{a}/mcp ({tools} tools) without Cloudflare Access: callers sign in \
245                 with isb API tokens or sessions"
246            ),
247            (ListenerKind::Tcp(a), None) => format!(
248                "http://{a}/mcp ({tools} tools) WITHOUT Cloudflare Access: anything that \
249                 reaches this port can call these tools"
250            ),
251        }
252    }
253}
254
255/// Where the CLI and the server meet: `$ISB_SERVE_SOCKET`, else
256/// `$XDG_RUNTIME_DIR/isb/serve.sock`, else a per-uid directory under /tmp.
257/// On macOS, where the daemon runs inside the `isb machine`, it is that
258/// machine's forwarded socket, `~/.isb/machine/isb/serve.sock`.
259pub fn default_socket_path() -> PathBuf {
260    if let Some(s) = std::env::var_os("ISB_SERVE_SOCKET").filter(|s| !s.is_empty()) {
261        return PathBuf::from(s);
262    }
263    #[cfg(target_os = "macos")]
264    if let Ok(s) = crate::machine::serve_socket(crate::machine::DEFAULT_NAME) {
265        return s;
266    }
267    if let Some(d) = std::env::var_os("XDG_RUNTIME_DIR").filter(|s| !s.is_empty()) {
268        return PathBuf::from(d).join("isb/serve.sock");
269    }
270    std::env::temp_dir()
271        .join(format!("isb-{}", rustix::process::getuid().as_raw()))
272        .join("serve.sock")
273}
274
275/// Serve until SIGINT or SIGTERM.
276pub fn serve(listeners: Vec<Listener>, registry: Registry, healthz: Healthz) -> Result<()> {
277    serve_until(listeners, registry, healthz, Shutdown::on_signals()?)
278}
279
280/// [`serve`], with a registry the embedder also keeps (to serve it on
281/// listeners it adds later, [`spawn_private`]).
282pub fn serve_shared(
283    listeners: Vec<Listener>,
284    registry: Arc<Registry>,
285    healthz: Healthz,
286) -> Result<()> {
287    serve_until_shared(listeners, registry, healthz, Shutdown::on_signals()?)
288}
289
290/// What a listener's requests go to: `/healthz`, `/mcp`, the REST surface
291/// and its routes, through its hooks and policy.
292pub fn handler(l: &Listener, registry: Arc<Registry>, healthz: Healthz) -> Handler {
293    let ep = Endpoint {
294        registry,
295        policy: l.policy.clone(),
296        access: l.access.clone(),
297        healthz,
298        routes: l.routes.clone(),
299        public_routes: l.public_routes.clone(),
300        hooks: l.hooks.clone(),
301    };
302    let preview = l.preview.clone();
303    Arc::new(move |r: &http::Request| {
304        if let Some(resp) = preview.as_ref().and_then(|p| p(r)) {
305            return resp;
306        }
307        ep.handle(r)
308    })
309}
310
311/// Serve `handler` on a private (RFC 1918) address that is not loopback,
312/// such as an org bridge's, until `stop` (or a bind failure, returned at
313/// once). The caller's `handler` decides who gets in: nothing about such
314/// an address keeps anyone out.
315pub fn spawn_private(
316    addr: std::net::SocketAddr,
317    handler: Handler,
318    stop: Shutdown,
319) -> Result<std::thread::JoinHandle<()>> {
320    let sock = HttpListener::bind_tcp_private(addr)?;
321    let server = HttpServer::new(Limits::default(), stop);
322    std::thread::Builder::new()
323        .name(format!("isb-listen-{addr}"))
324        .spawn(move || {
325            if let Err(e) = server.run(sock, handler) {
326                eprintln!("isb serve: listener {addr}: {e}");
327            }
328            server.drain(Duration::from_secs(5));
329        })
330        .map_err(|e| Error::Protocol(format!("cannot start a listener thread: {e}")))
331}
332
333/// Serve until `shutdown` is triggered, then give in-flight requests up to 10s.
334/// Every listener is bound before any is served, so a bad one fails startup.
335pub fn serve_until(
336    listeners: Vec<Listener>,
337    registry: Registry,
338    healthz: Healthz,
339    shutdown: Shutdown,
340) -> Result<()> {
341    serve_until_shared(listeners, Arc::new(registry), healthz, shutdown)
342}
343
344/// [`serve_until`] with a shared registry.
345pub fn serve_until_shared(
346    listeners: Vec<Listener>,
347    registry: Arc<Registry>,
348    healthz: Healthz,
349    shutdown: Shutdown,
350) -> Result<()> {
351    if listeners.is_empty() {
352        return Err(Error::invalid("isb serve needs at least one listener"));
353    }
354    let mut bound: Vec<(HttpListener, Handler)> = Vec::new();
355    for l in &listeners {
356        l.check()?;
357        let sock = match &l.kind {
358            ListenerKind::Tcp(a) if l.tailnet => HttpListener::bind_tcp_tailnet(a)?,
359            ListenerKind::Tcp(a) => HttpListener::bind_tcp(a)?,
360            ListenerKind::Unix(p) => HttpListener::bind_unix(p)?,
361            ListenerKind::Mtls(a, t) => HttpListener::bind_tls(a, t.clone())?,
362        };
363        let tools = registry
364            .tools()
365            .iter()
366            .filter(|t| l.policy.allows(&t.name))
367            .count();
368        let line = l.describe(tools);
369        let mtls = matches!(l.kind, ListenerKind::Mtls(..));
370        if l.access.is_none() && !l.is_trusted() && !mtls && l.hooks.authorize.is_none() {
371            eprintln!("isb serve: WARNING: {line}");
372        } else {
373            eprintln!("isb serve: listening on {line}");
374        }
375        bound.push((sock, handler(l, registry.clone(), healthz.clone())));
376    }
377    let server = HttpServer::new(Limits::default(), shutdown.clone());
378    let threads: Vec<_> = bound
379        .into_iter()
380        .map(|(sock, h)| {
381            let (srv, stop) = (server.clone(), shutdown.clone());
382            std::thread::spawn(move || {
383                let r = srv.run(sock, h);
384                // One listener failing takes the others down with it.
385                stop.trigger();
386                r
387            })
388        })
389        .collect();
390    let mut first = Ok(());
391    for t in threads {
392        let r = t
393            .join()
394            .unwrap_or_else(|_| Err(Error::Protocol("listener thread panicked".into())));
395        if first.is_ok() {
396            first = r;
397        }
398    }
399    server.drain(Duration::from_secs(10));
400    if server.active() > 0 {
401        eprintln!(
402            "isb serve: exiting with {} request(s) still running",
403            server.active()
404        );
405    }
406    first
407}
408
409#[cfg(test)]
410mod tests {
411    use super::*;
412    use serde_json::json;
413
414    #[test]
415    fn listener_config_is_checked() {
416        assert!(Listener::tcp("127.0.0.1:0").check().is_err());
417        assert!(
418            Listener::tcp("127.0.0.1:0")
419                .allow_unauthenticated(true)
420                .check()
421                .is_ok()
422        );
423        let v = || AccessValidator::new("team.cloudflareaccess.com", "aud").unwrap();
424        assert!(Listener::unix("/x").access(v()).check().is_err());
425        assert!(Listener::tcp("127.0.0.1:0").access(v()).check().is_ok());
426        assert!(Listener::unix("/x").is_trusted());
427        let r = serve_until(
428            vec![Listener::tcp("0.0.0.0:0").allow_unauthenticated(true)],
429            Registry::new(),
430            Arc::new(|| (true, json!({}))),
431            Shutdown::new(),
432        );
433        assert!(r.unwrap_err().to_string().contains("not loopback"));
434    }
435}
isb_server/server/mod.rs

isb_server/server/
mod.rs
