Skip to main content

isb_server/server/
mcp.rs

1//! MCP over Streamable HTTP, hand-rolled JSON-RPC 2.0.
2//!
3//! Stateless: no `Mcp-Session-Id`, and `tools/call` works without a prior
4//! `initialize`, so the CLI can make one call per connection. Every answer is
5//! plain `application/json`; the server never streams and never initiates, so
6//! `GET /mcp` (the server-to-client SSE stream) is 405.
7
8use std::net::SocketAddr;
9use std::sync::Arc;
10use std::time::Instant;
11
12use serde_json::{Value, json};
13
14use super::Healthz;
15use super::access::{ASSERTION_HEADER, AccessValidator, Identity};
16use super::http::{Peer, Request, Response};
17use crate::error::{Error, Result};
18
19mod ambient;
20mod downscope;
21mod modern;
22mod origin;
23mod skills;
24pub use ambient::ambient_ok;
25pub use origin::origin;
26pub use skills::Skill;
27
28pub use isb_core::serve_client::{LEGACY_VERSIONS, PROTOCOL_VERSIONS};
29
30const PARSE_ERROR: i64 = -32700;
31const INVALID_REQUEST: i64 = -32600;
32const METHOD_NOT_FOUND: i64 = -32601;
33const INVALID_PARAMS: i64 = -32602;
34
35/// Who made a call. Handlers use it for audit logs and to hold remote callers
36/// to a stricter policy than the local CLI.
37#[derive(Debug, Clone, PartialEq, Eq)]
38pub enum Caller {
39    /// Over the unix socket, where filesystem permissions are the gate.
40    Local { uid: Option<u32> },
41    /// Through Cloudflare Access, with a verified assertion.
42    Access(Identity),
43    /// Loopback TCP with Access validation explicitly turned off. Anyone who
44    /// can reach the port.
45    Unauthenticated { addr: SocketAddr },
46    /// A signed-in isb user: an API token, a session, or an Access identity
47    /// that maps to a user.
48    User {
49        principal: Arc<crate::auth::Principal>,
50    },
51    /// The unix socket's reach over HTTP: a superadmin token, or a tailnet
52    /// identity on the superadmin allow list ([`crate::auth::superadmin`]).
53    Superadmin(Arc<crate::auth::Superadmin>),
54}
55
56impl Caller {
57    /// A superadmin: the unix socket, or an HTTP caller with the socket's
58    /// reach. Every tool, no remote-spec policy, any instance.
59    pub fn is_trusted(&self) -> bool {
60        matches!(self, Caller::Local { .. } | Caller::Superadmin(_))
61    }
62
63    /// Literally the unix socket: the daemon's own user on this host.
64    pub fn is_local(&self) -> bool {
65        matches!(self, Caller::Local { .. })
66    }
67
68    pub fn superadmin(&self) -> Option<&crate::auth::Superadmin> {
69        match self {
70            Caller::Superadmin(s) => Some(s),
71            _ => None,
72        }
73    }
74
75    /// Where a superadmin's power comes from: `socket`, `token:<name>`,
76    /// `tailnet:<login>`; `None` for everyone else.
77    pub fn superadmin_source(&self) -> Option<String> {
78        match self {
79            Caller::Local { .. } => Some("socket".into()),
80            Caller::Superadmin(s) => Some(s.label()),
81            _ => None,
82        }
83    }
84
85    pub fn identity(&self) -> Option<&Identity> {
86        match self {
87            Caller::Access(id) => Some(id),
88            _ => None,
89        }
90    }
91
92    /// A signed-in user's principal. `None` for superadmins too, which
93    /// reach every org: code that filters by a principal's orgs treats them
94    /// as it treats the socket.
95    pub fn principal(&self) -> Option<&crate::auth::Principal> {
96        match self {
97            Caller::User { principal } => Some(principal),
98            _ => None,
99        }
100    }
101}
102
103impl std::fmt::Display for Caller {
104    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
105        match self {
106            Caller::Local { uid: Some(u) } => write!(f, "local(uid {u})"),
107            Caller::Local { uid: None } => f.write_str("local"),
108            Caller::Access(id) if id.is_service_token() => {
109                write!(f, "service-token {}", id.name())
110            }
111            Caller::Access(id) => f.write_str(id.name()),
112            Caller::Unauthenticated { addr } => write!(f, "unauthenticated {addr}"),
113            Caller::Superadmin(s) => write!(f, "superadmin {}", s.label()),
114            Caller::User { principal } => match &principal.kind {
115                crate::auth::PrincipalKind::ApiToken { .. } => {
116                    write!(f, "{} (token)", principal.user.email)
117                }
118                crate::auth::PrincipalKind::Workspace { .. } => {
119                    f.write_str(crate::auth::WORKSPACE_ACTOR)
120                }
121                _ => f.write_str(&principal.user.email),
122            },
123        }
124    }
125}
126
127pub type ToolHandler = Arc<dyn Fn(Value, &Caller) -> Result<Value> + Send + Sync>;
128
129/// One MCP tool. `handler` gets the call's `arguments` (an object, `{}` when
130/// omitted). An `Err` is reported to the client as a tool result with
131/// `isError: true`, as MCP specifies, not as a protocol error.
132#[derive(Clone)]
133pub struct Tool {
134    pub name: String,
135    pub title: Option<String>,
136    pub description: String,
137    pub input_schema: Value,
138    /// `readOnlyHint`, `destructiveHint`, `idempotentHint`, `openWorldHint`.
139    pub annotations: Option<Value>,
140    pub handler: ToolHandler,
141}
142
143impl std::fmt::Debug for Tool {
144    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
145        f.debug_struct("Tool").field("name", &self.name).finish()
146    }
147}
148
149impl Tool {
150    pub fn new(
151        name: impl Into<String>,
152        description: impl Into<String>,
153        input_schema: Value,
154        handler: impl Fn(Value, &Caller) -> Result<Value> + Send + Sync + 'static,
155    ) -> Self {
156        Tool {
157            name: name.into(),
158            title: None,
159            description: description.into(),
160            input_schema,
161            annotations: None,
162            handler: Arc::new(handler),
163        }
164    }
165
166    pub fn title(mut self, title: impl Into<String>) -> Self {
167        self.title = Some(title.into());
168        self
169    }
170
171    pub fn annotations(mut self, annotations: Value) -> Self {
172        self.annotations = Some(annotations);
173        self
174    }
175
176    fn describe(&self) -> Value {
177        let mut v = json!({
178            "name": self.name,
179            "description": self.description,
180            "inputSchema": self.input_schema,
181        });
182        if let Some(t) = &self.title {
183            v["title"] = json!(t);
184        }
185        if let Some(a) = &self.annotations {
186            v["annotations"] = a.clone();
187        }
188        v
189    }
190}
191
192/// Every tool the server offers, before any listener's policy.
193#[derive(Debug, Clone, Default)]
194pub struct Registry {
195    tools: Vec<Tool>,
196    instructions: Option<String>,
197    skills: Vec<Skill>,
198}
199
200impl Registry {
201    pub fn new() -> Self {
202        Self::default()
203    }
204
205    /// The `instructions` returned from `initialize`.
206    pub fn instructions(mut self, text: impl Into<String>) -> Self {
207        self.instructions = Some(text.into());
208        self
209    }
210
211    /// Add a tool. Names are 1-128 of `[A-Za-z0-9_.-]` and unique.
212    pub fn register(&mut self, tool: Tool) -> Result<()> {
213        let n = &tool.name;
214        if n.is_empty()
215            || n.len() > 128
216            || !n
217                .bytes()
218                .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'_' | b'-' | b'.'))
219        {
220            return Err(Error::invalid(format!("invalid tool name {n:?}")));
221        }
222        if self.get(n).is_some() {
223            return Err(Error::invalid(format!("tool {n:?} registered twice")));
224        }
225        self.tools.push(tool);
226        Ok(())
227    }
228
229    pub fn get(&self, name: &str) -> Option<&Tool> {
230        self.tools.iter().find(|t| t.name == name)
231    }
232
233    pub fn tools(&self) -> &[Tool] {
234        &self.tools
235    }
236}
237
238/// Which tools a listener exposes: the allow list (empty = all), then the deny
239/// list, which always wins. Entries are exact names or shell-style globs
240/// (`*`, `?`, `[a-z]`, `[!x]`).
241#[derive(Debug, Clone, Default, PartialEq, Eq)]
242pub struct ToolPolicy {
243    pub allow: Vec<String>,
244    pub deny: Vec<String>,
245}
246
247impl ToolPolicy {
248    /// From comma-separated lists, as flags and environment variables give them.
249    pub fn from_lists(allow: &str, deny: &str) -> Self {
250        let split = |s: &str| {
251            s.split(',')
252                .map(str::trim)
253                .filter(|p| !p.is_empty())
254                .map(String::from)
255                .collect()
256        };
257        ToolPolicy {
258            allow: split(allow),
259            deny: split(deny),
260        }
261    }
262
263    pub fn allows(&self, name: &str) -> bool {
264        (self.allow.is_empty() || self.allow.iter().any(|p| glob_match(p, name)))
265            && !self.deny.iter().any(|p| glob_match(p, name))
266    }
267}
268
269enum Tok {
270    Star,
271    Any,
272    Lit(char),
273    Class {
274        negated: bool,
275        ranges: Vec<(char, char)>,
276    },
277}
278
279fn tokenize(pattern: &str) -> Vec<Tok> {
280    let cs: Vec<char> = pattern.chars().collect();
281    let mut out = Vec::new();
282    let mut i = 0;
283    while i < cs.len() {
284        match cs[i] {
285            '*' => out.push(Tok::Star),
286            '?' => out.push(Tok::Any),
287            '[' => {
288                let mut j = i + 1;
289                let negated = matches!(cs.get(j), Some('!' | '^'));
290                if negated {
291                    j += 1;
292                }
293                let start = j;
294                // A `]` right after the opening bracket is a literal member.
295                if cs.get(j) == Some(&']') {
296                    j += 1;
297                }
298                while j < cs.len() && cs[j] != ']' {
299                    j += 1;
300                }
301                if j >= cs.len() {
302                    // Unclosed: the bracket is just a character.
303                    out.push(Tok::Lit('['));
304                    i += 1;
305                    continue;
306                }
307                let body = &cs[start..j];
308                let mut ranges = Vec::new();
309                let mut k = 0;
310                while k < body.len() {
311                    if k + 2 < body.len() && body[k + 1] == '-' {
312                        ranges.push((body[k], body[k + 2]));
313                        k += 3;
314                    } else {
315                        ranges.push((body[k], body[k]));
316                        k += 1;
317                    }
318                }
319                out.push(Tok::Class { negated, ranges });
320                i = j;
321            }
322            c => out.push(Tok::Lit(c)),
323        }
324        i += 1;
325    }
326    out
327}
328
329/// Shell-style glob over the whole name. Linear backtracking over the last
330/// `*`, so a hostile pattern cannot blow up.
331pub fn glob_match(pattern: &str, name: &str) -> bool {
332    let p = tokenize(pattern);
333    let s: Vec<char> = name.chars().collect();
334    let one = |t: &Tok, c: char| match t {
335        Tok::Any => true,
336        Tok::Lit(l) => *l == c,
337        Tok::Class { negated, ranges } => {
338            ranges.iter().any(|(a, b)| *a <= c && c <= *b) != *negated
339        }
340        Tok::Star => false,
341    };
342    let (mut pi, mut si) = (0, 0);
343    let mut back: Option<(usize, usize)> = None;
344    while si < s.len() {
345        if pi < p.len() && matches!(p[pi], Tok::Star) {
346            back = Some((pi, si));
347            pi += 1;
348        } else if pi < p.len() && one(&p[pi], s[si]) {
349            pi += 1;
350            si += 1;
351        } else if let Some((bp, bs)) = back {
352            pi = bp + 1;
353            si = bs + 1;
354            back = Some((bp, bs + 1));
355        } else {
356            return false;
357        }
358    }
359    p[pi..].iter().all(|t| matches!(t, Tok::Star))
360}
361
362/// `Origin` of a page on this machine. Without Access, a page on any other
363/// origin is a DNS-rebinding attempt to reach the loopback port.
364pub fn origin_is_local(origin: &str) -> bool {
365    let Some((scheme, rest)) = origin.trim().split_once("://") else {
366        return false;
367    };
368    if !(scheme.eq_ignore_ascii_case("http") || scheme.eq_ignore_ascii_case("https"))
369        || rest.contains(['/', '?', '#', '@'])
370    {
371        return false;
372    }
373    let (host, port) = if rest.starts_with('[') {
374        match rest.find(']') {
375            Some(i) => (&rest[..=i], &rest[i + 1..]),
376            None => return false,
377        }
378    } else {
379        match rest.find(':') {
380            Some(i) => (&rest[..i], &rest[i..]),
381            None => (rest, ""),
382        }
383    };
384    let port_ok = port.is_empty()
385        || port
386            .strip_prefix(':')
387            .is_some_and(|p| !p.is_empty() && p.bytes().all(|b| b.is_ascii_digit()));
388    port_ok && (host.eq_ignore_ascii_case("localhost") || host == "127.0.0.1" || host == "[::1]")
389}
390
391/// Who is calling, from an API token or session (`Authorization: Bearer`, or
392/// the session cookie) or from a verified Access identity.
393pub type Authn = Arc<dyn Fn(&Request, Option<&Identity>) -> Authenticated + Send + Sync>;
394
395pub enum Authenticated {
396    /// No isb credential: fall back to the listener's own notion of caller.
397    None,
398    User(Arc<crate::auth::Principal>),
399    /// A credential was presented and is not valid.
400    Refused,
401    /// The unix socket's reach: a superadmin token or a tailnet identity on
402    /// the allow list.
403    Superadmin(Arc<crate::auth::Superadmin>),
404}
405
406/// Is `path` an MCP endpoint (`/mcp`, `/orgs/<org>/mcp`)?
407fn is_mcp_path(path: &str) -> bool {
408    path == "/mcp"
409        || path
410            .strip_prefix("/orgs/")
411            .and_then(|r| r.strip_suffix("/mcp"))
412            .is_some_and(|o| !o.is_empty() && !o.contains('/'))
413}
414
415/// May `caller` run `tool` with these arguments? Returns the arguments to
416/// use (an org-scoped endpoint pins `org`), or the refusal, reported as a
417/// tool error. `scope` is the org of an `/orgs/<org>/...` endpoint.
418pub type Authorize = Arc<
419    dyn Fn(&Caller, &Tool, Value, Option<&crate::org::OrgId>) -> crate::Result<Value> + Send + Sync,
420>;
421
422/// Is `tool` listed to `caller` here? For `tools/list` only; [`Authorize`] judges calls.
423pub type Listed = Arc<dyn Fn(&Caller, &str, Option<&crate::org::OrgId>) -> bool + Send + Sync>;
424
425/// The body of `GET /api/v1/events`: a stream of server-sent events for
426/// this caller, starting after `since` (from `?since=` or `Last-Event-ID`).
427pub type Events = Arc<dyn Fn(&Caller, u64) -> crate::Result<super::http::StreamFn> + Send + Sync>;
428
429/// What the audit hook hears: a tool call, admitted or refused, or a
430/// terminal opening (`terminal.open`) and closing (`terminal.close`).
431pub struct Audited<'a> {
432    pub caller: &'a Caller,
433    /// The tool's name, or `terminal.open` / `terminal.close`.
434    pub action: &'a str,
435    /// The tool, when the action is one (its annotations say whether it
436    /// only reads).
437    pub tool: Option<&'a Tool>,
438    /// The arguments as authorized (for a refusal, as sent). The hook keeps
439    /// only what it knows is safe to keep.
440    pub args: &'a Value,
441    pub outcome: std::result::Result<(), &'a Error>,
442    pub origin: &'a crate::audit::Origin,
443}
444
445/// Records what happened; it must not fail the call.
446pub type Audit = Arc<dyn Fn(&Audited) + Send + Sync>;
447
448/// What the embedder plugs into every listener.
449#[derive(Clone, Default)]
450pub struct Hooks {
451    pub authn: Option<Authn>,
452    pub authorize: Option<Authorize>,
453    pub events: Option<Events>,
454    /// Opens a terminal for `GET /orgs/<org>/api/v1/terminal` (a websocket).
455    pub terminal: Option<super::terminal::Terminal>,
456    /// Opens an SSH session for `GET /orgs/<org>/api/v1/ssh` (a websocket).
457    pub ssh: Option<super::ssh::Ssh>,
458    /// Hears every tool call on every surface, and terminal sessions.
459    pub audit: Option<Audit>,
460    /// Which tools `tools/list` shows a caller (all, when unset).
461    pub listed: Option<Listed>,
462    /// Answer 401 to a network caller who sent no credential, on every
463    /// authenticated surface (`/mcp`, the tool list and calls, events),
464    /// instead of treating them as anonymous. Off only when the embedder
465    /// runs with anonymous access on.
466    pub refuse_anonymous: bool,
467}
468
469/// One listener's view of the server: its tools, its gate, its health.
470#[doc(hidden)]
471pub struct Endpoint {
472    pub registry: Arc<Registry>,
473    pub policy: ToolPolicy,
474    pub access: Option<Arc<AccessValidator>>,
475    pub healthz: Healthz,
476    pub routes: Option<super::Routes>,
477    pub public_routes: Option<super::Routes>,
478    pub hooks: Hooks,
479}
480
481fn rpc_error(id: Value, code: i64, message: impl Into<String>) -> Value {
482    json!({"jsonrpc": "2.0", "id": id, "error": {"code": code, "message": message.into()}})
483}
484
485type RpcResult = std::result::Result<Value, (i64, String)>;
486
487impl Endpoint {
488    #[expect(
489        clippy::excessive_nesting,
490        reason = "predates the lint ratchet; split it when next changed"
491    )]
492    pub fn handle(&self, req: &Request) -> Response {
493        match req.path.as_str() {
494            "/healthz" => match req.method.as_str() {
495                "GET" => {
496                    let (ok, v) = (self.healthz)();
497                    Response::json(if ok { 200 } else { 503 }, &v)
498                }
499                _ => Response::text(405, "method not allowed").header("Allow", "GET"),
500            },
501            "/mcp" => self.mcp(req, None),
502            "/api/v1/openapi.json" => Response::json(200, &self.openapi()),
503            "/api/v1/tools" => self.rest_list(req),
504            "/api/v1/events" => self.events(req),
505            p => {
506                if let Some(r) = self.public_routes.as_ref().and_then(|f| f(req)) {
507                    return r;
508                }
509                if let Some(rest) = p.strip_prefix("/api/v1/tools/") {
510                    return self.rest_call(req, rest, None);
511                }
512                // /orgs/<org>/mcp and /orgs/<org>/api/v1/tools/<tool>
513                if let Some(rest) = p.strip_prefix("/orgs/") {
514                    if let Some((org, tail)) = rest.split_once('/') {
515                        let Ok(org) = crate::org::OrgId::new(org) else {
516                            return Response::text(404, "no such org");
517                        };
518                        if tail == "mcp" {
519                            return self.mcp(req, Some(&org));
520                        }
521                        if let Some(tool) = tail.strip_prefix("api/v1/tools/") {
522                            return self.rest_call(req, tool, Some(&org));
523                        }
524                        if tail == "api/v1/terminal" {
525                            return self.terminal(req, &org);
526                        }
527                        if tail == "api/v1/ssh" {
528                            return self.ssh(req, &org);
529                        }
530                        if let Some(rest) = tail.strip_prefix("api/v1/workspace") {
531                            if rest.is_empty() || rest.starts_with('/') {
532                                return self.workspace_rest(req, rest, &org);
533                            }
534                        }
535                    }
536                }
537                self.extra(req)
538            }
539        }
540    }
541
542    fn mcp(&self, req: &Request, scope: Option<&crate::org::OrgId>) -> Response {
543        let caller = match self.authenticate_in(req, scope) {
544            Ok(c) => c,
545            Err(r) => return r,
546        };
547        match req.method.as_str() {
548            "POST" => self.post(req, &caller, scope, &origin(req, &caller, true)),
549            _ => Response::text(405, "method not allowed").header("Allow", "POST"),
550        }
551    }
552
553    /// The tool, if this listener offers it, with arguments the authorizer
554    /// accepted.
555    fn admit(
556        &self,
557        name: &str,
558        args: Value,
559        caller: &Caller,
560        scope: Option<&crate::org::OrgId>,
561    ) -> std::result::Result<(&Tool, Value), Admit> {
562        let tool = match self.registry.get(name) {
563            Some(t) if self.policy.allows(name) => t,
564            _ => return Err(Admit::Unknown),
565        };
566        let args = match &self.hooks.authorize {
567            Some(a) => a(caller, tool, args, scope).map_err(Admit::Refused)?,
568            None => args,
569        };
570        Ok((tool, args))
571    }
572
573    fn rest_list(&self, req: &Request) -> Response {
574        if req.method != "GET" {
575            return Response::text(405, "method not allowed").header("Allow", "GET");
576        }
577        let caller = match self.authenticate(req) {
578            Ok(c) => c,
579            Err(r) => return r,
580        };
581        // `org_endpoint`: whether an org's `/orgs/<org>/mcp` lists the tool
582        // (host, superadmin and platform tools are on `/mcp` only), so a
583        // page can count what an org connector sees.
584        let org = crate::org::OrgId::default_org();
585        let in_org = |t: &Tool| {
586            let l = &self.hooks.listed;
587            l.as_ref().is_none_or(|l| l(&caller, &t.name, Some(&org)))
588        };
589        let tools: Vec<Value> = self
590            .registry
591            .tools()
592            .iter()
593            .filter(|t| self.policy.allows(&t.name))
594            .map(|t| {
595                let mut v = t.describe();
596                v["org_endpoint"] = json!(in_org(t));
597                v
598            })
599            .collect();
600        Response::json(200, &json!({"tools": tools}))
601    }
602
603    /// `POST /api/v1/tools/<name>` with the arguments as the JSON body:
604    /// `{"result": ...}` on success, `{"error", "message", "data"}` with a
605    /// matching status otherwise.
606    fn rest_call(&self, req: &Request, name: &str, scope: Option<&crate::org::OrgId>) -> Response {
607        if req.method != "POST" {
608            return Response::text(405, "method not allowed").header("Allow", "POST");
609        }
610        self.rest_run(req, name, scope)
611    }
612
613    /// `/orgs/<org>/api/v1/workspace[/ACTION]`: the org's workspace as a
614    /// resource, over the `workspace_*` tools (docs/concepts/workspaces.md). GET
615    /// reads it (`?name=`), POST creates it, PATCH changes it, DELETE
616    /// deletes it (`{"confirm": true}`); POST `/start`, `/stop`,
617    /// `/restart`, `/rebuild`, `/token/rotate`; GET or PATCH `/settings`.
618    fn workspace_rest(&self, req: &Request, rest: &str, org: &crate::org::OrgId) -> Response {
619        let tool = match super::openapi::workspace_tool(rest, &req.method) {
620            Ok(t) => t,
621            Err(allowed) if allowed.is_empty() => {
622                return rest_error(404, "not_found", "no such workspace action");
623            }
624            Err(allowed) => {
625                return Response::text(405, "method not allowed")
626                    .header("Allow", allowed.join(", "));
627            }
628        };
629        if req.method == "GET" {
630            // Arguments from the query string.
631            let mut r = req.clone();
632            let mut args = serde_json::Map::new();
633            if let Some(n) = query_param(req, "name") {
634                args.insert("name".into(), json!(n));
635            }
636            r.body = serde_json::to_vec(&Value::Object(args)).unwrap_or_default();
637            return self.rest_run(&r, tool, Some(org));
638        }
639        self.rest_run(req, tool, Some(org))
640    }
641
642    fn rest_run(&self, req: &Request, name: &str, scope: Option<&crate::org::OrgId>) -> Response {
643        let caller = match self.authenticate_in(req, scope) {
644            Ok(c) => c,
645            Err(r) => return r,
646        };
647        let args: Value = if req.body.is_empty() {
648            json!({})
649        } else {
650            match serde_json::from_slice(&req.body) {
651                Ok(v @ Value::Object(_)) => v,
652                Ok(_) => return rest_error(400, "invalid", "the body must be a JSON object"),
653                Err(e) => return rest_error(400, "invalid", &format!("bad JSON: {e}")),
654            }
655        };
656        let origin = origin(req, &caller, false);
657        match self.call_audited(name, args, &caller, scope, &origin) {
658            Some(Ok(v)) => Response::json(200, &json!({"result": v})),
659            Some(Err(e)) => error_response(&e),
660            None => rest_error(404, "not_found", &format!("unknown tool: {name}")),
661        }
662    }
663
664    /// Admit and run a tool, telling the audit hook how it went. `None`
665    /// when this listener does not offer the tool.
666    fn call_audited(
667        &self,
668        name: &str,
669        args: Value,
670        caller: &Caller,
671        scope: Option<&crate::org::OrgId>,
672        origin: &crate::audit::Origin,
673    ) -> Option<crate::Result<Value>> {
674        // As sent, but in the org an org-bound endpoint acts in, so a
675        // refusal is filed under the org it was aimed at.
676        let sent = self.hooks.audit.as_ref().map(|_| {
677            let mut a = args.clone();
678            if let (Some(o), Some(m)) = (scope, a.as_object_mut()) {
679                m.insert("org".into(), json!(o.as_str()));
680            }
681            a
682        });
683        let (tool, args) = match self.admit(name, args, caller, scope) {
684            Ok(x) => x,
685            Err(Admit::Unknown) => return None,
686            Err(Admit::Refused(e)) => {
687                eprintln!("isb serve: {caller} called {name}: refused: {e}");
688                if let (Some(a), Some(sent)) = (&self.hooks.audit, &sent) {
689                    a(&Audited {
690                        caller,
691                        action: name,
692                        tool: self.registry.get(name),
693                        args: sent,
694                        outcome: Err(&e),
695                        origin,
696                    });
697                }
698                return Some(Err(e));
699            }
700        };
701        let kept = self.hooks.audit.as_ref().map(|_| args.clone());
702        let r = run(tool, args, caller);
703        if let (Some(a), Some(kept)) = (&self.hooks.audit, &kept) {
704            a(&Audited {
705                caller,
706                action: name,
707                tool: Some(tool),
708                args: kept,
709                outcome: r.as_ref().map(|_| ()),
710                origin,
711            });
712        }
713        Some(r)
714    }
715
716    /// `GET /orgs/<org>/api/v1/terminal?app=NAME` (or `?instance=NAME`): a
717    /// websocket to a shell, admitted as `sandbox_exec` in the org would be.
718    fn terminal(&self, req: &Request, org: &crate::org::OrgId) -> Response {
719        use super::terminal::{origin_allowed, term_request, websocket_key};
720        if req.method != "GET" {
721            return Response::text(405, "method not allowed").header("Allow", "GET");
722        }
723        let Some(open) = &self.hooks.terminal else {
724            return Response::text(404, "not found");
725        };
726        let caller = match self.authenticate_in(req, Some(org)) {
727            Ok(c) => c,
728            Err(r) => return r,
729        };
730        if !origin_allowed(req) {
731            eprintln!(
732                "isb serve: refused a terminal from origin {:?}",
733                req.header("origin")
734            );
735            return rest_error(403, "forbidden", "origin not allowed");
736        }
737        let Some(key) = websocket_key(req) else {
738            return rest_error(400, "invalid", "expected a websocket upgrade");
739        };
740        let t = match term_request(req) {
741            Ok(t) => t,
742            Err(m) => return rest_error(400, "invalid", &m),
743        };
744        let args = match &t.instance {
745            Some(i) => json!({"org": org.as_str(), "name": i}),
746            None => json!({"org": org.as_str(), "app": t.app}),
747        };
748        let (open, o) = (open.clone(), org.clone());
749        self.session(
750            req,
751            caller,
752            org,
753            Session::Terminal,
754            args,
755            &key,
756            move |c: &Caller| open(c, &o, &t),
757        )
758    }
759
760    /// `GET /orgs/<org>/api/v1/ssh?instance=NAME`: a websocket carrying an
761    /// SSH connection to the instance's sshd, admitted as `sandbox_exec` in
762    /// the org would be ([`super::ssh`]).
763    fn ssh(&self, req: &Request, org: &crate::org::OrgId) -> Response {
764        use super::ssh::{origin_allowed, ssh_request};
765        if req.method != "GET" {
766            return Response::text(405, "method not allowed").header("Allow", "GET");
767        }
768        let Some(open) = &self.hooks.ssh else {
769            return Response::text(404, "not found");
770        };
771        let caller = match self.authenticate_in(req, Some(org)) {
772            Ok(c) => c,
773            Err(r) => return r,
774        };
775        if !origin_allowed(req) {
776            eprintln!(
777                "isb serve: refused SSH from origin {:?}",
778                req.header("origin")
779            );
780            return rest_error(403, "forbidden", "origin not allowed");
781        }
782        let Some(key) = super::terminal::websocket_key(req) else {
783            return rest_error(400, "invalid", "expected a websocket upgrade");
784        };
785        let s = match ssh_request(req) {
786            Ok(s) => s,
787            Err(m) => return rest_error(400, "invalid", &m),
788        };
789        let args = json!({"org": org.as_str(), "name": s.instance});
790        let (open, o) = (open.clone(), org.clone());
791        self.session(
792            req,
793            caller,
794            org,
795            Session::Ssh,
796            args,
797            &key,
798            move |c: &Caller| open(c, &o, &s),
799        )
800    }
801
802    /// Admit a websocket session as `sandbox_exec` in `org`, then upgrade
803    /// and open it, telling the audit hook when it opens and closes.
804    #[expect(clippy::too_many_arguments)]
805    fn session<F>(
806        &self,
807        req: &Request,
808        caller: Caller,
809        org: &crate::org::OrgId,
810        kind: Session,
811        args: Value,
812        key: &str,
813        open: F,
814    ) -> Response
815    where
816        F: FnOnce(&Caller) -> crate::Result<Box<dyn super::terminal::Pty>> + Send + 'static,
817    {
818        let origin = origin(req, &caller, false);
819        let audit = self.hooks.audit.clone();
820        match self.admit("sandbox_exec", json!({}), &caller, Some(org)) {
821            Ok(_) => {}
822            Err(Admit::Unknown) => {
823                return rest_error(404, "not_found", kind.not_offered());
824            }
825            Err(Admit::Refused(e)) => {
826                if let Some(a) = &audit {
827                    a(&Audited {
828                        caller: &caller,
829                        action: kind.opened(),
830                        tool: None,
831                        args: &args,
832                        outcome: Err(&e),
833                        origin: &origin,
834                    });
835                }
836                return error_response(&e);
837            }
838        }
839        let what = args
840            .get("app")
841            .or_else(|| args.get("name"))
842            .and_then(Value::as_str)
843            .unwrap_or("")
844            .to_string();
845        eprintln!("isb serve: {caller} opened {} to {org}/{what}", kind.noun());
846        super::terminal::upgrade_with(key, kind.limits(), move || {
847            let r = open(&caller);
848            let Some(a) = audit else { return r };
849            let mut args = args;
850            if let (Ok(p), Session::Terminal, Some(_)) = (&r, kind, args.get("app")) {
851                args["replica"] = json!(p.target());
852            }
853            a(&Audited {
854                caller: &caller,
855                action: kind.opened(),
856                tool: None,
857                args: &args,
858                outcome: r.as_ref().map(|_| ()),
859                origin: &origin,
860            });
861            r.map(|inner| {
862                Box::new(AuditedPty {
863                    inner,
864                    started: Instant::now(),
865                    audit: a,
866                    caller,
867                    args,
868                    origin,
869                    closed: kind.closed(),
870                }) as Box<dyn super::terminal::Pty>
871            })
872        })
873    }
874
875    fn events(&self, req: &Request) -> Response {
876        if req.method != "GET" {
877            return Response::text(405, "method not allowed").header("Allow", "GET");
878        }
879        let caller = match self.authenticate(req) {
880            Ok(c) => c,
881            Err(r) => return r,
882        };
883        let Some(ev) = &self.hooks.events else {
884            return Response::text(404, "not found");
885        };
886        let since = req
887            .header("last-event-id")
888            .map(String::from)
889            .or_else(|| query_param(req, "since"))
890            .and_then(|s| s.parse().ok())
891            .unwrap_or(0);
892        match ev(&caller, since) {
893            Ok(f) => Response::stream(200, "text/event-stream", f),
894            Err(e) => error_response(&e),
895        }
896    }
897
898    /// The OpenAPI document of this listener's whole HTTP surface
899    /// ([`super::openapi`]).
900    fn openapi(&self) -> Value {
901        let tools: Vec<&Tool> = self
902            .registry
903            .tools()
904            .iter()
905            .filter(|t| self.policy.allows(&t.name))
906            .collect();
907        super::openapi::document(&tools)
908    }
909
910    /// The embedder's routes. They authenticate their own callers, but sit
911    /// behind Access when it is configured: Access is the front door.
912    fn extra(&self, req: &Request) -> Response {
913        let Some(routes) = &self.routes else {
914            return Response::text(404, "not found");
915        };
916        if let Some(v) = &self.access {
917            let token = req.header(ASSERTION_HEADER).unwrap_or("").trim();
918            if token.is_empty() {
919                return Response::text(401, "missing Cloudflare Access assertion");
920            }
921            if let Err(d) = v.validate(token) {
922                eprintln!("isb serve: refused {:?} {}: {d}", req.peer, req.path);
923                return Response::text(401, "invalid Cloudflare Access assertion");
924            }
925        }
926        routes(req).unwrap_or_else(|| Response::text(404, "not found"))
927    }
928
929    fn authenticate(&self, req: &Request) -> std::result::Result<Caller, Response> {
930        let bearer = req.header("authorization").is_some();
931        let cookie = req.header("cookie").is_some_and(|c| {
932            c.split(';')
933                .any(|p| p.trim_start().starts_with("isb_session="))
934        });
935        // A cookie rides along on cross-site requests; a custom header does
936        // not without a CORS preflight, which isb never grants.
937        let csrf = || {
938            if !bearer && cookie && req.method != "GET" && req.header("x-isb-csrf") != Some("1") {
939                Err(rest_error(403, "forbidden", "missing X-Isb-Csrf header"))
940            } else {
941                Ok(())
942            }
943        };
944        let user = |id: Option<&Identity>| match &self.hooks.authn {
945            Some(a) => a(req, id),
946            None => Authenticated::None,
947        };
948        let superadmin = |s: Arc<crate::auth::Superadmin>| {
949            if s.source.is_ambient() {
950                return self.refuse_ambient(
951                    req,
952                    &format!("superadmin {}", s.label()),
953                    Caller::Superadmin(s.clone()),
954                );
955            }
956            Ok(Caller::Superadmin(s))
957        };
958        // An org's tailnet or Access agent is ambient too; people's own
959        // sessions and tokens are not (a cookie has the CSRF header rule).
960        let signed_in = |p: Arc<crate::auth::Principal>| {
961            csrf()?;
962            if p.is_agent() {
963                return self.refuse_ambient(
964                    req,
965                    &format!("agent {}", p.user.email),
966                    Caller::User { principal: p },
967                );
968            }
969            Ok(Caller::User { principal: p })
970        };
971        if let Some(v) = &self.access {
972            let token = req.header(ASSERTION_HEADER).unwrap_or("").trim();
973            if token.is_empty() {
974                return Err(rest_error(
975                    401,
976                    "unauthorized",
977                    "missing Cloudflare Access assertion",
978                ));
979            }
980            let id = match v.validate(token) {
981                Ok(id) => id,
982                Err(d) => {
983                    eprintln!("isb serve: refused {:?}: {d}", req.peer);
984                    return Err(rest_error(
985                        401,
986                        "unauthorized",
987                        "invalid Cloudflare Access assertion",
988                    ));
989                }
990            };
991            return match user(Some(&id)) {
992                Authenticated::User(p) => signed_in(p),
993                Authenticated::Superadmin(s) => superadmin(s),
994                Authenticated::Refused => {
995                    Err(rest_error(401, "unauthorized", "invalid credentials"))
996                }
997                Authenticated::None => Ok(Caller::Access(id)),
998            };
999        }
1000        // Asked even without a credential: a tailnet identity is judged
1001        // from the connection itself.
1002        match user(None) {
1003            Authenticated::User(p) => return signed_in(p),
1004            Authenticated::Superadmin(s) => return superadmin(s),
1005            Authenticated::Refused => {
1006                return Err(rest_error(401, "unauthorized", "invalid credentials"));
1007            }
1008            // No authenticator, or no credential: fall through.
1009            Authenticated::None => {}
1010        }
1011        if let Some(o) = req.header("origin").filter(|o| !origin_is_local(o)) {
1012            eprintln!("isb serve: refused origin {o:?}");
1013            return Err(rest_error(403, "forbidden", "origin not allowed"));
1014        }
1015        match &req.peer {
1016            Peer::Unix { uid } => Ok(Caller::Local { uid: *uid }),
1017            Peer::Tcp(_) if self.hooks.refuse_anonymous => Err(origin::sign_in_required()),
1018            Peer::Tcp(addr) => Ok(Caller::Unauthenticated { addr: *addr }),
1019        }
1020    }
1021
1022    fn post(
1023        &self,
1024        req: &Request,
1025        caller: &Caller,
1026        scope: Option<&crate::org::OrgId>,
1027        origin: &crate::audit::Origin,
1028    ) -> Response {
1029        // Clients vary in what they send here; note oddities, never refuse.
1030        if let Some(ct) = req
1031            .header("content-type")
1032            .filter(|ct| !ct.to_ascii_lowercase().starts_with("application/json"))
1033        {
1034            eprintln!("isb serve: /mcp request with Content-Type {ct:?}");
1035        }
1036        let v: Value = match serde_json::from_slice(&req.body) {
1037            Ok(v) => v,
1038            Err(e) => {
1039                return Response::json(
1040                    400,
1041                    &rpc_error(Value::Null, PARSE_ERROR, format!("parse error: {e}")),
1042                );
1043            }
1044        };
1045        let answer = match v {
1046            Value::Array(items) if items.iter().any(|m| modern::is_modern(req, m)) => {
1047                Some(modern::batch_refused())
1048            }
1049            m if modern::is_modern(req, &m) => return self.modern(req, m, caller, scope, origin),
1050            Value::Array(items) if items.is_empty() => {
1051                Some(rpc_error(Value::Null, INVALID_REQUEST, "empty batch"))
1052            }
1053            Value::Array(items) => {
1054                let out: Vec<Value> = items
1055                    .into_iter()
1056                    .filter_map(|m| self.message(m, caller, scope, origin))
1057                    .collect();
1058                (!out.is_empty()).then_some(Value::Array(out))
1059            }
1060            m => self.message(m, caller, scope, origin),
1061        };
1062        match answer {
1063            Some(a) => Response::json(200, &a),
1064            None => Response::new(202),
1065        }
1066    }
1067
1068    /// Answer one JSON-RPC message; `None` for notifications and responses.
1069    fn message(
1070        &self,
1071        m: Value,
1072        caller: &Caller,
1073        scope: Option<&crate::org::OrgId>,
1074        origin: &crate::audit::Origin,
1075    ) -> Option<Value> {
1076        let Value::Object(mut o) = m else {
1077            return Some(rpc_error(Value::Null, INVALID_REQUEST, "invalid request"));
1078        };
1079        let id = o.remove("id");
1080        let Some(method) = o.get("method").and_then(Value::as_str).map(String::from) else {
1081            // A response to a request we never send: nothing to say.
1082            if id.is_some() && (o.contains_key("result") || o.contains_key("error")) {
1083                return None;
1084            }
1085            return Some(rpc_error(
1086                id.unwrap_or(Value::Null),
1087                INVALID_REQUEST,
1088                "invalid request",
1089            ));
1090        };
1091        if o.get("jsonrpc").is_some_and(|j| j != "2.0") {
1092            return Some(rpc_error(
1093                id.unwrap_or(Value::Null),
1094                INVALID_REQUEST,
1095                "jsonrpc must be \"2.0\"",
1096            ));
1097        }
1098        // Notifications (initialized, cancelled, ...) need nothing from a
1099        // stateless server.
1100        let id = id?;
1101        if !(id.is_string() || id.is_number()) {
1102            return Some(rpc_error(
1103                Value::Null,
1104                INVALID_REQUEST,
1105                "id must be a string or a number",
1106            ));
1107        }
1108        let params = o.remove("params").unwrap_or(Value::Null);
1109        Some(
1110            match self.dispatch(&method, params, caller, scope, origin) {
1111                Ok(result) => json!({"jsonrpc": "2.0", "id": id, "result": result}),
1112                Err((code, msg)) => rpc_error(id, code, msg),
1113            },
1114        )
1115    }
1116
1117    fn dispatch(
1118        &self,
1119        method: &str,
1120        params: Value,
1121        caller: &Caller,
1122        scope: Option<&crate::org::OrgId>,
1123        origin: &crate::audit::Origin,
1124    ) -> RpcResult {
1125        let bad = |m: &str| Err((INVALID_PARAMS, m.to_string()));
1126        if !(params.is_null() || params.is_object()) {
1127            return bad("params must be an object");
1128        }
1129        match method {
1130            "initialize" => Ok(self.initialize(&params)),
1131            "ping" => Ok(json!({})),
1132            "server/discover" => Ok(self.discover()),
1133            "tools/list" => {
1134                // We never paginate, so any cursor just means "everything".
1135                if params.get("cursor").is_some_and(|c| !c.is_string()) {
1136                    return bad("cursor must be a string");
1137                }
1138                Ok(json!({"tools": self.listed_tools(caller, scope)}))
1139            }
1140            "tools/call" => {
1141                let Some(name) = params.get("name").and_then(Value::as_str) else {
1142                    return bad("tools/call needs a tool name");
1143                };
1144                let args = match params.get("arguments") {
1145                    None | Some(Value::Null) => json!({}),
1146                    Some(a @ Value::Object(_)) => a.clone(),
1147                    Some(_) => return bad("arguments must be an object"),
1148                };
1149                // A tool hidden by policy does not exist for this listener.
1150                match self.call_audited(name, args, caller, scope, origin) {
1151                    Some(r) => Ok(tool_result(r)),
1152                    None => Err((INVALID_PARAMS, format!("unknown tool: {name}"))),
1153                }
1154            }
1155            _ => skills::dispatch(&self.registry.skills, method, &params)
1156                .unwrap_or_else(|| Err((METHOD_NOT_FOUND, format!("method not found: {method}")))),
1157        }
1158    }
1159}
1160
1161enum Admit {
1162    Unknown,
1163    Refused(Error),
1164}
1165
1166/// The kinds of websocket session, and what the audit log calls them.
1167#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1168enum Session {
1169    Terminal,
1170    Ssh,
1171}
1172
1173impl Session {
1174    fn opened(self) -> &'static str {
1175        match self {
1176            Session::Terminal => "terminal.open",
1177            Session::Ssh => "ssh.open",
1178        }
1179    }
1180    fn closed(self) -> &'static str {
1181        match self {
1182            Session::Terminal => "terminal.close",
1183            Session::Ssh => "ssh.close",
1184        }
1185    }
1186    fn noun(self) -> &'static str {
1187        match self {
1188            Session::Terminal => "a terminal",
1189            Session::Ssh => "an SSH session",
1190        }
1191    }
1192    fn not_offered(self) -> &'static str {
1193        match self {
1194            Session::Terminal => "terminals are not offered on this listener",
1195            Session::Ssh => "SSH is not offered on this listener",
1196        }
1197    }
1198    fn limits(self) -> &'static super::terminal::Limits {
1199        match self {
1200            Session::Terminal => &super::terminal::TERMINALS,
1201            Session::Ssh => &super::ssh::LIMITS,
1202        }
1203    }
1204}
1205
1206/// A session that tells the audit hook when it ends, and how long it ran.
1207/// Never what was typed.
1208struct AuditedPty {
1209    inner: Box<dyn super::terminal::Pty>,
1210    started: Instant,
1211    audit: Audit,
1212    caller: Caller,
1213    args: Value,
1214    origin: crate::audit::Origin,
1215    /// `terminal.close` or `ssh.close`.
1216    closed: &'static str,
1217}
1218
1219impl super::terminal::Pty for AuditedPty {
1220    fn input(&mut self, data: &[u8]) -> crate::Result<()> {
1221        self.inner.input(data)
1222    }
1223    fn resize(&mut self, cols: u16, rows: u16) {
1224        self.inner.resize(cols, rows)
1225    }
1226    fn output(&mut self, wait: std::time::Duration) -> super::terminal::PtyOutput {
1227        self.inner.output(wait)
1228    }
1229    fn close(&mut self) {
1230        self.inner.close()
1231    }
1232    fn target(&self) -> Option<String> {
1233        self.inner.target()
1234    }
1235    fn details(&self) -> Option<serde_json::Map<String, Value>> {
1236        self.inner.details()
1237    }
1238}
1239
1240impl Drop for AuditedPty {
1241    fn drop(&mut self) {
1242        let mut args = self.args.clone();
1243        if let (Some(d), Some(m)) = (self.inner.details(), args.as_object_mut()) {
1244            for (k, v) in d {
1245                m.entry(k).or_insert(v);
1246            }
1247        }
1248        args["duration_s"] = json!(self.started.elapsed().as_secs());
1249        (self.audit)(&Audited {
1250            caller: &self.caller,
1251            action: self.closed,
1252            tool: None,
1253            args: &args,
1254            outcome: Ok(()),
1255            origin: &self.origin,
1256        });
1257    }
1258}
1259
1260fn rest_error(status: u16, code: &str, message: &str) -> Response {
1261    Response::json(status, &json!({"error": code, "message": message}))
1262}
1263
1264/// An isb error as a REST response, with the status its code implies.
1265fn error_response(e: &Error) -> Response {
1266    let mut v = crate::rpc::error_json(e);
1267    let code = v["code"].as_str().unwrap_or("error").to_string();
1268    let status = match code.as_str() {
1269        "invalid" | "parse" | "interpolation" | "bad_request" => 400,
1270        "forbidden" => 403,
1271        "not_found" => 404,
1272        "already_exists" => 409,
1273        "request_timeout" | "operation_timeout" | "exec_timeout" | "not_ready" => 504,
1274        _ => 500,
1275    };
1276    let message = v["message"].take();
1277    let mut body = json!({"error": code, "message": message});
1278    if let Some(d) = v.get("data").cloned() {
1279        body["data"] = d;
1280    }
1281    Response::json(status, &body)
1282}
1283
1284fn query_param(req: &Request, key: &str) -> Option<String> {
1285    req.query.as_deref()?.split('&').find_map(|kv| {
1286        let (k, v) = kv.split_once('=').unwrap_or((kv, ""));
1287        (k == key).then(|| v.to_string())
1288    })
1289}
1290
1291/// Run a tool, logged as one line without the arguments (which can carry
1292/// secrets).
1293fn run(tool: &Tool, args: Value, caller: &Caller) -> crate::Result<Value> {
1294    let started = Instant::now();
1295    let r = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
1296        (tool.handler)(args, caller)
1297    }))
1298    .unwrap_or_else(|_| Err(Error::Protocol(format!("tool {} panicked", tool.name))));
1299    let ms = started.elapsed().as_millis();
1300    match &r {
1301        Ok(_) => eprintln!("isb serve: {caller} called {}: ok in {ms}ms", tool.name),
1302        Err(e) => eprintln!(
1303            "isb serve: {caller} called {}: error in {ms}ms: {e}",
1304            tool.name
1305        ),
1306    }
1307    r
1308}
1309
1310/// A tool's outcome as an MCP tool result.
1311fn tool_result(r: crate::Result<Value>) -> Value {
1312    match r {
1313        Ok(v) => {
1314            let text = serde_json::to_string_pretty(&v).unwrap_or_default();
1315            let structured = if v.is_object() {
1316                v
1317            } else {
1318                json!({"result": v})
1319            };
1320            json!({
1321                "content": [{"type": "text", "text": text}],
1322                "structuredContent": structured,
1323                "isError": false,
1324            })
1325        }
1326        Err(e) => json!({
1327            "content": [{"type": "text", "text": e.to_string()}],
1328            "structuredContent": crate::rpc::error_json(&e),
1329            "isError": true,
1330        }),
1331    }
1332}
1333
1334#[cfg(test)]
1335#[path = "mcp_agent_tests.rs"]
1336mod agent_tests;
1337
1338#[cfg(test)]
1339#[path = "mcp_anonymous_tests.rs"]
1340mod anonymous_tests;
1341
1342#[cfg(test)]
1343#[path = "mcp_rpc_tests.rs"]
1344mod rpc_tests;
1345
1346#[cfg(test)]
1347mod tests {
1348    use super::*;
1349    use crate::server::access::tests as at;
1350
1351    pub(super) fn registry() -> Registry {
1352        let mut r = Registry::new().instructions("use isb");
1353        r.register(
1354            Tool::new(
1355                "echo",
1356                "Echo arguments",
1357                json!({"type": "object"}),
1358                |a, c| Ok(json!({"args": a, "caller": c.to_string(), "trusted": c.is_trusted()})),
1359            )
1360            .title("Echo")
1361            .annotations(json!({"readOnlyHint": true})),
1362        )
1363        .unwrap();
1364        r.register(Tool::new("count", "A scalar", json!({}), |_, _| {
1365            Ok(json!(3))
1366        }))
1367        .unwrap();
1368        r.register(Tool::new("stack_rm", "Fails", json!({}), |_, _| {
1369            Err(Error::NotFound("stack web".into()))
1370        }))
1371        .unwrap();
1372        r.register(Tool::new("boom", "Panics", json!({}), |_, _| panic!("x")))
1373            .unwrap();
1374        r
1375    }
1376
1377    pub(super) fn endpoint(policy: ToolPolicy, access: Option<AccessValidator>) -> Endpoint {
1378        Endpoint {
1379            registry: Arc::new(registry()),
1380            policy,
1381            access: access.map(Arc::new),
1382            healthz: Arc::new(|| (true, json!({"ok": true}))),
1383            routes: None,
1384            public_routes: None,
1385            hooks: Hooks::default(),
1386        }
1387    }
1388
1389    pub(super) fn req(
1390        method: &str,
1391        path: &str,
1392        headers: &[(&str, &str)],
1393        body: &[u8],
1394        peer: Peer,
1395    ) -> Request {
1396        Request {
1397            method: method.into(),
1398            path: path.into(),
1399            query: None,
1400            headers: headers
1401                .iter()
1402                .map(|(k, v)| (k.to_string(), v.to_string()))
1403                .collect(),
1404            body: body.to_vec(),
1405            peer,
1406        }
1407    }
1408
1409    pub(super) fn local() -> Peer {
1410        Peer::Unix { uid: Some(1000) }
1411    }
1412
1413    pub(super) fn post(ep: &Endpoint, body: Value) -> (u16, Value) {
1414        let r = ep.handle(&req(
1415            "POST",
1416            "/mcp",
1417            &[("Content-Type", "application/json")],
1418            &serde_json::to_vec(&body).unwrap(),
1419            local(),
1420        ));
1421        let v = if r.body.is_empty() {
1422            Value::Null
1423        } else {
1424            serde_json::from_slice(&r.body).unwrap()
1425        };
1426        (r.status, v)
1427    }
1428
1429    pub(super) fn rpc(method: &str, params: Value) -> Value {
1430        json!({"jsonrpc": "2.0", "id": 7, "method": method, "params": params})
1431    }
1432
1433    #[test]
1434    fn tools_list_and_call() {
1435        let ep = endpoint(ToolPolicy::default(), None);
1436        let (_, v) = post(&ep, rpc("tools/list", json!({"cursor": "x"})));
1437        let tools = v["result"]["tools"].as_array().unwrap();
1438        assert_eq!(tools.len(), 4);
1439        assert_eq!(tools[0]["name"], "echo");
1440        assert_eq!(tools[0]["title"], "Echo");
1441        assert_eq!(tools[0]["inputSchema"], json!({"type": "object"}));
1442        assert_eq!(tools[0]["annotations"]["readOnlyHint"], true);
1443
1444        // No initialize first: the server is stateless.
1445        let (_, v) = post(
1446            &ep,
1447            rpc("tools/call", json!({"name": "echo", "arguments": {"x": 1}})),
1448        );
1449        let r = &v["result"];
1450        assert_eq!(r["isError"], false);
1451        assert_eq!(r["structuredContent"]["args"], json!({"x": 1}));
1452        assert_eq!(r["structuredContent"]["caller"], "local(uid 1000)");
1453        assert_eq!(r["structuredContent"]["trusted"], true);
1454        let text: Value = serde_json::from_str(r["content"][0]["text"].as_str().unwrap()).unwrap();
1455        assert_eq!(text, r["structuredContent"]);
1456
1457        let (_, v) = post(&ep, rpc("tools/call", json!({"name": "count"})));
1458        assert_eq!(v["result"]["structuredContent"], json!({"result": 3}));
1459
1460        let (_, v) = post(&ep, rpc("tools/call", json!({"name": "stack_rm"})));
1461        let r = &v["result"];
1462        assert_eq!(r["isError"], true);
1463        assert_eq!(r["content"][0]["text"], "stack web not found");
1464        assert_eq!(r["structuredContent"]["code"], "not_found");
1465
1466        let (_, v) = post(&ep, rpc("tools/call", json!({"name": "boom"})));
1467        assert_eq!(v["result"]["isError"], true);
1468    }
1469
1470    #[test]
1471    fn policy_filters_list_and_call() {
1472        let ep = endpoint(
1473            ToolPolicy::from_lists("echo, stack_*, count", "stack_rm"),
1474            None,
1475        );
1476        let (_, v) = post(&ep, rpc("tools/list", Value::Null));
1477        let names: Vec<&str> = v["result"]["tools"]
1478            .as_array()
1479            .unwrap()
1480            .iter()
1481            .map(|t| t["name"].as_str().unwrap())
1482            .collect();
1483        assert_eq!(names, ["echo", "count"]);
1484        let (_, v) = post(&ep, rpc("tools/call", json!({"name": "stack_rm"})));
1485        assert_eq!(v["error"]["code"], INVALID_PARAMS);
1486        let (_, v) = post(&ep, rpc("tools/call", json!({"name": "boom"})));
1487        assert_eq!(v["error"]["code"], INVALID_PARAMS);
1488    }
1489
1490    #[test]
1491    fn globs() {
1492        for (p, n, want) in [
1493            ("stack_*", "stack_up", true),
1494            ("stack_*", "stack_", true),
1495            ("stack_*", "stacks", false),
1496            ("*", "anything", true),
1497            ("*_get", "sandbox_get", true),
1498            ("*_get", "sandbox_gets", false),
1499            ("s?ack", "stack", true),
1500            ("s?ack", "sack", false),
1501            ("a*b*c", "aXbYbZc", true),
1502            ("a*b*c", "aXbYbZ", false),
1503            ("[sv]*", "volume_ls", true),
1504            ("[!sv]*", "volume_ls", false),
1505            ("x[a-c]", "xb", true),
1506            ("x[a-c]", "xd", false),
1507            ("x[]]", "x]", true),
1508            ("[unclosed", "[unclosed", true),
1509            ("exact", "exact", true),
1510            ("exact", "exactly", false),
1511            ("", "", true),
1512            ("", "a", false),
1513        ] {
1514            assert_eq!(glob_match(p, n), want, "{p} vs {n}");
1515        }
1516        // Pathological patterns stay fast.
1517        let hostile = "*a".repeat(50);
1518        assert!(!glob_match(&hostile, &"a".repeat(40)));
1519        assert!(glob_match(&hostile, &"a".repeat(60)));
1520        let p = ToolPolicy::from_lists("", " , ");
1521        assert!(p.allow.is_empty() && p.deny.is_empty() && p.allows("x"));
1522    }
1523
1524    #[test]
1525    fn origins() {
1526        for (o, ok) in [
1527            ("http://localhost:5173", true),
1528            ("https://LOCALHOST", true),
1529            ("http://127.0.0.1", true),
1530            ("http://[::1]:8080", true),
1531            ("http://localhost.evil.com", false),
1532            ("http://127.0.0.1.nip.io", false),
1533            ("https://evil.example", false),
1534            ("null", false),
1535            ("http://localhost:abc", false),
1536            ("http://localhost/path", false),
1537            ("file://localhost", false),
1538        ] {
1539            assert_eq!(origin_is_local(o), ok, "{o}");
1540        }
1541        let ep = endpoint(ToolPolicy::default(), None);
1542        let ping = serde_json::to_vec(&rpc("ping", Value::Null)).unwrap();
1543        let tcp = Peer::Tcp("127.0.0.1:1234".parse().unwrap());
1544        let r = ep.handle(&req(
1545            "POST",
1546            "/mcp",
1547            &[("Origin", "https://evil.example")],
1548            &ping,
1549            tcp.clone(),
1550        ));
1551        assert_eq!(r.status, 403);
1552        let r = ep.handle(&req(
1553            "POST",
1554            "/mcp",
1555            &[("Origin", "http://localhost:3000")],
1556            &ping,
1557            tcp.clone(),
1558        ));
1559        assert_eq!(r.status, 200);
1560        // No Origin at all: not a browser.
1561        let r = ep.handle(&req("POST", "/mcp", &[], &ping, tcp));
1562        assert_eq!(r.status, 200);
1563    }
1564
1565    #[test]
1566    fn access_gate() {
1567        let (v, _) = at::validator();
1568        let ep = endpoint(ToolPolicy::default(), Some(v));
1569        let tcp = Peer::Tcp("127.0.0.1:1234".parse().unwrap());
1570        let body = serde_json::to_vec(&rpc("tools/call", json!({"name": "echo"}))).unwrap();
1571        let r = ep.handle(&req("POST", "/mcp", &[], &body, tcp.clone()));
1572        assert_eq!(r.status, 401);
1573        let r = ep.handle(&req(
1574            "POST",
1575            "/mcp",
1576            &[(ASSERTION_HEADER, "a.b.c")],
1577            &body,
1578            tcp.clone(),
1579        ));
1580        assert_eq!(r.status, 401);
1581        let token = at::sign(&at::header(), &at::claims());
1582        // With Access on, the assertion is the gate and Origin is not checked.
1583        let r = ep.handle(&req(
1584            "POST",
1585            "/mcp",
1586            &[
1587                ("cf-access-jwt-assertion", &token),
1588                ("Origin", "https://claude.ai"),
1589            ],
1590            &body,
1591            tcp.clone(),
1592        ));
1593        assert_eq!(r.status, 200);
1594        let v: Value = serde_json::from_slice(&r.body).unwrap();
1595        assert_eq!(
1596            v["result"]["structuredContent"]["caller"],
1597            "alice@example.com"
1598        );
1599        assert_eq!(v["result"]["structuredContent"]["trusted"], false);
1600        // Health never needs the assertion.
1601        let r = ep.handle(&req("GET", "/healthz", &[], b"", tcp));
1602        assert_eq!(r.status, 200);
1603    }
1604
1605    #[test]
1606    fn routes() {
1607        let ep = endpoint(ToolPolicy::default(), None);
1608        assert_eq!(
1609            ep.handle(&req("GET", "/mcp", &[], b"", local())).status,
1610            405
1611        );
1612        assert_eq!(
1613            ep.handle(&req("DELETE", "/mcp", &[], b"", local())).status,
1614            405
1615        );
1616        assert_eq!(ep.handle(&req("GET", "/", &[], b"", local())).status, 404);
1617        assert_eq!(
1618            ep.handle(&req("POST", "/healthz", &[], b"", local()))
1619                .status,
1620            405
1621        );
1622        let sick = Endpoint {
1623            healthz: Arc::new(|| (false, json!({"ok": false}))),
1624            ..endpoint(ToolPolicy::default(), None)
1625        };
1626        let r = sick.handle(&req("GET", "/healthz", &[], b"", local()));
1627        assert_eq!(r.status, 503);
1628        assert_eq!(r.get_header("content-type"), Some("application/json"));
1629    }
1630
1631    #[test]
1632    fn extra_routes_sit_behind_access() {
1633        let routes: super::super::Routes =
1634            Arc::new(|r: &Request| (r.path == "/api/x").then(|| Response::text(200, "extra")));
1635        let tcp = Peer::Tcp("127.0.0.1:1234".parse().unwrap());
1636        // No Access: the routes answer their own paths; the rest is a 404.
1637        let open = Endpoint {
1638            routes: Some(routes.clone()),
1639            public_routes: None,
1640            ..endpoint(ToolPolicy::default(), None)
1641        };
1642        let get = |ep: &Endpoint, path: &str, h: &[(&str, &str)]| {
1643            ep.handle(&req("GET", path, h, b"", tcp.clone())).status
1644        };
1645        assert_eq!(get(&open, "/api/x", &[]), 200);
1646        assert_eq!(get(&open, "/api/y", &[]), 404);
1647        assert_eq!(get(&open, "/healthz", &[]), 200);
1648        // With Access, a route needs the assertion like /mcp does.
1649        let (v, _) = at::validator();
1650        let gated = Endpoint {
1651            routes: Some(routes),
1652            public_routes: None,
1653            ..endpoint(ToolPolicy::default(), Some(v))
1654        };
1655        assert_eq!(get(&gated, "/api/x", &[]), 401);
1656        assert_eq!(get(&gated, "/api/x", &[(ASSERTION_HEADER, "a.b.c")]), 401);
1657        let token = at::sign(&at::header(), &at::claims());
1658        assert_eq!(get(&gated, "/api/x", &[(ASSERTION_HEADER, &token)]), 200);
1659        assert_eq!(get(&gated, "/healthz", &[]), 200);
1660    }
1661
1662    #[test]
1663    fn registry_rejects_bad_and_duplicate_names() {
1664        let mut r = registry();
1665        let t = |n: &str| Tool::new(n, "", json!({}), |_, _| Ok(Value::Null));
1666        assert!(r.register(t("echo")).is_err());
1667        assert!(r.register(t("has space")).is_err());
1668        assert!(r.register(t("")).is_err());
1669        assert!(r.register(t(&"x".repeat(129))).is_err());
1670        assert!(r.register(t("ok.name-2")).is_ok());
1671    }
1672
1673    /// An endpoint whose authorizer pins `org` for scoped calls and refuses
1674    /// any org but "alpha", and whose events hook streams two lines.
1675    pub(super) fn hooked() -> Endpoint {
1676        let mut ep = endpoint(ToolPolicy::default(), None);
1677        ep.hooks = Hooks {
1678            authn: None,
1679            authorize: Some(Arc::new(|_c, _t, mut args, scope| {
1680                if let Some(o) = scope {
1681                    args["org"] = json!(o.as_str());
1682                }
1683                match args.get("org").and_then(Value::as_str) {
1684                    Some("alpha") | None => Ok(args),
1685                    Some(o) => Err(Error::Forbidden(format!("no access to org {o}"))),
1686                }
1687            })),
1688            events: Some(Arc::new(|_c, since| {
1689                Ok(Box::new(move |w: &mut dyn std::io::Write| {
1690                    write!(w, "id: {}\ndata: {{}}\n\n", since + 1)
1691                }))
1692            })),
1693            terminal: Some(Arc::new(|_c, _org, _t| {
1694                Err(Error::NotFound("no such app".into()))
1695            })),
1696            ssh: Some(Arc::new(|_c, _org, _s| {
1697                Err(Error::NotFound("no such instance".into()))
1698            })),
1699            audit: None,
1700            listed: None,
1701            refuse_anonymous: false,
1702        };
1703        ep
1704    }
1705
1706    /// An endpoint whose authn grants superadmin to `source` for any request
1707    /// from a tailnet peer (as the daemon's gate would after its own Host
1708    /// and whois checks), and to a bearer `isb_sa_ok`.
1709    fn superadmin_endpoint(
1710        source: crate::auth::SuperadminSource,
1711        access: Option<AccessValidator>,
1712    ) -> Endpoint {
1713        let mut ep = hooked();
1714        let mut r = registry();
1715        r.register(Tool::new("sandbox_exec", "Exec", json!({}), |_, _| {
1716            Ok(Value::Null)
1717        }))
1718        .unwrap();
1719        ep.registry = Arc::new(r);
1720        ep.access = access.map(Arc::new);
1721        let ambient = Arc::new(crate::auth::Superadmin::synthetic(source));
1722        let token = Arc::new(crate::auth::Superadmin::synthetic(
1723            crate::auth::SuperadminSource::Token {
1724                id: 1,
1725                name: "ci".into(),
1726            },
1727        ));
1728        ep.hooks.authn = Some(Arc::new(move |req: &Request, id: Option<&Identity>| {
1729            if req.header("authorization") == Some("Bearer isb_sa_ok") {
1730                return Authenticated::Superadmin(token.clone());
1731            }
1732            let tailnet =
1733                matches!(&req.peer, Peer::Tcp(a) if crate::server::tailnet::is_tailnet_ip(a.ip()));
1734            if tailnet || id.is_some() {
1735                return Authenticated::Superadmin(ambient.clone());
1736            }
1737            Authenticated::None
1738        }));
1739        ep
1740    }
1741
1742    fn tailnet_peer() -> Peer {
1743        Peer::Tcp("100.64.0.7:5000".parse().unwrap())
1744    }
1745
1746    fn ambient_cases(ep: &Endpoint, peer: Peer, extra: &[(&str, &str)]) {
1747        let body = serde_json::to_vec(&rpc("tools/call", json!({"name": "echo"}))).unwrap();
1748        let call = |path: &str, h: &[(&str, &str)], body: &[u8]| {
1749            let mut all: Vec<(&str, &str)> = extra.to_vec();
1750            all.extend_from_slice(h);
1751            ep.handle(&req("POST", path, &all, body, peer.clone()))
1752        };
1753        let host = ("Host", "100.86.22.100:18995");
1754        let json_ct = ("Content-Type", "application/json");
1755        // /mcp: JSON and no foreign Origin is a superadmin call.
1756        let r = call("/mcp", &[host, json_ct], &body);
1757        assert_eq!(r.status, 200, "{}", String::from_utf8_lossy(&r.body));
1758        let v: Value = serde_json::from_slice(&r.body).unwrap();
1759        assert_eq!(v["result"]["structuredContent"]["trusted"], true);
1760        // A form post (no JSON content type) from a page is refused.
1761        for ct in [
1762            None,
1763            Some("text/plain"),
1764            Some("application/x-www-form-urlencoded"),
1765        ] {
1766            let mut h = vec![host];
1767            if let Some(c) = ct {
1768                h.push(("Content-Type", c));
1769            }
1770            assert_eq!(call("/mcp", &h, &body).status, 403, "{ct:?}");
1771            assert_eq!(call("/orgs/alpha/mcp", &h, &body).status, 403, "{ct:?}");
1772        }
1773        // A foreign Origin is refused; the server's own passes.
1774        assert_eq!(
1775            call(
1776                "/mcp",
1777                &[host, json_ct, ("Origin", "https://evil.example")],
1778                &body
1779            )
1780            .status,
1781            403
1782        );
1783        assert_eq!(
1784            call(
1785                "/mcp",
1786                &[host, json_ct, ("Origin", "http://100.86.22.100:9999")],
1787                &body
1788            )
1789            .status,
1790            403
1791        );
1792        assert_eq!(
1793            call("/mcp", &[host, json_ct, ("Origin", "null")], &body).status,
1794            403
1795        );
1796        assert_eq!(
1797            call(
1798                "/mcp",
1799                &[host, json_ct, ("Origin", "http://100.86.22.100:18995")],
1800                &body
1801            )
1802            .status,
1803            200
1804        );
1805        // REST writes need X-Isb-Csrf, as sessions do.
1806        assert_eq!(
1807            call("/api/v1/tools/echo", &[host, json_ct], b"{}").status,
1808            403
1809        );
1810        let r = call(
1811            "/api/v1/tools/echo",
1812            &[host, json_ct, ("X-Isb-Csrf", "1")],
1813            b"{}",
1814        );
1815        assert_eq!(r.status, 200);
1816        let v: Value = serde_json::from_slice(&r.body).unwrap();
1817        assert_eq!(v["result"]["trusted"], true);
1818        assert_eq!(
1819            call(
1820                "/api/v1/tools/echo",
1821                &[
1822                    host,
1823                    ("X-Isb-Csrf", "1"),
1824                    ("Origin", "https://evil.example")
1825                ],
1826                b"{}"
1827            )
1828            .status,
1829            403
1830        );
1831        // The terminal: an upgrade needs an Origin naming the Host.
1832        let ws = |origin: Option<&str>| {
1833            let mut h: Vec<(&str, &str)> = extra.to_vec();
1834            h.extend([
1835                host,
1836                ("Upgrade", "websocket"),
1837                ("Connection", "Upgrade"),
1838                ("Sec-WebSocket-Version", "13"),
1839                ("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ=="),
1840            ]);
1841            if let Some(o) = origin {
1842                h.push(("Origin", o));
1843            }
1844            let mut q = req("GET", "/orgs/alpha/api/v1/terminal", &h, b"", peer.clone());
1845            q.query = Some("app=web".into());
1846            ep.handle(&q).status
1847        };
1848        assert_eq!(ws(Some("http://100.86.22.100:18995")), 101);
1849        assert_eq!(ws(Some("https://evil.example")), 403);
1850        assert_eq!(ws(None), 403);
1851    }
1852
1853    #[test]
1854    fn tailnet_superadmins_get_csrf_origin_and_content_type_checks() {
1855        let ep = superadmin_endpoint(
1856            crate::auth::SuperadminSource::Tailnet {
1857                login: "me@example.com".into(),
1858                node: "laptop.t.ts.net".into(),
1859                tags: vec![],
1860            },
1861            None,
1862        );
1863        ambient_cases(&ep, tailnet_peer(), &[]);
1864        // A superadmin token is not ambient: no CSRF header or JSON needed.
1865        let r = ep.handle(&req(
1866            "POST",
1867            "/api/v1/tools/echo",
1868            &[("Authorization", "Bearer isb_sa_ok")],
1869            b"{}",
1870            Peer::Tcp("127.0.0.1:1".parse().unwrap()),
1871        ));
1872        assert_eq!(r.status, 200);
1873        let v: Value = serde_json::from_slice(&r.body).unwrap();
1874        assert_eq!(v["result"]["caller"], "superadmin token:ci");
1875    }
1876
1877    #[test]
1878    fn access_superadmins_get_the_same_checks() {
1879        let (v, _) = at::validator();
1880        let ep = superadmin_endpoint(
1881            crate::auth::SuperadminSource::Access {
1882                name: "alice@example.com".into(),
1883                service_token: false,
1884            },
1885            Some(v),
1886        );
1887        let token = at::sign(&at::header(), &at::claims());
1888        ambient_cases(
1889            &ep,
1890            Peer::Tcp("127.0.0.1:4000".parse().unwrap()),
1891            &[("Cf-Access-Jwt-Assertion", token.as_str())],
1892        );
1893    }
1894
1895    #[test]
1896    fn terminal_upgrades_only_when_admitted() {
1897        let mut ep = hooked();
1898        let mut r = registry();
1899        r.register(Tool::new("sandbox_exec", "Exec", json!({}), |_, _| {
1900            Ok(Value::Null)
1901        }))
1902        .unwrap();
1903        ep.registry = Arc::new(r);
1904        let ws = [
1905            // Unauthenticated test callers pass only from a local page.
1906            ("Host", "localhost:8092"),
1907            ("Origin", "http://localhost:8092"),
1908            ("Upgrade", "websocket"),
1909            ("Connection", "Upgrade"),
1910            ("Sec-WebSocket-Version", "13"),
1911            ("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ=="),
1912        ];
1913        let get = |ep: &Endpoint, path: &str, query: &str, h: &[(&str, &str)]| {
1914            let mut q = req("GET", path, h, b"", local());
1915            q.query = Some(query.into());
1916            ep.handle(&q)
1917        };
1918        let ok = get(&ep, "/orgs/alpha/api/v1/terminal", "app=web", &ws);
1919        assert_eq!(ok.status, 101);
1920        assert_eq!(
1921            ok.get_header("sec-websocket-accept"),
1922            Some("s3pPLMBiTxaQ9kYGzzhZRbK+xOo=")
1923        );
1924        assert!(ok.upgrade.is_some());
1925        // Another org is refused before any upgrade.
1926        assert_eq!(
1927            get(&ep, "/orgs/beta/api/v1/terminal", "app=web", &ws).status,
1928            403
1929        );
1930        // A cross-site page riding the session cookie is refused.
1931        let mut evil = ws.to_vec();
1932        evil[1] = ("Origin", "http://localhost:9999");
1933        assert_eq!(
1934            get(&ep, "/orgs/alpha/api/v1/terminal", "app=web", &evil).status,
1935            403
1936        );
1937        // Not a websocket, or no app: 400; not GET: 405.
1938        assert_eq!(
1939            get(&ep, "/orgs/alpha/api/v1/terminal", "app=web", &ws[..2]).status,
1940            400
1941        );
1942        assert_eq!(get(&ep, "/orgs/alpha/api/v1/terminal", "", &ws).status, 400);
1943        let r = ep.handle(&req(
1944            "POST",
1945            "/orgs/alpha/api/v1/terminal",
1946            &ws,
1947            b"",
1948            local(),
1949        ));
1950        assert_eq!(r.status, 405);
1951        // An instance of the org instead of an app.
1952        assert_eq!(
1953            get(&ep, "/orgs/alpha/api/v1/terminal", "instance=box", &ws).status,
1954            101
1955        );
1956        // --deny-tools sandbox_exec turns terminals off.
1957        ep.policy = ToolPolicy::from_lists("", "sandbox_exec");
1958        assert_eq!(
1959            get(&ep, "/orgs/alpha/api/v1/terminal", "app=web", &ws).status,
1960            404
1961        );
1962    }
1963
1964    #[test]
1965    #[expect(
1966        clippy::too_many_lines,
1967        reason = "predates the lint ratchet; split it when next changed"
1968    )]
1969    fn ssh_upgrades_only_when_admitted() {
1970        let mut ep = hooked();
1971        let mut r = registry();
1972        r.register(Tool::new("sandbox_exec", "Exec", json!({}), |_, _| {
1973            Ok(Value::Null)
1974        }))
1975        .unwrap();
1976        ep.registry = Arc::new(r);
1977        let upgrade = [
1978            ("Upgrade", "websocket"),
1979            ("Connection", "Upgrade"),
1980            ("Sec-WebSocket-Version", "13"),
1981            ("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ=="),
1982        ];
1983        let get = |ep: &Endpoint, path: &str, query: &str, h: &[(&str, &str)], peer: Peer| {
1984            let mut hs = upgrade.to_vec();
1985            hs.extend_from_slice(h);
1986            let mut q = req("GET", path, &hs, b"", peer);
1987            q.query = Some(query.into());
1988            ep.handle(&q)
1989        };
1990        let page = [
1991            ("Host", "localhost:8092"),
1992            ("Origin", "http://localhost:8092"),
1993        ];
1994        let ok = get(
1995            &ep,
1996            "/orgs/alpha/api/v1/ssh",
1997            "instance=box",
1998            &page,
1999            local(),
2000        );
2001        assert_eq!(ok.status, 101);
2002        assert!(ok.upgrade.is_some());
2003        // The unix socket needs no Origin.
2004        let unix = Peer::Unix { uid: None };
2005        assert_eq!(
2006            get(
2007                &ep,
2008                "/orgs/alpha/api/v1/ssh",
2009                "instance=box",
2010                &[],
2011                unix.clone()
2012            )
2013            .status,
2014            101
2015        );
2016        // Another org: refused before any upgrade.
2017        assert_eq!(
2018            get(
2019                &ep,
2020                "/orgs/beta/api/v1/ssh",
2021                "instance=box",
2022                &[],
2023                unix.clone()
2024            )
2025            .status,
2026            403
2027        );
2028        // A cross-site page: refused.
2029        let evil = [
2030            ("Host", "localhost:8092"),
2031            ("Origin", "http://localhost:9999"),
2032        ];
2033        assert_eq!(
2034            get(
2035                &ep,
2036                "/orgs/alpha/api/v1/ssh",
2037                "instance=box",
2038                &evil,
2039                local()
2040            )
2041            .status,
2042            403
2043        );
2044        // No instance, a bad one, or not an upgrade: 400.
2045        assert_eq!(
2046            get(&ep, "/orgs/alpha/api/v1/ssh", "", &[], unix.clone()).status,
2047            400
2048        );
2049        assert_eq!(
2050            get(
2051                &ep,
2052                "/orgs/alpha/api/v1/ssh",
2053                "instance=A/b",
2054                &[],
2055                unix.clone()
2056            )
2057            .status,
2058            400
2059        );
2060        let mut q = req("GET", "/orgs/alpha/api/v1/ssh", &[], b"", unix.clone());
2061        q.query = Some("instance=box".into());
2062        assert_eq!(ep.handle(&q).status, 400);
2063        // --deny-tools sandbox_exec turns SSH off with terminals.
2064        ep.policy = ToolPolicy::from_lists("", "sandbox_exec");
2065        assert_eq!(
2066            get(&ep, "/orgs/alpha/api/v1/ssh", "instance=box", &[], unix).status,
2067            404
2068        );
2069        // No hook: no such endpoint.
2070        let mut ep = hooked();
2071        ep.hooks.ssh = None;
2072        assert_eq!(
2073            get(
2074                &ep,
2075                "/orgs/alpha/api/v1/ssh",
2076                "instance=box",
2077                &page,
2078                local()
2079            )
2080            .status,
2081            404
2082        );
2083    }
2084
2085    #[test]
2086    fn rest_calls_and_errors() {
2087        let ep = hooked();
2088        let call = |path: &str, body: &[u8]| {
2089            let r = ep.handle(&req("POST", path, &[], body, local()));
2090            (
2091                r.status,
2092                serde_json::from_slice::<Value>(&r.body).unwrap_or(Value::Null),
2093            )
2094        };
2095        let (st, v) = call("/api/v1/tools/echo", br#"{"org":"alpha","x":1}"#);
2096        assert_eq!(st, 200);
2097        assert_eq!(v["result"]["args"]["x"], 1);
2098        let (st, v) = call("/api/v1/tools/echo", br#"{"org":"beta"}"#);
2099        assert_eq!((st, v["error"].as_str()), (403, Some("forbidden")));
2100        let (st, _) = call("/api/v1/tools/missing", b"{}");
2101        assert_eq!(st, 404);
2102        let (st, v) = call("/api/v1/tools/stack_rm", b"{}");
2103        assert_eq!((st, v["error"].as_str()), (404, Some("not_found")));
2104        let (st, _) = call("/api/v1/tools/echo", b"[1]");
2105        assert_eq!(st, 400);
2106        // A scoped endpoint pins the org into the arguments.
2107        let (st, v) = call("/orgs/alpha/api/v1/tools/echo", b"{}");
2108        assert_eq!(
2109            (st, v["result"]["args"]["org"].as_str()),
2110            (200, Some("alpha"))
2111        );
2112        let (st, _) = call("/orgs/beta/api/v1/tools/echo", b"{}");
2113        assert_eq!(st, 403);
2114        let (st, _) = call("/orgs/Not_An_Org/api/v1/tools/echo", b"{}");
2115        assert_eq!(st, 404);
2116        let r = ep.handle(&req("GET", "/api/v1/openapi.json", &[], b"", local()));
2117        let doc: Value = serde_json::from_slice(&r.body).unwrap();
2118        assert!(doc["paths"]["/api/v1/tools/echo"]["post"].is_object());
2119    }
2120
2121    #[test]
2122    fn scoped_mcp_and_refusals_are_tool_errors() {
2123        let ep = hooked();
2124        let body = |org: Option<&str>| {
2125            let mut args = json!({});
2126            if let Some(o) = org {
2127                args["org"] = json!(o);
2128            }
2129            serde_json::to_vec(&rpc(
2130                "tools/call",
2131                json!({"name": "echo", "arguments": args}),
2132            ))
2133            .unwrap()
2134        };
2135        let r = ep.handle(&req("POST", "/orgs/alpha/mcp", &[], &body(None), local()));
2136        let v: Value = serde_json::from_slice(&r.body).unwrap();
2137        assert_eq!(v["result"]["structuredContent"]["args"]["org"], "alpha");
2138        let r = ep.handle(&req("POST", "/mcp", &[], &body(Some("beta")), local()));
2139        let v: Value = serde_json::from_slice(&r.body).unwrap();
2140        assert_eq!(v["result"]["isError"], true);
2141        assert_eq!(v["result"]["structuredContent"]["code"], "forbidden");
2142    }
2143
2144    #[test]
2145    fn events_stream() {
2146        let ep = hooked();
2147        let r = ep.handle(&req(
2148            "GET",
2149            "/api/v1/events",
2150            &[("Last-Event-ID", "41")],
2151            b"",
2152            local(),
2153        ));
2154        assert_eq!(r.get_header("content-type"), Some("text/event-stream"));
2155        let mut out = Vec::new();
2156        crate::server::http::write_response(&mut out, &r).unwrap();
2157        let text = String::from_utf8(out).unwrap();
2158        assert!(!text.contains("Content-Length"), "{text}");
2159        assert!(text.ends_with("id: 42\ndata: {}\n\n"), "{text}");
2160    }
2161}