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 {