feat(mcp): OAuth per-user connectors (§15) — providers, PKCE copy-paste flow, env credential delivery
- oauth_providers registry table (per-provider client creds) + db/oauth_providers.rs - mcp/oauth.rs: authorization-code + PKCE S256, RAM-only TTL'd flow store, copy-paste consent - mcp/install.rs + verify.rs: connector file install + manifest verification - activate persists a pending row (needs_oauth); /mcp/oauth/start + /complete exchange code for refresh token - credential delivery via env var on docker exec (google_authorized_user JSON), never on disk - mcp_catalog/mcp_user_servers: additive OAuth columns (ensure_column), catalog_name/oauth_provider/deliver_json bare TEXT snapshots - frontend: connector-detail.js (OAuth login panel), shared/connector-common.js, connectors.js admin Sign-in providers modal - API: /mcp/providers (admin OAuth creds), /mcp/oauth/start|complete - .gitignore: add /homes/ (instance data), /connectors/, /reset.sh; drop stale /secrets/
This commit is contained in:
@@ -0,0 +1,181 @@
|
||||
//! On-disk layout of installed connectors (blueprint §7/§14).
|
||||
//!
|
||||
//! One folder per connector, `{WD}/connectors/<name>/`, holding exactly what the
|
||||
//! marketplace served: the runtime files, the icons, and the `connector.json` the
|
||||
//! admin accepted. It sits beside `homes/` and `shared/` because it belongs to the
|
||||
//! **instance**, not to the checkout — `scripts/` was the wrong home for it, being
|
||||
//! a source-tree directory that also carries hand-written dev scripts.
|
||||
//!
|
||||
//! Two consumers, and the split matters:
|
||||
//!
|
||||
//! - A **global** connector runs on the host, straight out of this folder.
|
||||
//! - A **per-user** connector runs inside the user's container, so its runtime
|
||||
//! files are copied into the bind-mounted home ([`install_into_home`]) — the only
|
||||
//! durable zone (§6), so they survive a container recreate.
|
||||
//!
|
||||
//! `connector.json` is written but never read back: [`crate::db::mcp_catalog`] is
|
||||
//! the only thing that drives a connect. The file is provenance — what was accepted,
|
||||
//! and on what day — which is also what makes a later silent upstream change
|
||||
//! detectable. Reading it at runtime would create a second source of truth that
|
||||
//! diverges the moment the admin edits the catalog row.
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use anyhow::{Context, Result, bail};
|
||||
|
||||
use crate::container::{CONTAINER_HOME, HOMES_DIR};
|
||||
|
||||
/// Subdirectory of the working directory holding installed connector folders.
|
||||
pub const CONNECTORS_DIR: &str = "connectors";
|
||||
|
||||
/// The manifest, saved verbatim at install time as provenance (never read back).
|
||||
pub const MANIFEST_FILE: &str = "connector.json";
|
||||
|
||||
/// Where a per-user connector's files land inside the container, under the home
|
||||
/// mount. `{CONTAINER_HOME}/.skald/mcp/<runtime_name>/`.
|
||||
const IN_CONTAINER_MCP_SUBDIR: &str = ".skald/mcp";
|
||||
|
||||
/// The host directory holding `name`'s installed files. Does not check existence —
|
||||
/// callers that need the files present say so themselves, with their own message.
|
||||
pub fn connector_dir(name: &str) -> Result<PathBuf> {
|
||||
let wd = std::env::current_dir().context("failed to read working directory")?;
|
||||
Ok(wd.join(CONNECTORS_DIR).join(name))
|
||||
}
|
||||
|
||||
/// Splits a catalog `script_path` (`<folder>/<rel>`) into the connector folder and
|
||||
/// the entry file's path *inside* it.
|
||||
///
|
||||
/// The tail is kept whole rather than reduced to a basename: a connector may ship a
|
||||
/// tree (`pkg/server.py`), and flattening it would break the import that made it a
|
||||
/// tree in the first place.
|
||||
pub fn split_script_path(script_path: &str) -> Result<(&str, &str)> {
|
||||
match script_path.split_once('/') {
|
||||
Some((folder, rel)) if !folder.is_empty() && !rel.is_empty() => Ok((folder, rel)),
|
||||
_ => bail!("script_path `{script_path}` is not of the form `<connector>/<file>`"),
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether a file is a host-side asset rather than something the runtime needs.
|
||||
///
|
||||
/// Icons are for the browser and the manifest is provenance; neither has any job
|
||||
/// inside a user's container, so they stay out of the home. The rule is extension-
|
||||
/// based because the manifest names icons freely (`icon_sm.png`, `icon_lg.svg`);
|
||||
/// if some future connector ever ships an image it genuinely needs at runtime, this
|
||||
/// is the one place to reconsider.
|
||||
pub fn is_host_asset(rel: &str) -> bool {
|
||||
if rel == MANIFEST_FILE {
|
||||
return true;
|
||||
}
|
||||
let ext = Path::new(rel)
|
||||
.extension()
|
||||
.and_then(|e| e.to_str())
|
||||
.unwrap_or_default()
|
||||
.to_ascii_lowercase();
|
||||
matches!(ext.as_str(), "png" | "svg" | "jpg" | "jpeg" | "webp" | "gif" | "ico")
|
||||
}
|
||||
|
||||
/// The in-container path of a per-user connector's directory.
|
||||
fn container_dir_for(runtime_name: &str) -> PathBuf {
|
||||
Path::new(CONTAINER_HOME).join(IN_CONTAINER_MCP_SUBDIR).join(runtime_name)
|
||||
}
|
||||
|
||||
/// The host path of a per-user connector's directory, inside the bind-mounted home.
|
||||
fn home_dir_for(user_id: &str, runtime_name: &str) -> Result<PathBuf> {
|
||||
let wd = std::env::current_dir().context("failed to read working directory")?;
|
||||
Ok(wd
|
||||
.join(HOMES_DIR)
|
||||
.join(user_id)
|
||||
.join(IN_CONTAINER_MCP_SUBDIR)
|
||||
.join(runtime_name))
|
||||
}
|
||||
|
||||
/// Copies the runtime files of the installed connector `folder` into `user_id`'s
|
||||
/// home under `.skald/mcp/<runtime_name>/`, and returns the directory's path
|
||||
/// **inside** the container.
|
||||
///
|
||||
/// The whole tree is copied, minus host assets ([`is_host_asset`]) — which is what
|
||||
/// finally gets a connector's `requirements.txt` and its multi-file trees into the
|
||||
/// container, where copying a single entry file never did.
|
||||
///
|
||||
/// Returns `Ok(None)` when `folder` was never installed on this box, so a caller
|
||||
/// that does not actually need the files (a catalog entry pointing at nothing, a
|
||||
/// connector with no verify step) can carry on. Idempotent: re-running overwrites.
|
||||
pub fn install_into_home(
|
||||
user_id: &str,
|
||||
runtime_name: &str,
|
||||
folder: &str,
|
||||
) -> Result<Option<PathBuf>> {
|
||||
let src = connector_dir(folder)?;
|
||||
if !src.is_dir() {
|
||||
return Ok(None);
|
||||
}
|
||||
let dest = home_dir_for(user_id, runtime_name)?;
|
||||
std::fs::create_dir_all(&dest)
|
||||
.with_context(|| format!("failed to create {}", dest.display()))?;
|
||||
copy_runtime_files(&src, &dest, Path::new(""))?;
|
||||
Ok(Some(container_dir_for(runtime_name)))
|
||||
}
|
||||
|
||||
/// Recursively copies `src` into `dest`, skipping host assets. `rel` tracks the
|
||||
/// path relative to the connector root so [`is_host_asset`] sees the same string
|
||||
/// the manifest declared.
|
||||
fn copy_runtime_files(src: &Path, dest: &Path, rel: &Path) -> Result<()> {
|
||||
for entry in std::fs::read_dir(src).with_context(|| format!("cannot read {}", src.display()))? {
|
||||
let entry = entry?;
|
||||
let name = entry.file_name();
|
||||
let child_rel = rel.join(&name);
|
||||
let from = entry.path();
|
||||
let to = dest.join(&name);
|
||||
|
||||
if entry.file_type()?.is_dir() {
|
||||
std::fs::create_dir_all(&to)
|
||||
.with_context(|| format!("failed to create {}", to.display()))?;
|
||||
copy_runtime_files(&from, &to, &child_rel)?;
|
||||
continue;
|
||||
}
|
||||
if is_host_asset(&child_rel.to_string_lossy()) {
|
||||
continue;
|
||||
}
|
||||
std::fs::copy(&from, &to)
|
||||
.with_context(|| format!("failed to copy {}", child_rel.display()))?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn splits_a_script_path_into_folder_and_tail() {
|
||||
assert_eq!(split_script_path("gmail/server.py").unwrap(), ("gmail", "server.py"));
|
||||
// A tree keeps its shape — the tail is not reduced to a basename.
|
||||
assert_eq!(
|
||||
split_script_path("whatsapp/pkg/index.js").unwrap(),
|
||||
("whatsapp", "pkg/index.js")
|
||||
);
|
||||
for bad in ["server.py", "", "gmail/", "/server.py"] {
|
||||
assert!(split_script_path(bad).is_err(), "should have rejected `{bad}`");
|
||||
}
|
||||
}
|
||||
|
||||
/// Icons and the manifest are host-side only: they must never reach a user's
|
||||
/// container, while everything the server actually runs on must.
|
||||
#[test]
|
||||
fn host_assets_are_icons_and_the_manifest() {
|
||||
for asset in ["connector.json", "icon_sm.png", "icon_lg.svg", "a/b/logo.WEBP"] {
|
||||
assert!(is_host_asset(asset), "`{asset}` should be a host asset");
|
||||
}
|
||||
for runtime in ["server.py", "requirements.txt", "pkg/index.js", "verify.py"] {
|
||||
assert!(!is_host_asset(runtime), "`{runtime}` should reach the container");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn container_dir_hangs_off_the_home_mount() {
|
||||
assert_eq!(
|
||||
container_dir_for("gmail"),
|
||||
PathBuf::from("/root/.skald/mcp/gmail")
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -24,10 +24,16 @@ pub use mcp_client::{
|
||||
|
||||
use mcp_client::McpTransport;
|
||||
|
||||
pub mod install;
|
||||
mod logs;
|
||||
pub mod oauth;
|
||||
mod provider;
|
||||
pub mod verify;
|
||||
|
||||
pub use install::{CONNECTORS_DIR, MANIFEST_FILE, connector_dir, install_into_home, split_script_path};
|
||||
pub use oauth::DeliverSpec;
|
||||
pub use provider::{McpProvider, UserMcpView};
|
||||
pub use verify::{VerifyReport, VerifyTarget, apply_placeholders, run_verify};
|
||||
|
||||
const SERVER_START_TIMEOUT_SECS: u64 = 120;
|
||||
|
||||
@@ -409,10 +415,35 @@ fn apply_key_placeholder(
|
||||
) -> (Option<String>, Option<String>) {
|
||||
match (url, api_key) {
|
||||
(Some(u), Some(k)) if u.contains("{key}") => (Some(u.replace("{key}", &k)), None),
|
||||
// Unified {SECRET:<param>} placeholder (e.g. Tavily's
|
||||
// `?tavilyApiKey={SECRET:tavilyApiKey}`). Any SECRET token in a URL is
|
||||
// the api_key for a remote connector — a URL never carries the user's
|
||||
// other secrets — so we substitute every occurrence.
|
||||
(Some(u), Some(k)) if u.contains("{SECRET:") => (Some(substitute_secret_tokens(&u, &k)), None),
|
||||
(u, k) => (u, k),
|
||||
}
|
||||
}
|
||||
|
||||
/// Replaces every `{SECRET:…}` token in `text` with `value`. Used for the
|
||||
/// api-key-in-URL case; other placeholders are left untouched.
|
||||
fn substitute_secret_tokens(text: &str, value: &str) -> String {
|
||||
let mut out = String::with_capacity(text.len());
|
||||
let mut rest = text;
|
||||
while let Some(open) = rest.find("{SECRET:") {
|
||||
out.push_str(&rest[..open]);
|
||||
let after = &rest[open..];
|
||||
if let Some(close) = after.find('}') {
|
||||
out.push_str(value);
|
||||
rest = &after[close + 1..];
|
||||
} else {
|
||||
out.push_str(after);
|
||||
break;
|
||||
}
|
||||
}
|
||||
out.push_str(rest);
|
||||
out
|
||||
}
|
||||
|
||||
/// Builds a spec for a globally-active connector — host transport (`launch_in`
|
||||
/// = None), so it runs in the Skald process, not in any container (§7).
|
||||
pub fn global_row_spec(row: &crate::db::mcp_global_servers::McpGlobalServerRow) -> McpServerSpec {
|
||||
@@ -460,6 +491,51 @@ pub fn user_row_spec(
|
||||
}
|
||||
}
|
||||
|
||||
/// Like [`user_row_spec`], but for an OAuth connector it also resolves the stored
|
||||
/// refresh token into the credential the server reads and injects it into the
|
||||
/// process env per the delivery spec (§15). Non-OAuth rows are returned unchanged.
|
||||
///
|
||||
/// `registry` is the system pool, where `oauth_providers` (the client credentials)
|
||||
/// lives. A resolution failure is logged, not fatal: the server still starts, and
|
||||
/// fails its own auth visibly, rather than the whole login batch aborting.
|
||||
pub async fn user_row_spec_resolved(
|
||||
row: &crate::db::mcp_user_servers::McpUserServerRow,
|
||||
container: &str,
|
||||
registry: &SqlitePool,
|
||||
) -> McpServerSpec {
|
||||
let mut spec = user_row_spec(row, container);
|
||||
if let (Some(provider), Some(deliver), Some(refresh)) =
|
||||
(row.oauth_provider.as_deref(), row.deliver(), row.api_key.as_deref())
|
||||
{
|
||||
if let Err(e) = inject_oauth_env(&mut spec, provider, &deliver, refresh, registry).await {
|
||||
warn!("connector '{}': OAuth credential delivery failed: {e}", row.name);
|
||||
}
|
||||
}
|
||||
spec
|
||||
}
|
||||
|
||||
/// Assembles the credential from the provider's client creds + the refresh token and
|
||||
/// sets it on `spec.config.env` under the delivery spec's env name.
|
||||
async fn inject_oauth_env(
|
||||
spec: &mut McpServerSpec,
|
||||
provider_name: &str,
|
||||
deliver: &DeliverSpec,
|
||||
refresh_token: &str,
|
||||
registry: &SqlitePool,
|
||||
) -> Result<()> {
|
||||
if deliver.as_ != "env" {
|
||||
anyhow::bail!("only `env` credential delivery is wired (deliver.as = `{}`)", deliver.as_);
|
||||
}
|
||||
let env_name = deliver.env.as_deref()
|
||||
.ok_or_else(|| anyhow::anyhow!("deliver.as=env but no deliver.env name"))?;
|
||||
let format = deliver.format.as_deref().unwrap_or("google_authorized_user");
|
||||
let provider = crate::db::oauth_providers::get(registry, provider_name).await?
|
||||
.ok_or_else(|| anyhow::anyhow!("unknown OAuth provider `{provider_name}`"))?;
|
||||
let cred = oauth::assemble_credential(format, &provider, refresh_token)?;
|
||||
spec.config.env.get_or_insert_with(HashMap::new).insert(env_name.to_string(), cred);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Generates a 32-char alphanumeric id for a persisted media filename
|
||||
/// (mirrors `ImageGeneratorManager`).
|
||||
fn random_id() -> String {
|
||||
@@ -510,3 +586,50 @@ pub fn content_type_for_ext(ext: &str) -> &'static str {
|
||||
_ => "application/octet-stream",
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn key_placeholder_legacy_is_substituted() {
|
||||
let (url, key) = apply_key_placeholder(
|
||||
Some("https://x/?k={key}".into()),
|
||||
Some("secret123".into()),
|
||||
);
|
||||
assert_eq!(url.as_deref(), Some("https://x/?k=secret123"));
|
||||
assert!(key.is_none(), "api_key is consumed after substitution");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn key_placeholder_secret_token_is_substituted() {
|
||||
// Tavily's unified form: the URL carries {SECRET:tavilyApiKey}.
|
||||
let (url, key) = apply_key_placeholder(
|
||||
Some("https://mcp.tavily.com/mcp/?tavilyApiKey={SECRET:tavilyApiKey}".into()),
|
||||
Some("tvly-abc".into()),
|
||||
);
|
||||
assert_eq!(url.as_deref(), Some("https://mcp.tavily.com/mcp/?tavilyApiKey=tvly-abc"));
|
||||
assert!(key.is_none(), "api_key is consumed when the URL had a SECRET token");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn key_placeholder_no_token_keeps_key_for_bearer() {
|
||||
// No placeholder in the URL → the key stays, so the HTTP transport
|
||||
// sends it as `Authorization: Bearer`.
|
||||
let (url, key) = apply_key_placeholder(
|
||||
Some("https://x.example.com/mcp".into()),
|
||||
Some("bearer-key".into()),
|
||||
);
|
||||
assert_eq!(url.as_deref(), Some("https://x.example.com/mcp"));
|
||||
assert_eq!(key.as_deref(), Some("bearer-key"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn substitute_secret_tokens_replaces_every_occurrence() {
|
||||
let s = substitute_secret_tokens(
|
||||
"a={SECRET:K}&b={SECRET:K}&c={ENV:C}",
|
||||
"VAL",
|
||||
);
|
||||
assert_eq!(s, "a=VAL&b=VAL&c={ENV:C}");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,200 @@
|
||||
//! OAuth 2.0 authorization-code + PKCE for per-user connectors (blueprint §15).
|
||||
//!
|
||||
//! The consent step is a **human copy-paste**, not a headless action (§15): Skald
|
||||
//! builds a consent URL, the user approves it in a browser, and the provider lands
|
||||
//! the `code` on a static page (`redirect_uri`, e.g. `oauth/show.html`) that shows
|
||||
//! it for copying. Skald then exchanges the code for a refresh token. PKCE means an
|
||||
//! intercepted code is useless without the verifier, which never leaves this
|
||||
//! process — so the copy-paste page can be a plain static file with no backend.
|
||||
//!
|
||||
//! The obtained refresh token is delivered to the connector's server per its
|
||||
//! manifest `auth.deliver` spec; only `env` delivery is wired (the credential is
|
||||
//! injected as an environment variable at `docker exec` time — nothing on disk).
|
||||
|
||||
use anyhow::{Context, Result, bail};
|
||||
use base64::Engine;
|
||||
use rand::Rng as _;
|
||||
use serde::Deserialize;
|
||||
use sha2::{Digest, Sha256};
|
||||
|
||||
use crate::db::oauth_providers::OauthProviderRow;
|
||||
|
||||
/// How Skald delivers the obtained credential to the connector's server process,
|
||||
/// mirrored from the manifest's `auth.deliver` (§15).
|
||||
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
||||
pub struct DeliverSpec {
|
||||
/// `env` | `file`. Only `env` is implemented; a `file` target is rejected at
|
||||
/// activation with a clear message rather than silently half-working.
|
||||
#[serde(rename = "as")]
|
||||
pub as_: String,
|
||||
/// The serialization Skald must produce (`google_authorized_user` | `refresh_token`).
|
||||
#[serde(default)]
|
||||
pub format: Option<String>,
|
||||
/// `as=env`: the environment variable the credential is injected into.
|
||||
#[serde(default)]
|
||||
pub env: Option<String>,
|
||||
/// `as=file`: the target path (unused while file delivery is unimplemented).
|
||||
#[serde(default)]
|
||||
pub path: Option<String>,
|
||||
}
|
||||
|
||||
const URL_SAFE: base64::engine::general_purpose::GeneralPurpose =
|
||||
base64::engine::general_purpose::URL_SAFE_NO_PAD;
|
||||
|
||||
/// A high-entropy PKCE verifier and its S256 challenge (RFC 7636).
|
||||
pub struct Pkce {
|
||||
pub verifier: String,
|
||||
pub challenge: String,
|
||||
}
|
||||
|
||||
pub fn generate_pkce() -> Pkce {
|
||||
let mut bytes = [0u8; 32];
|
||||
rand::rng().fill_bytes(&mut bytes);
|
||||
let verifier = URL_SAFE.encode(bytes); // 43-char base64url, within the RFC range
|
||||
let challenge = URL_SAFE.encode(Sha256::digest(verifier.as_bytes()));
|
||||
Pkce { verifier, challenge }
|
||||
}
|
||||
|
||||
/// An opaque, URL-safe `state` value: CSRF guard and the key of the pending flow.
|
||||
pub fn random_state() -> String {
|
||||
let mut bytes = [0u8; 24];
|
||||
rand::rng().fill_bytes(&mut bytes);
|
||||
URL_SAFE.encode(bytes)
|
||||
}
|
||||
|
||||
/// Builds the authorization-endpoint URL the user opens to consent. Merges the
|
||||
/// provider's `extra_params` (Google needs `access_type=offline` + `prompt=consent`
|
||||
/// to return a refresh token) after the standard params.
|
||||
pub fn build_consent_url(
|
||||
provider: &OauthProviderRow,
|
||||
scopes: &[String],
|
||||
state: &str,
|
||||
challenge: &str,
|
||||
) -> Result<String> {
|
||||
let scope = scopes.join(" ");
|
||||
let mut params: Vec<(String, String)> = vec![
|
||||
("client_id".into(), provider.client_id.clone()),
|
||||
("redirect_uri".into(), provider.redirect_uri.clone()),
|
||||
("response_type".into(), "code".into()),
|
||||
("scope".into(), scope),
|
||||
("state".into(), state.into()),
|
||||
("code_challenge".into(), challenge.into()),
|
||||
("code_challenge_method".into(), "S256".into()),
|
||||
];
|
||||
for (k, v) in provider.extra() {
|
||||
params.push((k, v));
|
||||
}
|
||||
let url = reqwest::Url::parse_with_params(&provider.auth_url, ¶ms)
|
||||
.with_context(|| format!("invalid authorization endpoint `{}`", provider.auth_url))?;
|
||||
Ok(url.to_string())
|
||||
}
|
||||
|
||||
/// The token endpoint's response. Google returns `refresh_token` only on the first
|
||||
/// consent for a client, or when `prompt=consent` forces re-issue — hence the
|
||||
/// provider's `extra_params`.
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub struct TokenResponse {
|
||||
#[serde(default)] pub access_token: Option<String>,
|
||||
#[serde(default)] pub refresh_token: Option<String>,
|
||||
#[serde(default)] pub expires_in: Option<i64>,
|
||||
#[serde(default)] pub scope: Option<String>,
|
||||
#[serde(default)] pub error: Option<String>,
|
||||
#[serde(default)] pub error_description: Option<String>,
|
||||
}
|
||||
|
||||
/// Exchanges an authorization `code` (+ PKCE `verifier`) for tokens at the
|
||||
/// provider's token endpoint.
|
||||
pub async fn exchange_code(
|
||||
provider: &OauthProviderRow,
|
||||
code: &str,
|
||||
verifier: &str,
|
||||
) -> Result<TokenResponse> {
|
||||
let params = [
|
||||
("grant_type", "authorization_code"),
|
||||
("code", code),
|
||||
("client_id", provider.client_id.as_str()),
|
||||
("client_secret", provider.client_secret.as_str()),
|
||||
("redirect_uri", provider.redirect_uri.as_str()),
|
||||
("code_verifier", verifier),
|
||||
];
|
||||
// `RequestBuilder::form` needs reqwest's `urlencoded` feature, which this build
|
||||
// doesn't enable — so encode the body ourselves. Parsing a throwaway URL with
|
||||
// these params yields exactly the `application/x-www-form-urlencoded` string.
|
||||
let body = reqwest::Url::parse_with_params("http://form.local/", ¶ms)
|
||||
.ok()
|
||||
.and_then(|u| u.query().map(str::to_owned))
|
||||
.unwrap_or_default();
|
||||
let resp = reqwest::Client::new()
|
||||
.post(&provider.token_url)
|
||||
.header(reqwest::header::CONTENT_TYPE, "application/x-www-form-urlencoded")
|
||||
.body(body)
|
||||
.send()
|
||||
.await
|
||||
.context("token endpoint request failed")?;
|
||||
let status = resp.status();
|
||||
let body: TokenResponse = resp
|
||||
.json()
|
||||
.await
|
||||
.context("token endpoint returned a non-JSON body")?;
|
||||
if let Some(err) = &body.error {
|
||||
let detail = body.error_description.as_deref()
|
||||
.map(|d| format!(" — {d}")).unwrap_or_default();
|
||||
bail!("token exchange failed: {err}{detail}");
|
||||
}
|
||||
if !status.is_success() {
|
||||
bail!("token exchange failed with HTTP {status}");
|
||||
}
|
||||
Ok(body)
|
||||
}
|
||||
|
||||
/// Serializes a refresh token into the shape the connector's server reads, per
|
||||
/// `deliver.format`. `google_authorized_user` is the JSON that
|
||||
/// `google.oauth2.credentials.Credentials.from_authorized_user_info` accepts — the
|
||||
/// server refreshes access tokens from it on its own.
|
||||
pub fn assemble_credential(
|
||||
format: &str,
|
||||
provider: &OauthProviderRow,
|
||||
refresh_token: &str,
|
||||
) -> Result<String> {
|
||||
match format {
|
||||
"google_authorized_user" => Ok(serde_json::json!({
|
||||
"type": "authorized_user",
|
||||
"client_id": provider.client_id,
|
||||
"client_secret": provider.client_secret,
|
||||
"refresh_token": refresh_token,
|
||||
"token_uri": provider.token_url,
|
||||
}).to_string()),
|
||||
"refresh_token" => Ok(refresh_token.to_string()),
|
||||
other => bail!("unsupported deliver.format `{other}`"),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn pkce_challenge_is_s256_of_verifier() {
|
||||
let p = generate_pkce();
|
||||
let expect = URL_SAFE.encode(Sha256::digest(p.verifier.as_bytes()));
|
||||
assert_eq!(p.challenge, expect);
|
||||
assert!(!p.verifier.contains(['+', '/', '=']), "verifier must be url-safe, unpadded");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn authorized_user_credential_has_googles_fields() {
|
||||
let provider = OauthProviderRow {
|
||||
name: "google".into(), display_name: "Google".into(),
|
||||
auth_url: "https://a".into(), token_url: "https://t".into(),
|
||||
client_id: "cid".into(), client_secret: "csec".into(),
|
||||
redirect_uri: "https://r".into(), extra_params: None,
|
||||
created_at: String::new(), updated_at: String::new(),
|
||||
};
|
||||
let cred = assemble_credential("google_authorized_user", &provider, "rt-123").unwrap();
|
||||
let v: serde_json::Value = serde_json::from_str(&cred).unwrap();
|
||||
assert_eq!(v["type"], "authorized_user");
|
||||
assert_eq!(v["client_id"], "cid");
|
||||
assert_eq!(v["refresh_token"], "rt-123");
|
||||
assert_eq!(v["token_uri"], "https://t");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,417 @@
|
||||
//! Verify-before-save for MCP connectors (blueprint §15 verify step).
|
||||
//!
|
||||
//! When a user fills the activation form, Skald can run the connector's declared
|
||||
//! `verify` command to confirm the credentials actually work *before* persisting
|
||||
//! the activation. This module owns:
|
||||
//!
|
||||
//! - [`apply_placeholders`] — the single substitution engine for `{ENV:NAME}`
|
||||
//! and `{SECRET:NAME}` tokens (used here for the verify command, and by the
|
||||
//! MCP transport for URLs / env values).
|
||||
//! - [`run_verify`] — launches the resolved command either on the host (for a
|
||||
//! global `mcp_remote` connector) or inside the caller's container (for a
|
||||
//! per-user `mcp_local` connector), parses the JSON result, and returns a
|
||||
//! [`VerifyReport`].
|
||||
//!
|
||||
//! Output contract: the verify command must print one JSON object on stdout,
|
||||
//! `{"ok": bool, "message": string, "details"?: object}`, and exit 0 on success.
|
||||
//! If the JSON parse fails, [`run_verify`] falls back to the exit code. Secrets
|
||||
//! are never logged.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::path::Path;
|
||||
use std::process::Stdio;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use serde::Serialize;
|
||||
use serde_json::Value;
|
||||
use tokio::io::AsyncReadExt;
|
||||
use tracing::debug;
|
||||
|
||||
/// Default timeout for a verify command (seconds). Overridable per-connector via
|
||||
/// the manifest's `verify.timeout_secs`.
|
||||
pub const DEFAULT_VERIFY_TIMEOUT_SECS: u64 = 15;
|
||||
|
||||
/// The outcome of a verify run, surfaced to the UI verbatim.
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct VerifyReport {
|
||||
/// `true` when the credentials check out.
|
||||
pub ok: bool,
|
||||
/// Human-readable result line (shown next to the Test button).
|
||||
pub message: String,
|
||||
/// Optional structured details (shown in a `<pre>` block). Never holds
|
||||
/// secrets — the verify script is responsible for not echoing them.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub details: Option<Value>,
|
||||
/// Wall-clock time the command took.
|
||||
#[serde(skip)]
|
||||
pub elapsed: Duration,
|
||||
/// `true` when the connector declares no `verify` step, so "no test" must
|
||||
/// be distinguishable from "test passed" in the UI.
|
||||
#[serde(skip)]
|
||||
pub skipped: bool,
|
||||
}
|
||||
|
||||
impl VerifyReport {
|
||||
/// Synthesized when the connector has no `verify` step — the UI shows
|
||||
/// "no test available" rather than a pass/fail.
|
||||
pub fn skipped() -> Self {
|
||||
Self {
|
||||
ok: true,
|
||||
message: "This connector has no verification step.".into(),
|
||||
details: None,
|
||||
elapsed: Duration::ZERO,
|
||||
skipped: true,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Where [`run_verify`] executes the command. Mirrors `McpServerSpec.launch_in`:
|
||||
/// `None` runs on the host (a global `mcp_remote` connector), `Some(container)`
|
||||
/// runs inside the user's container via `docker exec`.
|
||||
pub enum VerifyTarget<'a> {
|
||||
/// Run on the Skald host process. `workdir` is an absolute host path
|
||||
/// (typically `<data_root>/scripts/<id>/`).
|
||||
Host { workdir: &'a Path },
|
||||
/// Run inside the user's sandbox container. `workdir` is an absolute path
|
||||
/// *inside* the container (e.g. `/root/.skald/mcp/<name>`).
|
||||
Container {
|
||||
container: &'a str,
|
||||
workdir: &'a Path,
|
||||
},
|
||||
}
|
||||
|
||||
/// Substitutes `{ENV:NAME}` and `{SECRET:NAME}` tokens in `text`.
|
||||
///
|
||||
/// - `{ENV:NAME}` → `env[NAME]`, or empty string if absent.
|
||||
/// - `{SECRET:NAME}` → `secret[NAME]`, or empty string if absent.
|
||||
/// - Any other `{...}` token is left untouched — `{key}` belongs to the remote
|
||||
/// transport's URL substitution (see `mcp::apply_key`), and anything else is a
|
||||
/// misconfiguration that should stay visible rather than be silently erased.
|
||||
pub fn apply_placeholders(
|
||||
text: &str,
|
||||
env: &HashMap<String, String>,
|
||||
secret: &HashMap<String, String>,
|
||||
) -> String {
|
||||
let mut out = String::with_capacity(text.len());
|
||||
let mut rest = text;
|
||||
while let Some(open) = rest.find('{') {
|
||||
// Append everything up to the '{'.
|
||||
out.push_str(&rest[..open]);
|
||||
let after = &rest[open..];
|
||||
if let Some(close) = after.find('}') {
|
||||
let token = &after[..=close]; // includes both braces
|
||||
let inner = token.strip_prefix('{').unwrap().strip_suffix('}').unwrap();
|
||||
// ENV:/SECRET: tokens are always consumed (missing key → empty);
|
||||
// any other `{...}` is left untouched so a misconfiguration stays
|
||||
// visible rather than being silently erased.
|
||||
if let Some(name) = inner.strip_prefix("ENV:") {
|
||||
out.push_str(env.get(name).map(|s| s.as_str()).unwrap_or(""));
|
||||
} else if let Some(name) = inner.strip_prefix("SECRET:") {
|
||||
out.push_str(secret.get(name).map(|s| s.as_str()).unwrap_or(""));
|
||||
} else {
|
||||
out.push_str(token);
|
||||
}
|
||||
rest = &after[close + 1..];
|
||||
} else {
|
||||
// No closing brace — emit the rest literally and stop.
|
||||
out.push_str(after);
|
||||
return out;
|
||||
}
|
||||
}
|
||||
out.push_str(rest);
|
||||
out
|
||||
}
|
||||
|
||||
/// Runs the verify `command` (after placeholder substitution) in the given
|
||||
/// target, injects the env/secret values as environment variables, captures
|
||||
/// stdout/stderr under a timeout, and parses the JSON result.
|
||||
///
|
||||
/// The command and resolved env are NOT logged (secrets may be inline). Only
|
||||
/// the final `ok`/`message` are traced at debug level.
|
||||
pub async fn run_verify(
|
||||
command: &str,
|
||||
env_values: &HashMap<String, String>,
|
||||
secret_values: &HashMap<String, String>,
|
||||
target: VerifyTarget<'_>,
|
||||
timeout_secs: u64,
|
||||
) -> VerifyReport {
|
||||
let resolved = apply_placeholders(command, env_values, secret_values);
|
||||
let timeout = Duration::from_secs(timeout_secs.max(1));
|
||||
let started = Instant::now();
|
||||
|
||||
// Build the process: `docker exec … sh -c "<cmd>"` or host `sh -c "<cmd>"`.
|
||||
let mut cmd = match target {
|
||||
VerifyTarget::Container { container, workdir } => {
|
||||
let mut c = tokio::process::Command::new("docker");
|
||||
c.arg("exec")
|
||||
.arg("-w").arg(workdir)
|
||||
.arg(container);
|
||||
inject_env_flags(&mut c, env_values, secret_values);
|
||||
c.arg("sh").arg("-c").arg(&resolved);
|
||||
c
|
||||
}
|
||||
VerifyTarget::Host { workdir } => {
|
||||
let mut c = tokio::process::Command::new("sh");
|
||||
c.arg("-c").arg(&resolved).current_dir(workdir);
|
||||
inject_env_vars(&mut c, env_values, secret_values);
|
||||
c
|
||||
}
|
||||
};
|
||||
cmd.stdout(Stdio::piped())
|
||||
.stderr(Stdio::piped())
|
||||
.stdin(Stdio::null())
|
||||
.kill_on_drop(true);
|
||||
|
||||
let child = match cmd.spawn() {
|
||||
Ok(c) => c,
|
||||
Err(e) => {
|
||||
return VerifyReport {
|
||||
ok: false,
|
||||
message: format!("Could not start the verify command: {e}"),
|
||||
details: None,
|
||||
elapsed: started.elapsed(),
|
||||
skipped: false,
|
||||
};
|
||||
}
|
||||
};
|
||||
|
||||
let outcome = run_with_timeout(child, timeout).await;
|
||||
let elapsed = started.elapsed();
|
||||
|
||||
let report = parse_verify_output(&outcome, elapsed);
|
||||
debug!(ok = report.ok, elapsed_ms = elapsed.as_millis() as u64, "verify");
|
||||
report
|
||||
}
|
||||
|
||||
/// Collects the child's stdout/stderr under a single timeout, returning the
|
||||
/// captured buffers and the exit code (None if killed by timeout).
|
||||
async fn run_with_timeout(
|
||||
mut child: tokio::process::Child,
|
||||
timeout: Duration,
|
||||
) -> VerifyOutcome {
|
||||
let mut stdout = child.stdout.take().expect("stdout piped");
|
||||
let mut stderr = child.stderr.take().expect("stderr piped");
|
||||
|
||||
let collect = async {
|
||||
let mut out = Vec::new();
|
||||
let mut err = Vec::new();
|
||||
// Read concurrently — the pipes are independent.
|
||||
let r1 = stdout.read_to_end(&mut out);
|
||||
let r2 = stderr.read_to_end(&mut err);
|
||||
let (ro, re, status) = tokio::join!(r1, r2, child.wait());
|
||||
ro.map_err(anyhow::Error::from)?;
|
||||
re.map_err(anyhow::Error::from)?;
|
||||
let code = status.ok().and_then(|s| s.code());
|
||||
Ok::<_, anyhow::Error>((out, err, code))
|
||||
};
|
||||
|
||||
match tokio::time::timeout(timeout, collect).await {
|
||||
Ok(Ok((out, err, code))) => VerifyOutcome { stdout: out, stderr: err, code, timed_out: false },
|
||||
// Inner error (spawn/io).
|
||||
Ok(Err(e)) => VerifyOutcome {
|
||||
stdout: Vec::new(),
|
||||
stderr: e.to_string().into_bytes(),
|
||||
code: None,
|
||||
timed_out: false,
|
||||
},
|
||||
// Timeout: kill_on_drop takes care of the child.
|
||||
Err(_) => VerifyOutcome {
|
||||
stdout: Vec::new(),
|
||||
stderr: format!("verify timed out after {}s", timeout.as_secs()).into_bytes(),
|
||||
code: None,
|
||||
timed_out: true,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
struct VerifyOutcome {
|
||||
stdout: Vec<u8>,
|
||||
stderr: Vec<u8>,
|
||||
code: Option<i32>,
|
||||
timed_out: bool,
|
||||
}
|
||||
|
||||
/// Parses the verify command's output into a [`VerifyReport`].
|
||||
///
|
||||
/// Contract: the command prints one JSON object on stdout:
|
||||
/// `{"ok": bool, "message": string, "details"?: object}`. If the parse fails,
|
||||
/// falls back to the exit code (0 = ok, anything else = fail) and uses stderr
|
||||
/// (or stdout) as the message.
|
||||
fn parse_verify_output(outcome: &VerifyOutcome, elapsed: Duration) -> VerifyReport {
|
||||
let stdout = String::from_utf8_lossy(&outcome.stdout);
|
||||
let stderr = String::from_utf8_lossy(&outcome.stderr);
|
||||
|
||||
if outcome.timed_out {
|
||||
return VerifyReport {
|
||||
ok: false,
|
||||
message: stderr.trim().to_string(),
|
||||
details: None,
|
||||
elapsed,
|
||||
skipped: false,
|
||||
};
|
||||
}
|
||||
|
||||
// Try JSON parse first (prefer the last line, in case the script emitted a
|
||||
// trailing newline or a preamble).
|
||||
let trimmed = stdout.trim();
|
||||
if !trimmed.is_empty() {
|
||||
if let Ok(v) = serde_json::from_str::<Value>(trimmed) {
|
||||
let ok = v.get("ok").and_then(|o| o.as_bool()).unwrap_or_else(|| outcome.code == Some(0));
|
||||
let message = v
|
||||
.get("message")
|
||||
.and_then(|m| m.as_str())
|
||||
.unwrap_or("")
|
||||
.to_string();
|
||||
let details = v.get("details").cloned();
|
||||
return VerifyReport { ok, message, details, elapsed, skipped: false };
|
||||
}
|
||||
}
|
||||
|
||||
// Fallback: exit-code semantics. Empty stdout → fall back to stderr.
|
||||
let ok = outcome.code == Some(0);
|
||||
let message = if !trimmed.is_empty() {
|
||||
trimmed.to_string()
|
||||
} else if !stderr.trim().is_empty() {
|
||||
stderr.trim().to_string()
|
||||
} else if ok {
|
||||
"Verification succeeded.".into()
|
||||
} else {
|
||||
format!("Verify failed (exit code {}).", outcome.code.unwrap_or(-1))
|
||||
};
|
||||
VerifyReport { ok, message, details: None, elapsed, skipped: false }
|
||||
}
|
||||
|
||||
/// Adds `-e KEY=VALUE` flags for `docker exec`, for both env and secret values.
|
||||
fn inject_env_flags(
|
||||
cmd: &mut tokio::process::Command,
|
||||
env: &HashMap<String, String>,
|
||||
secret: &HashMap<String, String>,
|
||||
) {
|
||||
for (k, v) in env.iter().chain(secret.iter()) {
|
||||
cmd.arg("-e").arg(format!("{k}={v}"));
|
||||
}
|
||||
}
|
||||
|
||||
/// Sets environment variables for a host `sh -c` process.
|
||||
fn inject_env_vars(
|
||||
cmd: &mut tokio::process::Command,
|
||||
env: &HashMap<String, String>,
|
||||
secret: &HashMap<String, String>,
|
||||
) {
|
||||
for (k, v) in env.iter().chain(secret.iter()) {
|
||||
cmd.env(k, v);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn m(items: &[(&str, &str)]) -> HashMap<String, String> {
|
||||
items.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn placeholders_env_and_secret() {
|
||||
let env = m(&[("HOST", "imap.example.com"), ("PORT", "993")]);
|
||||
let secret = m(&[("PASS", "hunter2")]);
|
||||
let s = apply_placeholders("h={ENV:HOST} p={ENV:PORT} s={SECRET:PASS}", &env, &secret);
|
||||
assert_eq!(s, "h=imap.example.com p=993 s=hunter2");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn placeholders_missing_become_empty() {
|
||||
let env = m(&[("HOST", "x")]);
|
||||
let secret = HashMap::new();
|
||||
let s = apply_placeholders("[{ENV:HOST}][{ENV:MISSING}][{SECRET:X}]", &env, &secret);
|
||||
assert_eq!(s, "[x][][]");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn placeholders_unknown_left_untouched() {
|
||||
let env = HashMap::new();
|
||||
let secret = HashMap::new();
|
||||
let s = apply_placeholders("{key} {ENV:A} {0}", &env, &secret);
|
||||
assert_eq!(s, "{key} {0}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn placeholders_no_braces() {
|
||||
let env = HashMap::new();
|
||||
let secret = HashMap::new();
|
||||
assert_eq!(apply_placeholders("plain text", &env, &secret), "plain text");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn placeholders_unclosed_brace_kept() {
|
||||
let env = HashMap::new();
|
||||
let secret = HashMap::new();
|
||||
assert_eq!(apply_placeholders("a {ENV:B c", &env, &secret), "a {ENV:B c");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_json_ok() {
|
||||
let o = VerifyOutcome {
|
||||
stdout: br#"{"ok": true, "message": "all good"}"#.to_vec(),
|
||||
stderr: vec![],
|
||||
code: Some(0),
|
||||
timed_out: false,
|
||||
};
|
||||
let r = parse_verify_output(&o, Duration::from_millis(10));
|
||||
assert!(r.ok);
|
||||
assert_eq!(r.message, "all good");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_json_fail_with_details() {
|
||||
let o = VerifyOutcome {
|
||||
stdout: br#"{"ok": false, "message": "bad creds", "details": {"imap": "ok", "smtp": "no"}}"#.to_vec(),
|
||||
stderr: vec![],
|
||||
code: Some(1),
|
||||
timed_out: false,
|
||||
};
|
||||
let r = parse_verify_output(&o, Duration::from_millis(10));
|
||||
assert!(!r.ok);
|
||||
assert_eq!(r.message, "bad creds");
|
||||
assert_eq!(r.details.unwrap()["smtp"], "no");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_fallback_exit_code() {
|
||||
let o = VerifyOutcome {
|
||||
stdout: b"some plain output".to_vec(),
|
||||
stderr: vec![],
|
||||
code: Some(0),
|
||||
timed_out: false,
|
||||
};
|
||||
let r = parse_verify_output(&o, Duration::from_millis(10));
|
||||
assert!(r.ok);
|
||||
assert_eq!(r.message, "some plain output");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_fallback_stderr_on_fail() {
|
||||
let o = VerifyOutcome {
|
||||
stdout: vec![],
|
||||
stderr: b"connection refused".to_vec(),
|
||||
code: Some(2),
|
||||
timed_out: false,
|
||||
};
|
||||
let r = parse_verify_output(&o, Duration::from_millis(10));
|
||||
assert!(!r.ok);
|
||||
assert_eq!(r.message, "connection refused");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_timeout_is_fail() {
|
||||
let o = VerifyOutcome {
|
||||
stdout: vec![],
|
||||
stderr: b"verify timed out after 15s".to_vec(),
|
||||
code: None,
|
||||
timed_out: true,
|
||||
};
|
||||
let r = parse_verify_output(&o, Duration::from_secs(15));
|
||||
assert!(!r.ok);
|
||||
assert!(r.message.contains("timed out"));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user