From 03888a2383c752b260a32395226b6fcea1f55790 Mon Sep 17 00:00:00 2001 From: Raymond Khalife Date: Thu, 12 Nov 2026 18:02:43 +0000 Subject: [PATCH] Serve the deploy tools on the hosted server, with the bundle uploaded on a ticket The hosted server could not deploy: the tool lived in the local binary alone. Now, behind the gateway crate's `deploy` feature, the MCP endpoint serves imcp2-deploy's tools beside the core surface, through the one handler both servers mount (`WithDeploy`, moved out of imcp2-local). Three things the hosted shape needs that the local one did not: How the bundle arrives. The agent's machine cannot hand this server a file, and a tool call carries only what the model writes, so `icp_deploy_upload` mints a one-time ticket inside the authenticated call and returns the curl command that PUTs the archive to `{mcp_path}/deploy/upload`. The ticket is 256 random bits, single use, good for five minutes, bound to the session that minted it (the upload it produces is deployable by that session alone), and sent in a header so it stays out of proxy logs. The route is outside the bearer gate (the ticket is its authentication) and streams the body to disk under a cap, a declared length over it refused unread. `icp_deploy_bundle` takes the upload id as `bundle_upload`; the archive goes with the deploy. How the deploy call is answered. The endpoint answers a POST with plain JSON, the one message a handler returns, and a deploy reports each phase before its result. A `tools/call` of `icp_deploy_bundle` is therefore dispatched to a second rmcp service configured for Server-Sent-Events responses, still stateless; every other request keeps the JSON service. A client that goes away. The event stream's body is wrapped so that its drop cancels a token the deploy watches (`ClientGone`), and the deploy stops at its next step, keeping its directory as the record. Also: one deploy per session plus a server-wide cap (`concurrent`, default 2 hosted, 1 local), refused past it rather than queued; deploy directories and uploads under `{state_dir}/deploys`; the knobs `MCP_DEPLOY_MAX_CONCURRENT` and `MCP_DEPLOY_MAX_UPLOAD_BYTES`; the native build turns the feature on and CI tests it. The feature keeps `cargo publish` of imcp2 from working until the icp CLI's crates are published. Co-Authored-By: Claude Fable 5.1 --- .github/workflows/ci.yml | 6 + Cargo.lock | 5 + Cargo.toml | 17 +- README.md | 59 ++- crates/imcp2-deploy/Cargo.toml | 5 + crates/imcp2-deploy/src/deploy.rs | 10 + crates/imcp2-deploy/src/lib.rs | 345 ++++++++++++-- crates/imcp2-deploy/src/server.rs | 167 +++++++ crates/imcp2-deploy/src/uploads.rs | 348 ++++++++++++++ .../static/skills/deploy-with-mcp.md | 52 ++- crates/imcp2-local/README.md | 10 +- crates/imcp2-local/src/server.rs | 99 ++-- deploy/native/README.md | 10 + deploy/native/build.sh | 2 +- src/auth.rs | 20 + src/deploy.rs | 437 ++++++++++++++++++ src/lib.rs | 57 ++- src/main.rs | 29 ++ 18 files changed, 1538 insertions(+), 140 deletions(-) create mode 100644 crates/imcp2-deploy/src/server.rs create mode 100644 crates/imcp2-deploy/src/uploads.rs create mode 100644 src/deploy.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ad55049..9c040b8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -43,6 +43,12 @@ jobs: - name: Test run: cargo test --locked --workspace --all-targets + # The deploy tools on the hosted server are a feature of `imcp2` (off in + # the default build, on in the native deployment's), so they get their + # own run. + - name: Test the hosted deploy feature + run: cargo test --locked -p imcp2 --features deploy --all-targets + # Formatting, enforced rather than merely requested: CONTRIBUTING.md asks for # `cargo fmt --all` and the coding guidelines ask for a rustfmt-clean tree, # and this is what keeps that true. Without it the tree drifts out of shape diff --git a/Cargo.lock b/Cargo.lock index c3f09de..6693817 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2427,17 +2427,20 @@ dependencies = [ "axum", "base64 0.22.1", "candid", + "futures-util", "getrandom 0.3.4", "hex", "http-body-util", "ic-agent", "imcp2-core", + "imcp2-deploy", "pocket-ic", "prometheus", "rmcp", "serde", "serde_json", "sha2 0.11.0", + "tempfile", "tokio", "tokio-util", "tower", @@ -2485,7 +2488,9 @@ dependencies = [ "camino", "candid", "flate2", + "getrandom 0.3.4", "hex", + "http", "ic-agent", "icp-canister-interfaces", "icp-events", diff --git a/Cargo.toml b/Cargo.toml index 86e2670..aff69e2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -99,9 +99,17 @@ serde = { workspace = true } serde_json = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } -tokio-util = "0.7" +tokio-util = { version = "0.7", features = ["io"] } uuid = { workspace = true } tower-http = { version = "0.7", features = ["cors"] } +# The deploy tools on the hosted server (src/deploy.rs), behind the `deploy` +# feature. imcp2-deploy pins icp-cli crates by git, which crates.io refuses, so +# the feature keeps `cargo publish` from working until the icp CLI's crates are +# published; a release build with the feature is a deployment's choice. +imcp2-deploy = { path = "crates/imcp2-deploy", optional = true } +# The deploy endpoint: streamed bodies (the upload, the watched event stream). +futures-util = "0.3" +http-body-util = "0.1" # Metrics exposition at /metrics. Pinned to the version dfinity/ic uses: these # series are destined for the same Prometheus / Victoria Metrics clusters that # scrape the IC, so matching the org's client keeps one exposition dialect in the @@ -125,12 +133,17 @@ pocket-ic = { version = "15", optional = true } # src/e2e_handshake.rs). Off by default: the standard `cargo test` neither # builds `pocket-ic` nor touches the network. e2e = ["dep:pocket-ic"] +# Serves the deploy tools (icp_deploy_bundle, icp_deploy_upload, +# icp_cycles_balance, icp_deploy_guide) on the MCP endpoint, with the bundle +# upload route and the event-stream transport for the deploy call. See +# src/deploy.rs and the note on imcp2-deploy above. +deploy = ["dep:imcp2-deploy"] [dev-dependencies] # Router-level contract tests (tests/routers.rs) drive the composed app with # tower::ServiceExt::oneshot and read bodies via http-body-util. tower = { version = "0.5", features = ["util"] } -http-body-util = "0.1" +tempfile = "3" # `test-util` is what `#[tokio::test(start_paused = true)]` needs: the pending # connect tests in `auth.rs` walk a connect past its TTL and retention on a # paused clock instead of waiting them out. A feature on a dependency the crate diff --git a/README.md b/README.md index dd9013f..5ce5dae 100644 --- a/README.md +++ b/README.md @@ -542,6 +542,60 @@ erroring. The design mirrors the reference IC connector's OQL primer (detect + teach), adding an ergonomic executor suited to this server's structured-output conventions. +## Deploying a project bundle (the `deploy` feature) + +Built with `cargo build --features deploy`, the server also serves the +deploy tools of the `imcp2-deploy` crate on the MCP endpoint: `icp_deploy_bundle` +deploys an [icp CLI](https://github.com/dfinity/icp-cli) project bundle to the +Internet Computer as the user — every phase of `icp deploy`, creating the +canisters that do not exist yet from the user's cycles balance (`icp_cycles_balance` +reports it) and upgrading those named in `canister_ids` — and `icp_deploy_guide` +returns the skill that walks an agent through the loop (also the +`skill://deploy-with-mcp` resource). It signs as the user's default Internet +Identity account at `https://cli.id.ai`, the principal `icp identity link web` +links an icp CLI identity to; `crates/imcp2-local/README.md` says what that +means for sharing canisters and cycles with the CLI. + +What the hosted composition adds to the local one: + +- **The bundle arrives by upload.** The agent's machine cannot hand this server + a file, and a tool call carries only what the model writes, so + `icp_deploy_upload` mints a one-time ticket inside the authenticated call and + returns the `curl` command that `PUT`s the archive to + `{mcp_path}/deploy/upload`. The ticket is 256 random bits, single use, good + for five minutes, bound to the session that minted it (the upload it produces + is deployable by that session alone), and sent in a request header so it + stays out of proxy logs. The route is outside the bearer gate — the ticket is + its authentication — and streams the body to disk under + `$MCP_DEPLOY_MAX_UPLOAD_BYTES`, refusing a declared length over it unread. An + upload is held fifteen minutes for its deploy, and removed with it. +- **The deploy call is answered as an event stream.** The endpoint answers a + POST with plain JSON, the one message a handler returns; a deploy reports + each phase as a progress notification before its result, so a `tools/call` + of `icp_deploy_bundle` is dispatched to a second rmcp service configured for + Server-Sent-Events responses, still stateless. Every other request keeps the + JSON service. The stream carries each phase as it completes, which is also + what keeps a client's per-call timeout from expiring on a minutes-long + deploy (send a progress token). +- **A client that goes away stops its deploy.** The event stream's body is + wrapped so that its drop — the client disconnecting — cancels a token the + deploy watches; the deploy stops at its next step, and its working directory + stays under `{state_dir}/deploys` with the ids it recorded, for an operator + to hand the user. +- **Bounds.** One deploy per session and `$MCP_DEPLOY_MAX_CONCURRENT` across + sessions, refused past the cap rather than queued. Each deploy's directory is + made under `{state_dir}/deploys`, readable by the server's user alone, and + removed when the deploy returns. The archive and what it unpacks are bounded + by the `imcp2-deploy` host (archive, extracted bytes, entries, module sizes), + and its sync plugins run in WASI with read-only access to their own deploy + directory and no network. + +The feature is off in the default build and on in the native deployment's +(`deploy/native/build.sh`). `imcp2-deploy` pins icp-cli crates by git, which +crates.io refuses, so the published `imcp2` crate cannot carry the feature +until the icp CLI's crates are published; `cargo publish` of this crate waits +on that. + ## Connect from an MCP client Add the server to Claude Code (replace the URL with wherever it's hosted): @@ -569,7 +623,10 @@ cargo run # serves http://0.0.0.0:8000 (MCP at /mcp against production II, OAuth under it; / redirects to the landing site) # honours $PORT (default 8000), $PUBLIC_URL (default http://localhost:8000), # $MCP_SERVE_BETA (set it to also serve the beta II instance at /mcp-beta, for staging), -# and $MCP_SERVE_METRICS (set it to serve the Prometheus exposition at /metrics) +# and $MCP_SERVE_METRICS (set it to serve the Prometheus exposition at /metrics). +# Built with `--features deploy` it also serves the deploy tools, with +# $MCP_DEPLOY_MAX_CONCURRENT deploys at once (default 2) and uploads of at most +# $MCP_DEPLOY_MAX_UPLOAD_BYTES (default 64 MiB); see "Deploying a project bundle". ``` The human-facing pages — the landing page and the `/privacy-policy`, diff --git a/crates/imcp2-deploy/Cargo.toml b/crates/imcp2-deploy/Cargo.toml index 18dd54a..21ad7e8 100644 --- a/crates/imcp2-deploy/Cargo.toml +++ b/crates/imcp2-deploy/Cargo.toml @@ -31,6 +31,11 @@ tracing = { workspace = true } base64 = { workspace = true } sha2 = { workspace = true } hex = { workspace = true } +# Upload tickets and ids are 256 random bits each. +getrandom = { workspace = true } +# The HTTP request parts a transport leaves in the request context, where the +# hosted server says the client has gone. +http = { workspace = true } url = { workspace = true } async-trait = "0.1" # `icp-project` spells every path as a UTF-8 path. diff --git a/crates/imcp2-deploy/src/deploy.rs b/crates/imcp2-deploy/src/deploy.rs index e54f048..b3b1eff 100644 --- a/crates/imcp2-deploy/src/deploy.rs +++ b/crates/imcp2-deploy/src/deploy.rs @@ -42,6 +42,10 @@ pub enum BundleSource { Bytes(Vec), /// An absolute path to the archive on this machine. Path(Utf8PathBuf), + /// An archive uploaded ahead of the call, by the id the upload answered + /// with (see [`crate::Uploads`]). The tool turns it into a [`Self::Path`] + /// before the deploy; one reaching the deploy as is, is a request error. + Upload(String), } // By hand so a request never prints megabytes of archive. @@ -50,6 +54,7 @@ impl fmt::Debug for BundleSource { match self { BundleSource::Bytes(bytes) => write!(f, "Bytes({} bytes)", bytes.len()), BundleSource::Path(path) => write!(f, "Path({path})"), + BundleSource::Upload(id) => write!(f, "Upload({id})"), } } } @@ -200,6 +205,11 @@ pub async fn deploy_bundle( let bytes = match request.bundle { BundleSource::Bytes(bytes) => bytes, BundleSource::Path(path) => read_bundle_file(&path, host::MAX_BUNDLE_BYTES).await?, + BundleSource::Upload(id) => { + return Err(RequestError(format!( + "the upload '{id}' was not resolved to a file before the deploy" + ))) + } }; // The working directory holds the bundle's files, the ids the deploy diff --git a/crates/imcp2-deploy/src/lib.rs b/crates/imcp2-deploy/src/lib.rs index 206a990..d2d3d5f 100644 --- a/crates/imcp2-deploy/src/lib.rs +++ b/crates/imcp2-deploy/src/lib.rs @@ -25,17 +25,27 @@ //! controlled by the principal here, and reaching it through the lifecycle //! tools means adding their principal as a controller first. //! -//! One deploy runs at a time on a server: two at once for the same project, +//! One deploy runs at a time per session: two at once for the same project, //! each with the mapping as it was before either, would create two sets of //! canisters. The second call is refused, not queued, since its mapping would -//! be stale by the time it ran. And a deploy's working directory is kept -//! under the directory the embedder names until the outcome is assembled, -//! so a call cut off mid-deploy leaves the ids it created on disk. +//! be stale by the time it ran. A server also caps how many deploys run at +//! once across every session ([`IcDeployTools::concurrent`]; one, unless it +//! says otherwise), and refuses rather than queues past the cap. A deploy's +//! working directory is kept under the directory the embedder names until the +//! outcome is assembled, so a call cut off mid-deploy leaves the ids it +//! created on disk. //! -//! This crate is composed into the local server only. It depends on -//! `icp-project` straight from the icp-cli repository, which crates.io does -//! not allow, so it cannot be part of the published `imcp2-core`/`imcp2` -//! crates (see `Cargo.toml`). +//! A deploy lasts as long as it lasts: the call reports each phase to the +//! client as it completes and is interrupted by the client cancelling the +//! request or, on the hosted server, going away ([`ClientGone`]). Where the +//! agent cannot hand the server a file — the hosted server — the bundle +//! arrives through an upload ticket ([`Uploads`], `icp_deploy_upload`). +//! +//! The crate is composed into the local server, and into the hosted one +//! behind its `deploy` feature ([`WithDeploy`] is the handler both mount). +//! It depends on `icp-project` straight from the icp-cli repository, which +//! crates.io does not allow, so the published `imcp2` crate cannot carry the +//! feature until the icp CLI's crates are published (see `Cargo.toml`). //! //! The bundle is untrusted input, and the rule it is held to is the //! library's own: `icp_project::untrusted`, the one `icp --untrusted` @@ -44,7 +54,10 @@ //! the rule as the CLI does, and adds the bounds on the archive itself and the //! checks on the bundle's shape (see [`host`]). -use std::{collections::BTreeMap, sync::Arc}; +use std::{ + collections::{BTreeMap, HashSet}, + sync::{Arc, Mutex}, +}; use base64::Engine; use camino::Utf8PathBuf; @@ -61,16 +74,36 @@ use rmcp::{ tool, tool_router, ErrorData as McpError, RoleServer, }; use serde::{Deserialize, Serialize}; +use tokio_util::sync::CancellationToken; mod calls; mod deploy; mod host; mod identity; +mod server; +mod uploads; pub use deploy::{ deploy_bundle, BundleSource, DeployOutcome, DeployRequest, RequestError, Status, Watch, }; use identity::RefreshingIdentity; +pub use server::WithDeploy; +pub use uploads::{ + ReceiveError, Received, Uploads, DEFAULT_MAX_UPLOAD_BYTES, TICKET_HEADER, TICKET_TTL, + UPLOAD_TTL, +}; + +/// The name of the upload-ticket tool, listed only by a server with an +/// upload store (see [`IcDeployTools::with_uploads`]). +pub const UPLOAD_TOOL: &str = "icp_deploy_upload"; + +/// A transport's word that the client of a request has gone: a token it +/// cancels when the connection carrying the request's response is dropped, +/// left in the HTTP request's extensions (which rmcp hands the tool as the +/// request parts). A deploy stops when it fires, since nobody would read +/// the answer. +#[derive(Clone, Debug)] +pub struct ClientGone(pub CancellationToken); /// The icp CLI's default funding for a canister it creates (`icp deploy /// --cycles`): 2 trillion cycles. @@ -109,6 +142,11 @@ pub struct DeployBundleArgs { /// `bundle_path`, not both. #[serde(default)] pub bundle_base64: Option, + /// The upload id the curl command from `icp_deploy_upload` printed: the + /// archive as this server received it. Only on a server that lists + /// `icp_deploy_upload`; exactly one of the three bundle arguments. + #[serde(default)] + pub bundle_upload: Option, /// The environment to deploy, by name, among those the bundle's manifest /// declares — `ic` for a bundle built with the default settings. It must /// be on the `ic` network: this tool deploys to the Internet Computer only. @@ -251,12 +289,29 @@ const MAX_BUNDLE_BASE64_CHARS: usize = host::MAX_BUNDLE_BYTES.div_ceil(3) * 4; /// Turn the tool's arguments into a request, rejecting what cannot be deployed /// before anything is read or written. -pub fn request_from_args(args: DeployBundleArgs) -> Result { - let bundle = match (args.bundle_path, args.bundle_base64) { +pub fn request_from_args(mut args: DeployBundleArgs) -> Result { + let given = + [args.bundle_path.is_some(), args.bundle_base64.is_some(), args.bundle_upload.is_some()] + .iter() + .filter(|g| **g) + .count(); + if given > 1 { + return Err("pass one of bundle_path, bundle_base64 or bundle_upload, not several".into()); + } + if let Some(id) = args.bundle_upload.take() { + let id = id.trim(); + if id.is_empty() { + return Err("bundle_upload is empty; pass the upload id the upload printed".into()); + } + return finish_request(args, BundleSource::Upload(id.to_string())); + } + let bundle = match (args.bundle_path.take(), args.bundle_base64.take()) { (Some(_), Some(_)) => { return Err("pass either bundle_path or bundle_base64, not both".into()) } - (None, None) => return Err("pass the bundle as bundle_path or bundle_base64".into()), + (None, None) => { + return Err("pass the bundle as bundle_path, bundle_base64 or bundle_upload".into()) + } (Some(path), None) => { let path = Utf8PathBuf::from(path.trim()); if path.as_str().is_empty() { @@ -293,6 +348,11 @@ pub fn request_from_args(args: DeployBundleArgs) -> Result Result { let environment = args.environment.trim().to_string(); if environment.is_empty() { return Err("environment must name an environment of the bundle, e.g. \"ic\"".into()); @@ -344,15 +404,59 @@ pub struct IcDeployTools { /// The authentication seam (see [`SessionResolver`]): asks the embedding /// binary which already-validated session this call acts as. session: SessionResolver, - /// Held for the length of a deploy; a second deploy finds it taken. - running: Arc>, + /// The deploys running: one slot per deploy the server allows at once, + /// and the sessions holding one. + slots: Slots, /// Where each deploy's working directory is made; see /// [`DeployRequest::workdir`]. deploys_dir: Option, + /// Where uploaded bundles are received, and the URL a `PUT` goes to, + /// on a server the agent cannot hand a file to. + uploads: Option<(Uploads, String)>, +} + +/// How many deploys a server runs at once, and for whom. +#[derive(Clone)] +struct Slots { + /// One permit per deploy the server runs at once. + free: Arc, + /// How many there are in all, for the instructions. + total: usize, + /// The sessions with a deploy running: one each. + sessions: Arc>>, } -/// A running deploy's claim on the server: dropped when the deploy returns. -pub type DeployClaim = tokio::sync::OwnedMutexGuard<()>; +/// A running deploy's claim on the server: a slot, and the session's one +/// deploy. Both are released when it drops, that is, when the deploy returns. +pub struct DeployClaim { + _permit: tokio::sync::OwnedSemaphorePermit, + session: Option, + sessions: Arc>>, +} + +impl DeployClaim { + /// Make the claim the session's: refused if the session already has a + /// deploy running. + fn for_session(&mut self, sid: &str) -> Result<(), String> { + let mut sessions = self.sessions.lock().unwrap_or_else(|e| e.into_inner()); + if !sessions.insert(sid.to_string()) { + return Err("a deploy is already running for this session; wait for it to finish, \ + write the mapping it returns into the project, and deploy again from \ + that" + .into()); + } + self.session = Some(sid.to_string()); + Ok(()) + } +} + +impl Drop for DeployClaim { + fn drop(&mut self) { + if let Some(sid) = self.session.take() { + self.sessions.lock().unwrap_or_else(|e| e.into_inner()).remove(&sid); + } + } +} /// The derivation origin of the identity the deploy tools act as: the one the /// icp CLI's `icp identity link web` (with no `--app`) signs in at. Internet @@ -364,7 +468,46 @@ pub const CLI_ORIGIN: &str = "https://cli.id.ai"; #[tool_router(vis = "pub")] impl IcDeployTools { pub fn new(identities: Identities, session: SessionResolver) -> Self { - Self { identities, session, running: Arc::default(), deploys_dir: None } + Self { + identities, + session, + slots: Slots { + free: Arc::new(tokio::sync::Semaphore::new(1)), + total: 1, + sessions: Arc::default(), + }, + deploys_dir: None, + uploads: None, + } + } + + /// Run up to `n` deploys at once across sessions (one per session either + /// way); the default is one. Past the cap a deploy is refused, not + /// queued. `n` is at least one. + pub fn concurrent(mut self, n: usize) -> Self { + let n = n.max(1); + self.slots.free = Arc::new(tokio::sync::Semaphore::new(n)); + self.slots.total = n; + self + } + + /// How many deploys this server runs at once. + pub fn concurrency(&self) -> usize { + self.slots.total + } + + /// Serve `icp_deploy_upload` on `uploads`, with `upload_url` the absolute + /// URL the agent `PUT`s an archive to (the server mounts the route; see + /// [`Uploads`]). Without this, the tool is not listed and `bundle_upload` + /// is refused. + pub fn with_uploads(mut self, uploads: Uploads, upload_url: String) -> Self { + self.uploads = Some((uploads, upload_url)); + self + } + + /// Whether this server receives uploaded bundles. + pub fn has_uploads(&self) -> bool { + self.uploads.is_some() } /// Make each deploy's working directory under `dir` (made on first use) @@ -377,9 +520,10 @@ impl IcDeployTools { Ok(self) } - /// Claim the server for a deploy, or `None` while another holds it. + /// Claim a slot for a deploy, or `None` while every slot is held. pub fn claim(&self) -> Option { - self.running.clone().try_lock_owned().ok() + let permit = self.slots.free.clone().try_acquire_owned().ok()?; + Some(DeployClaim { _permit: permit, session: None, sessions: self.slots.sessions.clone() }) } /// The session's identity at [`CLI_ORIGIN`]: its default account there, @@ -399,27 +543,63 @@ impl IcDeployTools { Parameters(args): Parameters, ctx: RequestContext, ) -> Result { - // One deploy at a time, and the refusal reads nothing: it comes even - // before who is asking. - let Some(_claim) = self.claim() else { - return Ok(err("a deploy is already running on this server; wait for it to \ - finish, write the mapping it returns into the project, and \ - deploy again from that" - .into())); + // So many deploys at a time, and the refusal reads nothing: it comes + // even before who is asking. + let Some(mut claim) = self.claim() else { + return Ok(err(if self.concurrency() == 1 { + "a deploy is already running on this server; wait for it to finish, write \ + the mapping it returns into the project, and deploy again from that" + .into() + } else { + format!( + "this server is running its {} deploys at once already; wait and try again", + self.concurrency() + ) + })); }; // Who is asking comes before what they sent: an inline bundle is - // decoded only for a session allowed to deploy. + // decoded only for a session allowed to deploy, and one deploy per + // session. let Some(sid) = (self.session)(&ctx) else { return Ok(err("deploying a bundle needs an authenticated session".into())); }; if let Err(e) = self.identities.require_write(&sid).await { return Ok(err(e)); } + if let Err(e) = claim.for_session(&sid) { + return Ok(err(e)); + } let mut request = match request_from_args(args) { Ok(r) => r, Err(e) => return Ok(err(e)), }; request.workdir = self.deploys_dir.clone(); + // An uploaded bundle is the session's own, and the file goes with + // the handle once the deploy has returned. + let _upload = match request.bundle { + BundleSource::Upload(ref id) => { + let Some((uploads, _)) = &self.uploads else { + return Ok(err("this server takes no uploads; pass the bundle as \ + bundle_path or bundle_base64" + .into())); + }; + let Some(taken) = uploads.take(id, &sid).await else { + return Ok(err(format!( + "no upload '{id}' for this session: it may have expired ({} minutes \ + after the upload) or been deployed already; call {UPLOAD_TOOL} and \ + upload the archive again", + UPLOAD_TTL.as_secs() / 60 + ))); + }; + let path = match Utf8PathBuf::from_path_buf(taken.path().to_path_buf()) { + Ok(p) => p, + Err(p) => return Ok(err(format!("upload path is not UTF-8: {}", p.display()))), + }; + request.bundle = BundleSource::Path(path); + Some(taken) + } + _ => None, + }; // A per-app delegation lives five minutes and a deploy can outlast // it (an asset sync, several installs), so the identity the agent // signs with is kept fresh for as long as the deploy runs. @@ -433,7 +613,11 @@ impl IcDeployTools { let (lines, receiver) = tokio::sync::mpsc::unbounded_channel(); let forwarder = tokio::spawn(forward_lines(ctx.peer.clone(), ctx.meta.get_progress_token(), receiver)); - let watch = Watch { stop: ctx.ct.child_token(), lines: Some(lines) }; + // The stop: the client cancelling the request, or (on a server that + // says so) going away. + let stop = ctx.ct.child_token(); + let _link = link_client_gone(&ctx, stop.clone()); + let watch = Watch { stop, lines: Some(lines) }; let result = deploy_bundle(identity, request, watch).await; // The sender went with the run; the forwarder ends once it has // delivered what is left. @@ -496,6 +680,94 @@ impl IcDeployTools { let output = DeployGuideOutput { content: DEPLOY_SKILL_MD.to_string() }; Ok(ok_structured(output.content.clone(), &output)) } + + #[tool( + name = "icp_deploy_upload", + description = "Get a one-time ticket to upload a bundle archive to this server, for icp_deploy_bundle to deploy: this server runs elsewhere than the project, so the archive `icp project bundle` wrote cannot be read by path. Returns the upload URL, the ticket, and the exact curl command to run; the command prints a JSON answer whose `upload_id` is what to pass to icp_deploy_bundle as `bundle_upload`. The ticket is good for a few minutes and one upload, and the upload for a quarter of an hour; ask for a new ticket after either lapses.", + annotations(title = "Get a bundle upload ticket", read_only_hint = false, destructive_hint = false, idempotent_hint = false, open_world_hint = false), + output_schema = schema_for_output::(), + )] + async fn icp_deploy_upload( + &self, + Parameters(_args): Parameters, + ctx: RequestContext, + ) -> Result { + let Some((uploads, url)) = &self.uploads else { + return Ok(err("this server takes no uploads; pass the bundle as bundle_path or \ + bundle_base64" + .into())); + }; + // A ticket is for a session that could deploy what it uploads. + let Some(sid) = (self.session)(&ctx) else { + return Ok(err("uploading a bundle needs an authenticated session".into())); + }; + if let Err(e) = self.identities.require_write(&sid).await { + return Ok(err(e)); + } + let minted = uploads.mint(&sid).await; + let output = UploadTicketOutput { + curl: format!( + "curl -fsS -T -H '{TICKET_HEADER}: {}' '{url}'", + minted.ticket + ), + upload_url: url.clone(), + ticket: minted.ticket, + header: TICKET_HEADER.into(), + expires_in_secs: minted.expires_in.as_secs(), + max_bytes: uploads.max_bytes(), + }; + let text = format!( + "Upload the bundle within {} minutes by running, with the archive's path in place of \ + :\n\n{}\n\nThe command prints a JSON answer; pass its `upload_id` to \ + icp_deploy_bundle as `bundle_upload` within {} minutes. The archive may be at most \ + {} bytes. The ticket is good for one upload.", + output.expires_in_secs / 60, + output.curl, + UPLOAD_TTL.as_secs() / 60, + output.max_bytes + ); + Ok(ok_structured(text, &output)) + } +} + +/// Tie `stop` to the transport's [`ClientGone`] token when the request carries +/// one, for as long as the returned guard lives: the deploy stops when the +/// client goes, and the tie is undone when the deploy returns. +fn link_client_gone( + ctx: &RequestContext, + stop: CancellationToken, +) -> Option { + let gone = ctx + .extensions + .get::() + .and_then(|parts| parts.extensions.get::()) + .map(|g| g.0.clone())?; + let done = CancellationToken::new(); + let watched = done.clone(); + tokio::spawn(async move { + tokio::select! { + _ = gone.cancelled() => stop.cancel(), + _ = watched.cancelled() => {} + } + }); + Some(done.drop_guard()) +} + +/// What `icp_deploy_upload` returns. +#[derive(Debug, Serialize, schemars::JsonSchema)] +pub struct UploadTicketOutput { + /// The absolute URL to `PUT` the archive to. + pub upload_url: String, + /// The one-time ticket, sent in the request header named by `header`. + pub ticket: String, + /// The request header the ticket goes in. + pub header: String, + /// Seconds the ticket is good for. + pub expires_in_secs: u64, + /// The most bytes the archive may have. + pub max_bytes: usize, + /// The command to run, with the archive's path in place of ``. + pub curl: String, } /// Deliver each log line of a running deploy to the client as it is made: as @@ -635,7 +907,9 @@ mod tests { #[test] fn the_router_carries_the_deploy_tool_the_balance_check_and_the_guide() { let tools = IcDeployTools::tool_router().list_all(); - assert_eq!(tools.len(), 3); + assert_eq!(tools.len(), 4); + let upload = tools.iter().find(|t| &*t.name == UPLOAD_TOOL).expect("upload ticket tool"); + assert_eq!(upload.annotations.clone().expect("annotations").destructive_hint, Some(false)); let guide = tools.iter().find(|t| &*t.name == "icp_deploy_guide").expect("guide"); let ann = guide.annotations.clone().expect("annotations"); assert_eq!(ann.read_only_hint, Some(true)); @@ -677,9 +951,18 @@ mod tests { let both = args(serde_json::json!({ "bundle_path": "/tmp/b.tar.gz", "bundle_base64": "AAAA", "environment": "ic" })); - assert!(request_from_args(both).unwrap_err().contains("not both")); + assert!(request_from_args(both).unwrap_err().contains("not several")); + let three = args(serde_json::json!({ + "bundle_path": "/tmp/b.tar.gz", "bundle_upload": "abc", "environment": "ic" + })); + assert!(request_from_args(three).unwrap_err().contains("not several")); let neither = args(serde_json::json!({ "environment": "ic" })); - assert!(request_from_args(neither).unwrap_err().contains("bundle_path or bundle_base64")); + assert!(request_from_args(neither).unwrap_err().contains("bundle_upload")); + let upload = args(serde_json::json!({ "bundle_upload": " abc ", "environment": "ic" })); + let request = request_from_args(upload).unwrap(); + assert!(matches!(request.bundle, BundleSource::Upload(ref id) if id == "abc")); + let empty = args(serde_json::json!({ "bundle_upload": " ", "environment": "ic" })); + assert!(request_from_args(empty).unwrap_err().contains("empty")); let bad = args(serde_json::json!({ "bundle_base64": "***", "environment": "ic" })); assert!(request_from_args(bad).unwrap_err().contains("base64")); } diff --git a/crates/imcp2-deploy/src/server.rs b/crates/imcp2-deploy/src/server.rs new file mode 100644 index 0000000..ed3f35b --- /dev/null +++ b/crates/imcp2-deploy/src/server.rs @@ -0,0 +1,167 @@ +//! The handler a server mounts to serve the deploy tools next to its own: an +//! inner [`ServerHandler`] (the core surface, or a wrapper of it) plus +//! [`IcDeployTools`], answering `tools/list` with the merged list, dispatching +//! `tools/call` to the deploy router first and the inner handler otherwise, +//! listing and reading the deploy skill next to the inner resources, and +//! appending to the inner instructions the paragraph that says deploying is +//! possible here, and how. +//! +//! Both servers compose this: the local one around its login tools, the +//! hosted one around the core surface. What differs between them — how the +//! bundle reaches the server, how many deploys run at once — the deploy tools +//! know themselves, so the paragraph reads from them. + +use rmcp::{ + handler::server::tool::ToolCallContext, + model::{ + CallToolRequestParams, CallToolResult, ListResourcesResult, ListToolsResult, + PaginatedRequestParams, ReadResourceRequestParams, ReadResourceResult, Resource, + ResourceContents, ServerInfo, Tool, + }, + service::RequestContext, + ErrorData as McpError, RoleServer, ServerHandler, +}; + +use crate::{skill_document, IcDeployTools, DEPLOY_SKILL_NAME, DEPLOY_SKILL_TITLE}; + +/// An inner handler with the deploy tools served beside it. +pub struct WithDeploy { + inner: H, + deploy: IcDeployTools, +} + +impl WithDeploy { + pub fn new(inner: H, deploy: IcDeployTools) -> Self { + Self { inner, deploy } + } + + pub fn inner(&self) -> &H { + &self.inner + } + + pub fn deploy(&self) -> &IcDeployTools { + &self.deploy + } + + /// The deploy tools this server lists: `icp_deploy_upload` only where an + /// upload store exists for it to mint tickets on. + pub fn deploy_tools(&self) -> Vec { + IcDeployTools::tool_router() + .list_all() + .into_iter() + .filter(|t| self.deploy.has_uploads() || &*t.name != crate::UPLOAD_TOOL) + .collect() + } + + fn serves(&self, name: &str) -> bool { + IcDeployTools::tool_router().has_route(name) + && (self.deploy.has_uploads() || name != crate::UPLOAD_TOOL) + } + + /// The paragraph appended to the inner instructions. + fn deploying(&self) -> String { + let concurrency = match self.deploy.concurrency() { + 1 => "runs one deploy at a time".to_string(), + n => format!("runs at most {n} deploys at a time, one per session"), + }; + let bundle = if self.deploy.has_uploads() { + "The bundle reaches this server through `icp_deploy_upload`: it returns a one-time \ + ticket and the curl command that uploads the archive, and the upload id that \ + command prints is what `icp_deploy_bundle` takes as `bundle_upload`." + } else { + "Pass the bundle as `bundle_path` (an absolute path on this machine) or \ + `bundle_base64`." + }; + format!( + "DEPLOYING. This server adds `icp_deploy_bundle`, the one exception to the rule above \ + that no tool deploys: it deploys an icp project bundle (`icp project bundle --output \ + `, built in the user's environment) to the environment it names, creating the \ + canisters that do not exist yet from the user's cycles balance (`icp_cycles_balance` \ + reports it) and upgrading those passed in `canister_ids`. It deploys as the user's \ + default Internet Identity account at https://cli.id.ai — the principal `icp identity \ + link web` links the CLI to, and so the CLI's own only when the CLI's selected \ + identity is that link; a local key or another account is a different principal, \ + which `icp identity principal` shows — not as the management identity the canister \ + tools act as. It needs an \"Actions & questions\" session, {concurrency}, lasts as \ + long as the deploy (send a progressToken to see each phase as it completes; \ + cancelling the call interrupts the deploy at its next step), and returns the \ + canister-id mapping to pass back next time. {bundle} The `skill://{DEPLOY_SKILL_NAME}` \ + resource walks through the loop: bundle, deploy, write the mapping back into the \ + project; `icp_deploy_guide` returns the same text." + ) + } +} + +impl ServerHandler for WithDeploy { + fn get_info(&self) -> ServerInfo { + let mut info = self.inner.get_info(); + // A running deploy reports each phase mid-call: as progress when the + // request carried a token, else as a logging message, which a client + // only expects from a server that declares it. + info.capabilities.logging = Some(Default::default()); + let inner = info.instructions.take().unwrap_or_default(); + info.instructions = Some(format!("{inner}\n\n{}", self.deploying())); + info + } + + async fn list_tools( + &self, + request: Option, + context: RequestContext, + ) -> Result { + let mut result = self.inner.list_tools(request, context).await?; + result.tools.extend(self.deploy_tools()); + Ok(result) + } + + async fn call_tool( + &self, + request: CallToolRequestParams, + context: RequestContext, + ) -> Result { + if self.serves(&request.name) { + let tcc = ToolCallContext::new(&self.deploy, request, context); + IcDeployTools::tool_router().call(tcc).await + } else { + self.inner.call_tool(request, context).await + } + } + + fn get_tool(&self, name: &str) -> Option { + if self.serves(name) { + IcDeployTools::tool_router().get(name).cloned() + } else { + self.inner.get_tool(name) + } + } + + async fn list_resources( + &self, + request: Option, + context: RequestContext, + ) -> Result { + let first_page = request.as_ref().and_then(|r| r.cursor.as_ref()).is_none(); + let mut result = self.inner.list_resources(request, context).await?; + // The one skill the deploy tools add, next to the inner skills. Core + // lists everything on one page today; should it ever page, the skill + // belongs on the first page alone, not on every one. + if first_page { + result.resources.push(Resource::new( + format!("skill://{DEPLOY_SKILL_NAME}"), + format!("IC skill: {DEPLOY_SKILL_TITLE}"), + )); + } + Ok(result) + } + + async fn read_resource( + &self, + request: ReadResourceRequestParams, + context: RequestContext, + ) -> Result { + if let Some(md) = request.uri.strip_prefix("skill://").and_then(skill_document) { + return Ok(ReadResourceResult::new(vec![ResourceContents::text(md, request.uri)])); + } + self.inner.read_resource(request, context).await + } +} diff --git a/crates/imcp2-deploy/src/uploads.rs b/crates/imcp2-deploy/src/uploads.rs new file mode 100644 index 0000000..b33bc3b --- /dev/null +++ b/crates/imcp2-deploy/src/uploads.rs @@ -0,0 +1,348 @@ +//! Bundles uploaded ahead of a deploy, for a server the agent cannot hand a +//! file to. +//! +//! An MCP tool call carries JSON the model writes, and a model cannot write +//! an archive, so a server on another machine than the agent's needs the +//! bytes to arrive some other way. This is that way: `icp_deploy_upload` +//! mints a ticket inside an authenticated tool call, the agent presents it on +//! an ordinary `PUT` of the archive (the one thing the ticket is good for), +//! and the upload id that `PUT` answers with is what `icp_deploy_bundle` +//! takes as `bundle_upload`. The ticket derives its authority from the bearer +//! token presented on the tool call — the same chain of trust as any tool +//! call — scoped down to one `PUT`: +//! +//! * 256 bits of randomness, single use (taken before a byte of the body +//! is read), and gone after [`TICKET_TTL`]; +//! * one pending ticket per session — minting another drops the previous; +//! * bound to the session that minted it: the upload it produces is +//! deployable by that session alone, so an id leaking elsewhere is useless; +//! * the body is streamed to disk under the cap, a declared length over the +//! cap refused unread, so a ticket flood costs a lookup each. +//! +//! An upload is held for [`UPLOAD_TTL`] for its deploy to begin; the deploy +//! that takes it removes it, and one never taken is swept at expiry. Each +//! archive is a file of its own under the store's directory, readable by the +//! server's user alone. + +use std::{ + collections::HashMap, + path::{Path, PathBuf}, + sync::Arc, + time::{Duration, Instant}, +}; + +use tokio::{ + io::{AsyncRead, AsyncReadExt, AsyncWriteExt}, + sync::Mutex, +}; + +/// The default cap on an uploaded archive, in bytes: a real bundle is a +/// manifest, a few modules and a frontend's assets, well inside this. +pub const DEFAULT_MAX_UPLOAD_BYTES: usize = 64 * 1024 * 1024; +/// How long a ticket is good for: minutes, enough to run one `curl`. +pub const TICKET_TTL: Duration = Duration::from_secs(5 * 60); +/// How long an upload waits for its deploy to begin. +pub const UPLOAD_TTL: Duration = Duration::from_secs(15 * 60); +/// The request header a `PUT` presents its ticket in — a header rather than +/// the path, so the ticket stays out of a proxy's access log. +pub const TICKET_HEADER: &str = "x-deploy-upload-ticket"; + +/// The store: a directory of received archives and the tickets and uploads +/// outstanding. Cheap to clone; one per server. +#[derive(Clone)] +pub struct Uploads { + inner: Arc, +} + +struct Inner { + dir: PathBuf, + max_bytes: usize, + state: Mutex, +} + +#[derive(Default)] +struct State { + tickets: HashMap, + uploads: HashMap, +} + +struct Ticket { + session: String, + expires: Instant, +} + +struct Upload { + session: String, + path: PathBuf, + expires: Instant, +} + +/// A ticket just minted, and how long it is good for. +#[derive(Debug)] +pub struct Minted { + pub ticket: String, + pub expires_in: Duration, +} + +/// An archive just received: the id its deploy names, its size, and how long +/// the store holds it. +#[derive(Debug)] +pub struct Received { + pub upload_id: String, + pub bytes: usize, + pub expires_in: Duration, +} + +/// Why a `PUT` was refused. +#[derive(Debug, PartialEq, Eq)] +pub enum ReceiveError { + /// No such ticket, or one already used or expired. + UnknownTicket, + /// The body is over the cap (the value); nothing is kept. + TooLarge(usize), + /// The body could not be read, or the file not written. + Io(String), +} + +impl std::fmt::Display for ReceiveError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + ReceiveError::UnknownTicket => f.write_str( + "unknown, used or expired upload ticket; call icp_deploy_upload for a new one", + ), + ReceiveError::TooLarge(max) => write!(f, "the archive is over the {max}-byte limit"), + ReceiveError::Io(e) => write!(f, "could not receive the archive: {e}"), + } + } +} + +/// An upload handed to its deploy: the archive's path, and the file's removal +/// when this is dropped, whichever way the deploy ends. +#[derive(Debug)] +pub struct TakenUpload { + path: PathBuf, +} + +impl TakenUpload { + pub fn path(&self) -> &Path { + &self.path + } +} + +impl Drop for TakenUpload { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.path); + } +} + +impl Uploads { + /// A store writing its archives under `dir` (made on first use, readable by + /// the server's user alone), each at most `max_bytes`. + pub fn new(dir: PathBuf, max_bytes: usize) -> Self { + Self { inner: Arc::new(Inner { dir, max_bytes, state: Mutex::default() }) } + } + + pub fn max_bytes(&self) -> usize { + self.inner.max_bytes + } + + /// Mint a ticket for `session`, dropping any it already holds. + pub async fn mint(&self, session: &str) -> Minted { + let mut state = self.inner.state.lock().await; + sweep(&mut state); + state.tickets.retain(|_, t| t.session != session); + let ticket = random_token(); + state.tickets.insert( + ticket.clone(), + Ticket { session: session.to_string(), expires: Instant::now() + TICKET_TTL }, + ); + Minted { ticket, expires_in: TICKET_TTL } + } + + /// Receive an archive on `ticket`: the ticket is spent first, a declared + /// length over the cap is refused unread, and the body is streamed to a + /// file up to the cap and no further. + pub async fn receive( + &self, + ticket: &str, + declared_len: Option, + mut body: impl AsyncRead + Unpin, + ) -> Result { + let max = self.inner.max_bytes; + let session = { + let mut state = self.inner.state.lock().await; + sweep(&mut state); + state.tickets.remove(ticket).ok_or(ReceiveError::UnknownTicket)?.session + }; + if declared_len.is_some_and(|len| len > max as u64) { + return Err(ReceiveError::TooLarge(max)); + } + make_private_dir(&self.inner.dir).map_err(|e| ReceiveError::Io(e.to_string()))?; + let upload_id = random_token(); + let path = self.inner.dir.join(format!("{upload_id}.tar.gz")); + let received = write_up_to(&path, &mut body, max).await; + let bytes = match received { + Ok(bytes) => bytes, + Err(e) => { + let _ = tokio::fs::remove_file(&path).await; + return Err(e); + } + }; + let mut state = self.inner.state.lock().await; + state.uploads.insert( + upload_id.clone(), + Upload { session, path, expires: Instant::now() + UPLOAD_TTL }, + ); + Ok(Received { upload_id, bytes, expires_in: UPLOAD_TTL }) + } + + /// Hand `upload_id` to its deploy, if `session` is the one that uploaded + /// it: the archive's path, removed when the returned handle drops. + pub async fn take(&self, upload_id: &str, session: &str) -> Option { + let mut state = self.inner.state.lock().await; + sweep(&mut state); + let owned = state.uploads.get(upload_id).is_some_and(|u| u.session == session); + if !owned { + return None; + } + state.uploads.remove(upload_id).map(|u| TakenUpload { path: u.path }) + } +} + +/// Write `body` to a new file at `path`, readable by the owner alone, refusing +/// — and reading no further — once more than `max` bytes have arrived. +async fn write_up_to( + path: &Path, + body: &mut (impl AsyncRead + Unpin), + max: usize, +) -> Result { + let mut options = tokio::fs::OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + options.mode(0o600); + let mut file = options.open(path).await.map_err(|e| ReceiveError::Io(e.to_string()))?; + let mut total = 0usize; + let mut buf = vec![0u8; 64 * 1024]; + loop { + let n = body.read(&mut buf).await.map_err(|e| ReceiveError::Io(e.to_string()))?; + if n == 0 { + break; + } + total += n; + if total > max { + return Err(ReceiveError::TooLarge(max)); + } + file.write_all(&buf[..n]).await.map_err(|e| ReceiveError::Io(e.to_string()))?; + } + file.flush().await.map_err(|e| ReceiveError::Io(e.to_string()))?; + Ok(total) +} + +/// Drop expired tickets, and expired uploads with their files. +fn sweep(state: &mut State) { + let now = Instant::now(); + state.tickets.retain(|_, t| t.expires > now); + let expired: Vec = + state.uploads.iter().filter(|(_, u)| u.expires <= now).map(|(id, _)| id.clone()).collect(); + for id in expired { + if let Some(upload) = state.uploads.remove(&id) { + let _ = std::fs::remove_file(&upload.path); + } + } +} + +/// `dir`, created if missing, and on Unix readable by its owner alone. +fn make_private_dir(dir: &Path) -> std::io::Result<()> { + if dir.is_dir() { + return Ok(()); + } + #[cfg(unix)] + { + use std::os::unix::fs::DirBuilderExt; + std::fs::DirBuilder::new().recursive(true).mode(0o700).create(dir) + } + #[cfg(not(unix))] + std::fs::create_dir_all(dir) +} + +/// 256 random bits, hex-encoded: a ticket or an upload id. +fn random_token() -> String { + let mut bytes = [0u8; 32]; + getrandom::fill(&mut bytes).expect("the OS random source is available"); + hex::encode(bytes) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn store(max: usize) -> (tempfile::TempDir, Uploads) { + let dir = tempfile::tempdir().unwrap(); + let uploads = Uploads::new(dir.path().join("uploads"), max); + (dir, uploads) + } + + #[tokio::test] + async fn a_ticket_receives_one_archive_for_its_session_alone() { + let (_dir, uploads) = store(1024); + let minted = uploads.mint("sess-a").await; + assert_eq!(minted.ticket.len(), 64); + + let received = uploads + .receive(&minted.ticket, Some(5), std::io::Cursor::new(b"hello".to_vec())) + .await + .unwrap(); + assert_eq!(received.bytes, 5); + // Spent: a second PUT on the same ticket is refused. + let again = uploads.receive(&minted.ticket, None, std::io::Cursor::new(Vec::new())).await; + assert_eq!(again.unwrap_err(), ReceiveError::UnknownTicket); + + // Another session cannot take it; the owner can, once. + assert!(uploads.take(&received.upload_id, "sess-b").await.is_none()); + let taken = uploads.take(&received.upload_id, "sess-a").await.expect("owner takes it"); + assert_eq!(std::fs::read(taken.path()).unwrap(), b"hello"); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let mode = std::fs::metadata(taken.path()).unwrap().permissions().mode() & 0o777; + assert_eq!(mode, 0o600, "mode {mode:o}"); + } + assert!(uploads.take(&received.upload_id, "sess-a").await.is_none()); + let path = taken.path().to_path_buf(); + drop(taken); + assert!(!path.exists(), "the archive goes with the handle"); + } + + #[tokio::test] + async fn an_archive_over_the_cap_is_refused_and_nothing_kept() { + let (dir, uploads) = store(16); + // Declared over the cap: refused unread. + let t = uploads.mint("s").await; + let err = uploads.receive(&t.ticket, Some(17), std::io::Cursor::new(vec![0; 17])).await; + assert_eq!(err.unwrap_err(), ReceiveError::TooLarge(16)); + // Undeclared and over the cap: refused once the cap is passed, the + // partial file removed. + let t = uploads.mint("s").await; + let err = uploads.receive(&t.ticket, None, std::io::Cursor::new(vec![0; 17])).await; + assert_eq!(err.unwrap_err(), ReceiveError::TooLarge(16)); + let kept = std::fs::read_dir(dir.path().join("uploads")).unwrap().count(); + assert_eq!(kept, 0, "nothing is kept of a refused upload"); + // Exactly at the cap passes. + let t = uploads.mint("s").await; + let ok = uploads.receive(&t.ticket, None, std::io::Cursor::new(vec![0; 16])).await.unwrap(); + assert_eq!(ok.bytes, 16); + } + + #[tokio::test] + async fn minting_again_drops_the_previous_ticket_and_unknown_tickets_cost_nothing() { + let (dir, uploads) = store(1024); + let first = uploads.mint("s").await; + let second = uploads.mint("s").await; + let err = uploads.receive(&first.ticket, None, std::io::Cursor::new(vec![1])).await; + assert_eq!(err.unwrap_err(), ReceiveError::UnknownTicket); + let err = uploads.receive("not-a-ticket", None, std::io::Cursor::new(vec![1])).await; + assert_eq!(err.unwrap_err(), ReceiveError::UnknownTicket); + assert!(!dir.path().join("uploads").exists(), "no directory is made for a refusal"); + assert!(uploads.receive(&second.ticket, None, std::io::Cursor::new(vec![1])).await.is_ok()); + } +} diff --git a/crates/imcp2-deploy/static/skills/deploy-with-mcp.md b/crates/imcp2-deploy/static/skills/deploy-with-mcp.md index ef1fb9d..ef3a764 100644 --- a/crates/imcp2-deploy/static/skills/deploy-with-mcp.md +++ b/crates/imcp2-deploy/static/skills/deploy-with-mcp.md @@ -22,7 +22,9 @@ A deploy is three steps, in order: 1. Bundle the project with the icp CLI. 2. Call `icp_deploy_bundle` with the bundle, the environment, and the - environment's current canister-id mapping. + environment's current canister-id mapping. On a server that runs + elsewhere than the project — one that lists `icp_deploy_upload` — the + bundle is uploaded first, on a ticket that tool returns. 3. Write the mapping the tool returns back into the project. Skipping step 3 is the mistake that costs the most: the next deploy would @@ -98,6 +100,25 @@ It is a JSON object from canister name to principal: Read it if it exists. If it does not, this is the first deploy of that environment and every canister will be created. +### Upload the bundle, on a server that cannot read your files + +A server that lists `icp_deploy_upload` runs on another machine than the +project, so it cannot open `bundle_path`, and a tool call cannot carry the +archive itself. Upload it first: + +1. Call `icp_deploy_upload`. It returns a one-time ticket, good for a few + minutes, and the exact `curl` command to run. +2. Run that command with the archive's path in place of ``. + It `PUT`s the archive to the server and prints a JSON answer whose + `upload_id` names the upload, held for a quarter of an hour. +3. Pass that id to `icp_deploy_bundle` as `bundle_upload`, in place of + `bundle_path`. + +A ticket is spent by one upload, successful or not; an upload is spent by +one deploy. If either lapses, ask for a new ticket and upload again. A +server that does not list `icp_deploy_upload` takes `bundle_path` or +`bundle_base64` and refuses `bundle_upload`. + ### Call the tool Pass the bundle, the environment name, and the mapping as `canister_ids`: @@ -113,9 +134,10 @@ Pass the bundle, the environment name, and the mapping as `canister_ids`: } ``` -- `bundle_path` must be absolute. When the server cannot read your - filesystem, send the archive inline as `bundle_base64` instead; exactly one - of the two is accepted. +- `bundle_path` must be absolute. On a server that cannot read your + filesystem, pass `bundle_upload` from the upload above instead (or, for a + client that can attach the bytes itself, `bundle_base64`); exactly one of + the three is accepted. - `environment` is the name as the manifest declares it, `ic` for a default project. - `canister_ids` is the file's content. Omit it, or send `{}`, on a first @@ -195,19 +217,21 @@ interchangeably for that environment. 6. **Expecting the server to keep state.** It keeps nothing between calls: no bundle, no mapping. The project's mapping file is the record. The one exception is the working directory of a call that never returned, see 8. -7. **Deploying twice at once.** One deploy runs at a time on this server; a - second call while one runs is refused, not queued. Wait for the answer, - do step 3, then deploy again from the updated mapping. +7. **Deploying twice at once.** One deploy runs at a time per session, and + a server caps how many run at once in all; a call past either is refused, + not queued. Wait for the answer, do step 3, then deploy again from the + updated mapping; past the server's cap, wait and try again. 8. **A call that never returned.** Cancelling the call, or losing the connection, interrupts the deploy at its next step; a canister call already in flight completes on the network. If the server stopped mid-deploy, or you cancelled and never read the answer, the canisters it - created are not lost: the deploy's working directory is kept under the - user's local data directory + created are not lost: the deploy's working directory is kept — on the + local server under the user's local data directory (`~/.local/share/imcp2-local/deploys/` on Linux, `~/Library/Application Support/imcp2-local/deploys/` on macOS, - `%LOCALAPPDATA%\imcp2-local\deploys\` on Windows), and its - `.icp/data/mappings/.ids.json` records every canister the - deploy had created. Do step 3 with that file's contents, delete the - directory, and only then deploy again; otherwise the retry creates a - second set. + `%LOCALAPPDATA%\imcp2-local\deploys\` on Windows; on a hosted server + under its state directory, where its operators can read it for the + user), and its `.icp/data/mappings/.ids.json` records every + canister the deploy had created. Do step 3 with that file's contents, + delete the directory, and only then deploy again; otherwise the retry + creates a second set. diff --git a/crates/imcp2-local/README.md b/crates/imcp2-local/README.md index 6ab7b76..d5af910 100644 --- a/crates/imcp2-local/README.md +++ b/crates/imcp2-local/README.md @@ -212,10 +212,12 @@ rather than creates. Not every client reads resources, so `icp_deploy_guide` returns the same document as a tool result. It lives at `crates/imcp2-deploy/static/skills/deploy-with-mcp.md`. -Why only here: the tool is built on `icp-project`, a crate the icp CLI does not -publish, so it is pulled from git — which the crates.io-published `imcp2-core` -and `imcp2` cannot do. It lives in its own `imcp2-deploy` crate, composed into -this binary alone. +The tool is built on `icp-project`, a crate the icp CLI does not publish, so +it is pulled from git — which the crates.io-published `imcp2-core` and `imcp2` +cannot do. It lives in its own `imcp2-deploy` crate, composed into this binary +and, behind the `deploy` feature, into the hosted server (see the repository +README), where the bundle arrives through an upload ticket since that server +cannot read the user's files. ## Upgrading diff --git a/crates/imcp2-local/src/server.rs b/crates/imcp2-local/src/server.rs index b9a7781..62b2fe9 100644 --- a/crates/imcp2-local/src/server.rs +++ b/crates/imcp2-local/src/server.rs @@ -1,34 +1,29 @@ -//! The local MCP server: [`IcTools`] wrapped in a thin handler that adds the -//! two **local-only login tools** and the **deploy tool**. The login tools +//! The local MCP server: [`IcTools`] with the **deploy tools** beside it +//! ([`WithDeploy`], the handler the hosted server mounts too), wrapped in a +//! thin handler that adds the two **local-only login tools**. The login tools //! cannot live in `imcp2-core` — they drive this binary's loopback listener //! and browser opener, which core cannot call back into — so this wrapper -//! carries them in its own `#[tool_router]`. The deploy tools -//! (`icp_deploy_bundle`, `icp_cycles_balance` and `icp_deploy_guide`, -//! [`IcDeployTools`]) cannot either, for a different reason: it runs on the icp CLI's project library, a git dependency the -//! published crates cannot carry. The wrapper answers `tools/list` with the -//! merged list, dispatches `tools/call` to its own router, then the deploy -//! router, and forwards everything else to the inner [`IcTools`] handler -//! verbatim. The hosted server contains none of this code, so its tool surface -//! stays login- and deploy-free by construction (a core-side test guards the -//! boundary too). +//! carries them in its own `#[tool_router]`, answers `tools/list` with the +//! merged list, dispatches `tools/call` to its own router first, and forwards +//! everything else to the inner handler verbatim. The hosted server contains +//! no login tool, so its surface stays login-free by construction (a +//! core-side test guards the boundary too). use rmcp::handler::server::tool::ToolCallContext; use rmcp::handler::server::wrapper::Parameters; use rmcp::model::{ CallToolRequestParams, CallToolResult, ContentBlock, ListResourcesResult, ListToolsResult, - PaginatedRequestParams, ReadResourceRequestParams, ReadResourceResult, Resource, - ResourceContents, Tool, + PaginatedRequestParams, ReadResourceRequestParams, ReadResourceResult, Tool, }; use rmcp::service::RequestContext; use rmcp::{schemars, tool, tool_router, ErrorData as McpError, RoleServer, ServerHandler}; use crate::login::{BeginOutcome, Grant, LoginDriver, LoginStatus}; use imcp2_core::IcTools; -use imcp2_deploy::IcDeployTools; +use imcp2_deploy::{IcDeployTools, WithDeploy}; pub struct LocalServer { - tools: IcTools, - deploy: IcDeployTools, + tools: WithDeploy, login: LoginDriver, /// Whether the login flow will also try to open the browser itself — /// only phrasing: the link is always returned in-band. @@ -167,7 +162,13 @@ fn human_minutes(mins: u64) -> String { #[tool_router] impl LocalServer { pub fn new(tools: IcTools, deploy: IcDeployTools, login: LoginDriver, auto_open: bool) -> Self { - Self { tools, deploy, login, auto_open } + Self { tools: WithDeploy::new(tools, deploy), login, auto_open } + } + + /// The deploy tools this server serves. + #[cfg(test)] + pub fn deploy(&self) -> &IcDeployTools { + self.tools.deploy() } #[tool( @@ -264,37 +265,16 @@ impl ServerHandler for LocalServer { let mut info = self.tools.get_info(); info.server_info.name = "imcp2-local".into(); info.server_info.version = env!("CARGO_PKG_VERSION").into(); - // A running deploy reports each phase mid-call: as progress when the - // request carried a token, else as a logging message, which a client - // only expects from a server that declares it. - info.capabilities.logging = Some(Default::default()); - // The core instructions apply verbatim; add how signing in works here - // (the hosted server logs in via OAuth outside the tool surface, so - // core's text says nothing about it). - let core = info.instructions.take().unwrap_or_default(); + // The core and deploy instructions apply verbatim; add how signing in + // works here (the hosted server logs in via OAuth outside the tool + // surface, so neither says anything about it). + let inner = info.instructions.take().unwrap_or_default(); info.instructions = Some(format!( - "{core}\n\n\ + "{inner}\n\n\ SIGNING IN. On this local server the user signs in through `authenticate`, which \ returns an id.ai link for them to open. Sessions are held in memory, so the user \ signs in again after a restart, when the grant expires, or if they revoke it at \ - id.ai.\n\n\ - DEPLOYING. This local server adds `icp_deploy_bundle`, the one exception to the \ - rule above that no tool deploys: it deploys an icp project bundle (`icp project \ - bundle --output `, built in the user's environment) to the environment it \ - names, creating the canisters that do not exist yet from the user's cycles \ - balance (`icp_cycles_balance` reports it) and upgrading those passed in \ - `canister_ids`. It deploys as the user's default Internet Identity account at \ - https://cli.id.ai — the principal `icp identity link web` links the CLI to, and \ - so the CLI's own only when the CLI's selected identity is that link; a local key \ - or another account is a different principal, which `icp identity principal` \ - shows — not as the management identity the canister tools act as. It needs an \ - \"Actions & questions\" session, runs one deploy at a time, lasts as long as the \ - deploy (send a progressToken to see each phase as it completes; cancelling the \ - call interrupts the deploy at its next step), and returns the canister-id \ - mapping to pass back next time. \ - The `skill://deploy-with-mcp` resource walks through the loop: bundle, deploy, \ - write the mapping back into the project; `icp_deploy_guide` returns the same \ - text." + id.ai." )); info } @@ -306,7 +286,6 @@ impl ServerHandler for LocalServer { ) -> Result { let mut result = self.tools.list_tools(request, context).await?; result.tools.extend(Self::tool_router().list_all()); - result.tools.extend(IcDeployTools::tool_router().list_all()); Ok(result) } @@ -318,20 +297,13 @@ impl ServerHandler for LocalServer { if Self::tool_router().has_route(&request.name) { let tcc = ToolCallContext::new(self, request, context); Self::tool_router().call(tcc).await - } else if IcDeployTools::tool_router().has_route(&request.name) { - let tcc = ToolCallContext::new(&self.deploy, request, context); - IcDeployTools::tool_router().call(tcc).await } else { self.tools.call_tool(request, context).await } } fn get_tool(&self, name: &str) -> Option { - Self::tool_router() - .get(name) - .cloned() - .or_else(|| IcDeployTools::tool_router().get(name).cloned()) - .or_else(|| self.tools.get_tool(name)) + Self::tool_router().get(name).cloned().or_else(|| self.tools.get_tool(name)) } async fn list_resources( @@ -339,19 +311,7 @@ impl ServerHandler for LocalServer { request: Option, context: RequestContext, ) -> Result { - let first_page = request.as_ref().and_then(|r| r.cursor.as_ref()).is_none(); - let mut result = self.tools.list_resources(request, context).await?; - // The one skill this server adds, next to core's bundled IC skills: the - // deploy loop around `icp_deploy_bundle`, which only exists here. Core - // lists everything on one page today; should it ever page, the skill - // belongs on the first page alone, not on every one. - if first_page { - result.resources.push(Resource::new( - format!("skill://{}", imcp2_deploy::DEPLOY_SKILL_NAME), - format!("IC skill: {}", imcp2_deploy::DEPLOY_SKILL_TITLE), - )); - } - Ok(result) + self.tools.list_resources(request, context).await } async fn read_resource( @@ -359,11 +319,6 @@ impl ServerHandler for LocalServer { request: ReadResourceRequestParams, context: RequestContext, ) -> Result { - if let Some(md) = - request.uri.strip_prefix("skill://").and_then(imcp2_deploy::skill_document) - { - return Ok(ReadResourceResult::new(vec![ResourceContents::text(md, request.uri)])); - } self.tools.read_resource(request, context).await } } @@ -452,7 +407,7 @@ mod tests { async fn a_real_mcp_client_sees_the_merged_surface_and_the_login_flow() { let (client_io, server_io) = tokio::io::duplex(64 * 1024); let local = test_server(); - let deploy_tools = local.deploy.clone(); + let deploy_tools = local.deploy().clone(); let server = local.serve(server_io); let client = ().serve(client_io); let (server, client) = tokio::join!(server, client); diff --git a/deploy/native/README.md b/deploy/native/README.md index c11d78d..596f48f 100644 --- a/deploy/native/README.md +++ b/deploy/native/README.md @@ -183,6 +183,16 @@ The deploy sets `MCP_SERVE_BETA=1` for the `staging` environment (via beta Internet Identity instance at `/mcp-beta` alongside the production `/mcp`; a binary run without it serves `/mcp` (production II) alone. +The binary is built with the `deploy` feature (`build.sh`), so the MCP endpoint +also serves the deploy tools (see the repository README, "Deploying a project +bundle"): deploy directories and uploaded bundles go under the unit's state +directory (`/var/lib/imcp2/deploys`), `MCP_DEPLOY_MAX_CONCURRENT` (default 2) caps +the deploys running at once and `MCP_DEPLOY_MAX_UPLOAD_BYTES` (default 64 MiB) an +uploaded archive. Caddy's `read_body 30s` applies to the upload `PUT` too: an +archive at the cap needs roughly 2 MB/s upstream to arrive in time, and the +timeout is the knob to raise if that bites, since the app's own caps are what +bound memory and disk. + The `staging` environment also sets `SERVE_STATUS=1`. `deploy.sh` then keeps the Caddyfile's `/status/` block (between its `__STATUS_BEGIN__`/`__STATUS_END__` markers), so the status dashboard is published at `https:///status/`; without it diff --git a/deploy/native/build.sh b/deploy/native/build.sh index 39c3468..f671ce7 100755 --- a/deploy/native/build.sh +++ b/deploy/native/build.sh @@ -79,7 +79,7 @@ RUN printf 'deb [check-valid-until=no] https://snapshot.debian.org/archive/debia COPY Cargo.toml Cargo.lock ./ COPY src ./src COPY crates ./crates -RUN cargo build --release +RUN cargo build --release --features deploy FROM scratch AS bin COPY --from=build /app/target/release/imcp2 /imcp2 DOCKERFILE diff --git a/src/auth.rs b/src/auth.rs index fafe0f8..dcf7b84 100644 --- a/src/auth.rs +++ b/src/auth.rs @@ -3073,6 +3073,26 @@ fn token_ttl(default: Duration, grant_expiration_ns: Option, now_ns: u64) - } } +#[cfg(test)] +impl AuthStore { + /// A live bearer token for a made-up session, so a test in the crate root + /// can get past [`require_token`] without an Internet Identity handshake. + #[allow(dead_code)] // used by the `deploy` feature's tests + pub(crate) async fn test_bearer(&self) -> String { + let token = format!("mcp-token-test-{}", Uuid::new_v4()); + self.tokens.write().await.insert( + token.clone(), + TokenInfo { + principal: "test-principal".into(), + session_id: "test-session".into(), + created: Instant::now(), + ttl: Duration::from_secs(600), + }, + ); + token + } +} + /// Mint + store an access token bound to the session key's principal, its /// lifetime matched to the II grant's expiration (see [`token_ttl`]). async fn issue_token(store: &AuthStore, session_id: &str) -> Response { diff --git a/src/deploy.rs b/src/deploy.rs new file mode 100644 index 0000000..97ef52c --- /dev/null +++ b/src/deploy.rs @@ -0,0 +1,437 @@ +//! The deploy tools on the hosted server, behind the `deploy` feature: the +//! core surface served with [`imcp2_deploy`]'s tools beside it, a bundle +//! upload route, and a transport that streams a deploy's progress and tells +//! it when the client has gone. +//! +//! Three things differ from the rest of the MCP endpoint: +//! +//! * **The response framing of a deploy call.** The endpoint answers a POST +//! with plain JSON, the one message the handler returns. A deploy reports +//! each phase as a notification *before* its result, which plain JSON +//! cannot carry — in rmcp's JSON mode the first message out would be +//! taken for the result. So a `tools/call` of `icp_deploy_bundle` is +//! handed to a second rmcp service configured for Server-Sent-Events +//! responses, still stateless, whose stream carries every notification +//! and then the result; every other request keeps the JSON service. The +//! dispatch reads the JSON-RPC method and tool name off the buffered body. +//! * **The client going away.** The SSE response body is wrapped so that +//! its drop — the client disconnecting, or the stream ending — cancels a +//! token the deploy tool watches ([`ClientGone`]): a deploy nobody will +//! read the answer of stops at its next step. +//! * **How the bundle arrives.** The agent's machine cannot hand this server +//! a file, and a tool call carries only what the model writes, so an +//! archive is `PUT` to `{mcp_path}/deploy/upload` on a one-time ticket that +//! `icp_deploy_upload` mints inside an authenticated call (see +//! [`imcp2_deploy::Uploads`] for the ticket's properties). The route is +//! outside the bearer gate — the ticket is its authentication — and +//! streams the body to disk under its own cap. +//! +//! The working directories and uploads live under the state directory +//! (`{state_dir}/deploys`), each readable by the server's user alone. + +use std::sync::Arc; + +use axum::{ + body::Body, + extract::{Request, State}, + http::{header, Method, StatusCode}, + response::{IntoResponse, Response}, + routing::put, + Json, Router, +}; +use futures_util::{StreamExt, TryStreamExt}; +use imcp2_deploy::{ClientGone, IcDeployTools, ReceiveError, Uploads, WithDeploy, TICKET_HEADER}; +use rmcp::transport::{ + streamable_http_server::{session::local::LocalSessionManager, tower::StreamableHttpService}, + StreamableHttpServerConfig, +}; +use tokio_util::sync::CancellationToken; + +use crate::{auth, tools::IcTools, McpServer}; + +/// What a hosted deployment configures of the deploy tools. +#[derive(Clone, Debug)] +pub struct DeployConfig { + /// How many deploys run at once across every session (one per session + /// either way); past it a deploy is refused, not queued. + pub concurrent: usize, + /// The cap on an uploaded archive, in bytes. + pub max_upload_bytes: usize, +} + +impl Default for DeployConfig { + fn default() -> Self { + Self { concurrent: 2, max_upload_bytes: imcp2_deploy::DEFAULT_MAX_UPLOAD_BYTES } + } +} + +/// The path, under the MCP mount, an archive is `PUT` to. +pub const UPLOAD_PATH: &str = "/deploy/upload"; + +/// The most of a request body the dispatch buffers to read the method and +/// tool name: an MCP request is small JSON, and a bundle never rides inline +/// here. (The body cap in front of the endpoint, where there is one, holds +/// it lower still.) +const DISPATCH_MAX_BODY: usize = 4 * 1024 * 1024; + +type McpService = StreamableHttpService, LocalSessionManager>; + +/// The two transports and the upload store behind the deploy-aware endpoint. +pub(crate) struct DeployMcp { + json: McpService, + sse: McpService, + uploads: Uploads, +} + +impl McpServer { + /// The MCP endpoint with the deploy tools, as two routers: the dispatch + /// between the JSON and SSE transports, for the caller to put behind the + /// bearer gate, and the upload route, which goes outside it (the ticket + /// is its authentication). `transport` is the endpoint's configuration + /// as the plain endpoint would use it. + pub(crate) fn deploy_mcp_routers( + &self, + config: &DeployConfig, + transport: StreamableHttpServerConfig, + ) -> (Router, Router) { + let deploys_dir = self.state_dir().join("deploys"); + let uploads = Uploads::new(deploys_dir.join("uploads"), config.max_upload_bytes); + let upload_url = format!("{}{}{UPLOAD_PATH}", self.public_url, self.mcp_path); + let deploy = IcDeployTools::new(self.identities.clone(), auth::bearer_session_resolver()) + .concurrent(config.concurrent) + .deploys_in(&deploys_dir) + .unwrap_or_else(|e| panic!("the state directory must be a UTF-8 path: {e}")) + .with_uploads(uploads.clone(), upload_url); + let service = |transport: StreamableHttpServerConfig| { + let (agent, identities, deploy) = + (self.agent.clone(), self.identities.clone(), deploy.clone()); + StreamableHttpService::new( + move || { + Ok(WithDeploy::new( + IcTools::new( + agent.clone(), + identities.clone(), + auth::bearer_session_resolver(), + ), + deploy.clone(), + )) + }, + LocalSessionManager::default().into(), + transport, + ) + }; + let mcp = Arc::new(DeployMcp { + json: service(transport.clone().with_json_response(true)), + sse: service(transport.with_json_response(false)), + uploads, + }); + ( + Router::new().fallback(dispatch).with_state(mcp.clone()), + Router::new().route(UPLOAD_PATH, put(upload)).with_state(mcp), + ) + } +} + +/// Hand a deploy call to the SSE transport, with its response body watched +/// for the client going away, and everything else to the JSON transport. +async fn dispatch(State(mcp): State>, request: Request) -> Response { + let (parts, body) = request.into_parts(); + let bytes = match axum::body::to_bytes(body, DISPATCH_MAX_BODY).await { + Ok(bytes) => bytes, + Err(_) => return (StatusCode::PAYLOAD_TOO_LARGE, "request body too large").into_response(), + }; + let deploy_call = parts.method == Method::POST && is_deploy_call(&bytes); + let mut request = Request::from_parts(parts, Body::from(bytes)); + if !deploy_call { + return mcp.json.handle(request).await.into_response(); + } + let gone = CancellationToken::new(); + request.extensions_mut().insert(ClientGone(gone.clone())); + let response = mcp.sse.handle(request).await; + cancel_on_drop(response.into_response(), gone) +} + +/// Whether `body` is a JSON-RPC `tools/call` of `icp_deploy_bundle`. +fn is_deploy_call(body: &[u8]) -> bool { + let Ok(message) = serde_json::from_slice::(body) else { + return false; + }; + message["method"] == "tools/call" && message["params"]["name"] == "icp_deploy_bundle" +} + +/// `response` with its body wrapped so that dropping the body — the client +/// disconnecting, or the stream ending — cancels `gone`. +fn cancel_on_drop(response: Response, gone: CancellationToken) -> Response { + let (parts, body) = response.into_parts(); + let guard = gone.drop_guard(); + let stream = futures_util::stream::unfold( + (body.into_data_stream(), guard), + |(mut data, guard)| async move { data.next().await.map(|item| (item, (data, guard))) }, + ); + Response::from_parts(parts, Body::from_stream(stream)) +} + +/// What a successful `PUT` answers with. +#[derive(serde::Serialize)] +struct Uploaded { + upload_id: String, + bytes: usize, + expires_in_secs: u64, +} + +/// Receive an archive on the ticket in [`TICKET_HEADER`]. +async fn upload(State(mcp): State>, request: Request) -> Response { + let Some(ticket) = request.headers().get(TICKET_HEADER).and_then(|v| v.to_str().ok()) else { + return ( + StatusCode::UNAUTHORIZED, + format!("send the upload ticket in the {TICKET_HEADER} header"), + ) + .into_response(); + }; + let ticket = ticket.trim().to_string(); + let declared = request + .headers() + .get(header::CONTENT_LENGTH) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.trim().parse::().ok()); + let reader = tokio_util::io::StreamReader::new( + request.into_body().into_data_stream().map_err(std::io::Error::other), + ); + match mcp.uploads.receive(&ticket, declared, reader).await { + Ok(received) => ( + StatusCode::CREATED, + Json(Uploaded { + upload_id: received.upload_id, + bytes: received.bytes, + expires_in_secs: received.expires_in.as_secs(), + }), + ) + .into_response(), + Err(e @ ReceiveError::UnknownTicket) => { + (StatusCode::FORBIDDEN, e.to_string()).into_response() + } + Err(e @ ReceiveError::TooLarge(_)) => { + (StatusCode::PAYLOAD_TOO_LARGE, e.to_string()).into_response() + } + Err(e @ ReceiveError::Io(_)) => (StatusCode::BAD_REQUEST, e.to_string()).into_response(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{Agent, IiInstance, McpConfig, SharedClients, IC_URL}; + use axum::http::Request; + use tower::ServiceExt; + + fn server() -> (tempfile::TempDir, McpServer) { + let dir = tempfile::tempdir().unwrap(); + let state_dir = dir.path().to_path_buf(); + let server = McpServer::new(McpConfig { + agent: Agent::builder().with_url(IC_URL).build().expect("agent"), + instance: IiInstance::prod().expect("prod instance"), + public_url: "https://mcp.example.com".into(), + mcp_path: "/mcp".into(), + clients: SharedClients::load(&state_dir), + state_dir, + require_resource: true, + cimd_enabled: false, + }) + .with_deploy(DeployConfig { concurrent: 2, max_upload_bytes: 1024 }); + (dir, server) + } + + fn rpc(method: &str, params: serde_json::Value) -> String { + serde_json::json!({ "jsonrpc": "2.0", "id": 1, "method": method, "params": params }) + .to_string() + } + + async fn post(app: &Router, bearer: &str, body: String) -> Response { + app.clone() + .oneshot( + Request::post("/") + .header("host", "mcp.example.com") + .header("authorization", bearer) + .header("content-type", "application/json") + .header("accept", "application/json, text/event-stream") + .body(Body::from(body)) + .unwrap(), + ) + .await + .unwrap() + } + + async fn text(resp: Response) -> String { + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + String::from_utf8(bytes.to_vec()).unwrap() + } + + fn content_type(resp: &Response) -> String { + resp.headers().get(header::CONTENT_TYPE).and_then(|v| v.to_str().ok()).unwrap().to_string() + } + + /// The whole loop over the composed router: the deploy tools are listed, + /// a ticket is minted inside an authenticated call, the archive is PUT on + /// it, and the deploy call — and only it — is answered as an event stream + /// that carries the result. + #[tokio::test(flavor = "multi_thread")] + async fn the_hosted_endpoint_serves_the_deploy_loop() { + let (_dir, server) = server(); + let bearer = format!("Bearer {}", server.store.test_bearer().await); + // Built once, as a deployment does: the upload store lives in it. + let app = server.mcp_router(); + + // tools/list stays plain JSON, and lists the deploy tools. + let resp = post(&app, &bearer, rpc("tools/list", serde_json::json!({}))).await; + let status = resp.status(); + let kind = content_type(&resp); + if status != StatusCode::OK { + panic!("tools/list: {status}: {}", text(resp).await); + } + assert!(kind.starts_with("application/json"), "{kind}"); + let listed: serde_json::Value = serde_json::from_str(&text(resp).await).unwrap(); + let names: Vec<&str> = listed["result"]["tools"] + .as_array() + .unwrap() + .iter() + .map(|t| t["name"].as_str().unwrap()) + .collect(); + for tool in ["open_app", "icp_deploy_bundle", "icp_deploy_upload", "icp_cycles_balance"] { + assert!(names.contains(&tool), "{tool} must be listed: {names:?}"); + } + assert!(!names.contains(&"authenticate"), "no login tool on the hosted server"); + + // A ticket, minted inside the authenticated call. + let resp = post( + &app, + &bearer, + rpc("tools/call", serde_json::json!({ "name": "icp_deploy_upload", "arguments": {} })), + ) + .await; + assert_eq!(resp.status(), StatusCode::OK); + let answer: serde_json::Value = serde_json::from_str(&text(resp).await).unwrap(); + let result = &answer["result"]; + assert_ne!(result["isError"], true, "{result}"); + let ticket = result["structuredContent"]["ticket"].as_str().unwrap().to_string(); + assert_eq!( + result["structuredContent"]["upload_url"], + "https://mcp.example.com/mcp/deploy/upload" + ); + assert!(result["structuredContent"]["curl"].as_str().unwrap().contains(&ticket)); + + // The PUT: refused without a ticket, with a wrong one, and over the cap; + // accepted on the ticket, once. + let put = |ticket: Option<&str>, body: Vec| { + let mut req = Request::put(UPLOAD_PATH); + if let Some(t) = ticket { + req = req.header(TICKET_HEADER, t); + } + app.clone().oneshot(req.body(Body::from(body)).unwrap()) + }; + assert_eq!(put(None, vec![1]).await.unwrap().status(), StatusCode::UNAUTHORIZED); + assert_eq!(put(Some("bogus"), vec![1]).await.unwrap().status(), StatusCode::FORBIDDEN); + let resp = put(Some(&ticket), vec![0; 1025]).await.unwrap(); + assert_eq!(resp.status(), StatusCode::PAYLOAD_TOO_LARGE); + // The over-cap attempt spent the ticket: mint another. + let resp = post( + &app, + &bearer, + rpc("tools/call", serde_json::json!({ "name": "icp_deploy_upload", "arguments": {} })), + ) + .await; + let answer: serde_json::Value = serde_json::from_str(&text(resp).await).unwrap(); + let ticket = answer["result"]["structuredContent"]["ticket"].as_str().unwrap().to_string(); + let resp = put(Some(&ticket), b"not a bundle".to_vec()).await.unwrap(); + assert_eq!(resp.status(), StatusCode::CREATED); + let uploaded: serde_json::Value = serde_json::from_str(&text(resp).await).unwrap(); + let upload_id = uploaded["upload_id"].as_str().unwrap().to_string(); + assert_eq!(uploaded["bytes"], 12); + assert_eq!(put(Some(&ticket), vec![1]).await.unwrap().status(), StatusCode::FORBIDDEN); + + // The deploy call is answered as an event stream carrying the result: + // offline, with a made-up session, the error that the identity cannot + // be derived — raised after the upload was found for this session, + // which is what the next call shows. + let resp = post( + &app, + &bearer, + rpc( + "tools/call", + serde_json::json!({ + "name": "icp_deploy_bundle", + "arguments": { "bundle_upload": upload_id, "environment": "ic" } + }), + ), + ) + .await; + assert_eq!(resp.status(), StatusCode::OK); + assert!(content_type(&resp).starts_with("text/event-stream"), "{}", content_type(&resp)); + let stream = text(resp).await; + let data = stream + .lines() + .filter_map(|l| l.strip_prefix("data:")) + .map(str::trim) + .find(|d| d.contains("\"result\"")) + .expect("the stream carries the result"); + let answer: serde_json::Value = serde_json::from_str(data).unwrap(); + assert_eq!(answer["result"]["isError"], true, "{answer}"); + let said = answer["result"]["content"][0]["text"].as_str().unwrap(); + assert!(!said.contains("no upload"), "the upload must be found: {said}"); + // Spent with the deploy: a second deploy of the same upload finds none. + let resp = post( + &app, + &bearer, + rpc( + "tools/call", + serde_json::json!({ + "name": "icp_deploy_bundle", + "arguments": { "bundle_upload": upload_id, "environment": "ic" } + }), + ), + ) + .await; + let stream = text(resp).await; + assert!(stream.contains("no upload"), "{stream}"); + + // Unauthenticated: the gate answers first, for the deploy call too. + let resp = app + .clone() + .oneshot( + Request::post("/") + .header("host", "mcp.example.com") + .header("content-type", "application/json") + .body(Body::from(rpc( + "tools/call", + serde_json::json!({ "name": "icp_deploy_bundle", "arguments": {} }), + ))) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); + } + + /// Dropping the response body — what a disconnect does — fires the token. + #[tokio::test] + async fn dropping_the_response_body_says_the_client_has_gone() { + let gone = CancellationToken::new(); + let pending = futures_util::stream::pending::>(); + let response = Response::new(Body::from_stream(pending)); + let watched = cancel_on_drop(response, gone.clone()); + assert!(!gone.is_cancelled()); + drop(watched); + assert!(gone.is_cancelled(), "the body's drop must cancel the token"); + } + + #[test] + fn only_a_deploy_call_is_dispatched_to_the_event_stream() { + assert!(is_deploy_call( + rpc("tools/call", serde_json::json!({ "name": "icp_deploy_bundle" })).as_bytes() + )); + assert!(!is_deploy_call( + rpc("tools/call", serde_json::json!({ "name": "icp_deploy_upload" })).as_bytes() + )); + assert!(!is_deploy_call(rpc("tools/list", serde_json::json!({})).as_bytes())); + assert!(!is_deploy_call(b"not json")); + } +} diff --git a/src/lib.rs b/src/lib.rs index 0d72665..599ae78 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -75,6 +75,8 @@ mod auth; mod branding; +#[cfg(feature = "deploy")] +pub mod deploy; // Docs live in the module itself (`//!` in src/metrics.rs): an outer doc // comment here would resolve its intra-doc links in *this* scope rather than // the module's, silently breaking the links to `Registry` and `MatchedPath`. @@ -170,6 +172,10 @@ pub struct McpServer { store: auth::AuthStore, public_url: String, mcp_path: String, + /// The deploy tools, when a deployment turns them on (see + /// [`Self::with_deploy`]). + #[cfg(feature = "deploy")] + deploy: Option, /// Cancels the streamable-HTTP sessions and the session reaper on /// [`Self::shutdown`]. ct: CancellationToken, @@ -205,10 +211,20 @@ impl McpServer { store, public_url, mcp_path, + #[cfg(feature = "deploy")] + deploy: None, ct: CancellationToken::new(), } } + /// Serve the deploy tools on the MCP endpoint (see [`deploy`]): deploy + /// records and uploads go under `{state_dir}/deploys`. + #[cfg(feature = "deploy")] + pub fn with_deploy(mut self, config: deploy::DeployConfig) -> Self { + self.deploy = Some(config); + self + } + /// The operational-files directory this instance uses — where imcp2 creates /// the state a restart must survive (today, the client-registration store). /// Reported straight from the client store, so it is the directory files @@ -263,9 +279,20 @@ impl McpServer { /// won't allow; for a root-mounted instance `.merge` this router instead /// and set `mcp_path: "".into()`. pub fn mcp_router(&self) -> Router { - let mcp_service = { + // Stateless + plain-JSON responses: our tools are pure + // request/response with no server-initiated messages, and this is + // the most compatible mode across MCP clients (ChatGPT's connector + // does not complete the stateful SSE/session handshake the rmcp + // defaults require). The one exception is a deploy call, which + // reports progress before its result (see [`deploy`]). + let transport = StreamableHttpServerConfig::default() + .with_stateful_mode(false) + .with_json_response(true) + .with_cancellation_token(self.ct.child_token()) + .with_allowed_hosts(self.allowed_hosts()); + let plain_mcp = || { let (agent, identities) = (self.agent.clone(), self.identities.clone()); - StreamableHttpService::new( + Router::new().fallback_service(StreamableHttpService::new( move || { Ok(tools::IcTools::new( agent.clone(), @@ -277,18 +304,18 @@ impl McpServer { )) }, LocalSessionManager::default().into(), - // Stateless + plain-JSON responses: our tools are pure - // request/response with no server-initiated messages, and this - // is the most compatible mode across MCP clients (ChatGPT's - // connector does not complete the stateful SSE/session - // handshake the rmcp defaults require). - StreamableHttpServerConfig::default() - .with_stateful_mode(false) - .with_json_response(true) - .with_cancellation_token(self.ct.child_token()) - .with_allowed_hosts(self.allowed_hosts()), - ) + transport.clone(), + )) + }; + // With the deploy tools, the endpoint also has an open route next to + // the gated one: the bundle upload, authenticated by its ticket. + #[cfg(feature = "deploy")] + let (mcp_service, open_routes) = match &self.deploy { + Some(config) => self.deploy_mcp_routers(config, transport.clone()), + None => (plain_mcp(), Router::new()), }; + #[cfg(not(feature = "deploy"))] + let (mcp_service, open_routes) = (plain_mcp(), Router::new()); // The MCP resource is gated by a bearer token issued after Internet // Identity login. Browser-based MCP clients (e.g. the Grok/ChatGPT @@ -317,8 +344,7 @@ impl McpServer { axum::http::header::WWW_AUTHENTICATE, axum::http::HeaderName::from_static("mcp-session-id"), ]); - let gated_mcp = Router::new() - .fallback_service(mcp_service) + let gated_mcp = mcp_service .layer(middleware::from_fn_with_state(self.store.clone(), auth::require_token)) .layer(mcp_cors); @@ -361,6 +387,7 @@ impl McpServer { .nest("/oauth", oauth) .merge(oidc_alternate) .merge(branding) + .merge(open_routes) // Everything else is the MCP endpoint: the bare mount path, its // trailing-slash form, and any sub-path (the streamable service // dispatches on method, not path) — same breadth `nest_service` diff --git a/src/main.rs b/src/main.rs index d2418bc..95012b3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -69,6 +69,27 @@ fn require_resource() -> bool { } } +/// The deploy tools' knobs, when this binary was built with the `deploy` +/// feature: `$MCP_DEPLOY_MAX_CONCURRENT` deploys at once across sessions +/// (default 2) and `$MCP_DEPLOY_MAX_UPLOAD_BYTES` per uploaded archive +/// (default 64 MiB). A value that is not a whole number refuses to start. +#[cfg(feature = "deploy")] +fn deploy_config() -> anyhow::Result { + let mut config = imcp2::deploy::DeployConfig::default(); + if let Ok(v) = std::env::var("MCP_DEPLOY_MAX_CONCURRENT") { + config.concurrent = v + .trim() + .parse() + .map_err(|e| anyhow::anyhow!("MCP_DEPLOY_MAX_CONCURRENT must be a count: {e}"))?; + } + if let Ok(v) = std::env::var("MCP_DEPLOY_MAX_UPLOAD_BYTES") { + config.max_upload_bytes = v.trim().parse().map_err(|e| { + anyhow::anyhow!("MCP_DEPLOY_MAX_UPLOAD_BYTES must be a byte count: {e}") + })?; + } + Ok(config) +} + /// Whether to also serve the beta Internet Identity instance at `/mcp-beta`. /// Off unless `$MCP_SERVE_BETA` is truthy (`1`/`true`/`yes`/`on`), so a /// production deployment serves only `/mcp` (production II) and the staging @@ -309,6 +330,10 @@ async fn main() -> anyhow::Result<()> { // the operational directory. let clients = SharedClients::load(&state_dir); + // The deploy tools, on every instance, when built in. + #[cfg(feature = "deploy")] + let deploy = deploy_config()?; + // Client ID Metadata Documents: on for every instance unless the deploy says no. let cimd_on = cimd_enabled(); if !cimd_on { @@ -328,6 +353,8 @@ async fn main() -> anyhow::Result<()> { require_resource: require_resource(), cimd_enabled: cimd_on, }); + #[cfg(feature = "deploy")] + let prod = prod.with_deploy(deploy.clone()); prod.spawn_session_reaper(); // Beta Internet Identity at `/mcp-beta`: opt in via `$MCP_SERVE_BETA`, so @@ -343,6 +370,8 @@ async fn main() -> anyhow::Result<()> { require_resource: require_resource(), cimd_enabled: cimd_on, }); + #[cfg(feature = "deploy")] + let beta = beta.with_deploy(deploy.clone()); beta.spawn_session_reaper(); Some(beta) } else {