diff --git a/Cargo.lock b/Cargo.lock index 49b5182..3470483 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -111,6 +111,15 @@ version = "2.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c4512299f36f043ab09a583e57bceb5a5aab7a73db1805848e8fef3c9e8c78b3" +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + [[package]] name = "bumpalo" version = "3.20.2" @@ -206,16 +215,22 @@ dependencies = [ "chrono", "clap", "jmespath", + "keyring", + "open", "pretty_assertions", + "rand", "regex", "reqwest", "schemars", "serde", "serde_json", + "sha2", "tempfile", "thiserror", "tokio", "tracing", + "url", + "zeroize", ] [[package]] @@ -230,6 +245,25 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + [[package]] name = "deunicode" version = "1.6.2" @@ -242,6 +276,16 @@ version = "0.1.13" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "56254986775e3233ffa9c4d7d3faaf6d36a2c09d30b20687e9f88bc8bafc16c8" +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", +] + [[package]] name = "displaydoc" version = "0.2.5" @@ -335,6 +379,16 @@ dependencies = [ "slab", ] +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + [[package]] name = "getrandom" version = "0.2.17" @@ -645,6 +699,25 @@ version = "2.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" +[[package]] +name = "is-docker" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "928bae27f42bc99b60d9ac7334e3a21d10ad8f1835a4e12ec3ec0464765ed1b3" +dependencies = [ + "once_cell", +] + +[[package]] +name = "is-wsl" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "173609498df190136aa7dea1a91db051746d339e18476eed5ca40521f02d7aa5" +dependencies = [ + "is-docker", + "once_cell", +] + [[package]] name = "is_terminal_polyfill" version = "1.70.2" @@ -680,6 +753,16 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "keyring" +version = "3.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eebcc3aff044e5944a8fbaf69eb277d11986064cba30c468730e8b9909fb551c" +dependencies = [ + "log", + "zeroize", +] + [[package]] name = "leb128fmt" version = "0.1.0" @@ -770,6 +853,23 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "open" +version = "5.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2fbaa89d2ddc8473c78a3adf69eea8cffa28c483b8e02a971ef31527cd0fc92c" +dependencies = [ + "is-wsl", + "libc", + "pathdiff", +] + +[[package]] +name = "pathdiff" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df94ce210e5bc13cb6651479fa48d14f601d9858cfe0467f43ae157023b938d3" + [[package]] name = "percent-encoding" version = "2.3.2" @@ -1200,6 +1300,17 @@ dependencies = [ "serde", ] +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "shlex" version = "1.3.0" @@ -1474,6 +1585,12 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "typenum" +version = "1.20.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "40ce102ab67701b8526c123c1bab5cbe42d7040ccfd0f64af1a385808d2f43de" + [[package]] name = "unicase" version = "2.9.0" @@ -1522,6 +1639,12 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + [[package]] name = "want" version = "0.3.1" @@ -2063,6 +2186,20 @@ name = "zeroize" version = "1.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b97154e67e32c85465826e8bcc1c59429aaaf107c1e4a9e53c8d8ccd5eff88d0" +dependencies = [ + "zeroize_derive", +] + +[[package]] +name = "zeroize_derive" +version = "1.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85a5b4158499876c763cb03bc4e49185d3cccbabb15b33c627f7884f43db852e" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] [[package]] name = "zerotrie" diff --git a/Cargo.toml b/Cargo.toml index 759f934..0caccb3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,12 @@ path = "src/lib.rs" [dependencies] async-trait = "0.1.89" base64 = "0.22.1" +keyring = { version = "3.6.1", optional = true } +open = { version = "5.3.2", optional = true } +rand = { version = "0.9", optional = true } +sha2 = { version = "0.10.9", optional = true } +url = { version = "2.5.4", optional = true } +zeroize = { version = "1.8", optional = true, features = ["derive"] } chrono = { version = "0.4.42", default-features = false, features = ["clock", "serde"] } clap = { version = "4.5.53", features = ["derive", "std", "string"] } jmespath = "0.5.0" @@ -28,9 +34,12 @@ schemars = { version = "1.2.1", features = ["derive"] } serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.145" thiserror = "2.0.17" -tokio = { version = "1.48.0", features = ["fs", "io-util", "macros", "process", "rt-multi-thread", "signal", "sync", "time"] } +tokio = { version = "1.48.0", features = ["fs", "io-std", "io-util", "macros", "process", "rt-multi-thread", "signal", "sync", "time"] } tracing = "0.1.43" +[features] +pkce-auth = ["dep:keyring", "dep:open", "dep:rand", "dep:sha2", "dep:url", "dep:zeroize"] + [dev-dependencies] pretty_assertions = "1.4.1" tempfile = "3.23.0" diff --git a/README.md b/README.md index 854eba0..21e2180 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,6 @@ # cli-engine -`cli-engine` is a Rust library for building consistent, domain-oriented CLI applications. -It provides shared framework pieces for command registration, authentication, authorization hooks, -middleware, structured output, schemas, guides, search, command tree rendering, and HTTP transport. +`cli-engine` is a Rust library for building consistent, domain-oriented CLI applications. It provides shared framework pieces for command registration, authentication, authorization hooks, middleware, structured output, schemas, guides, search, command tree rendering, and HTTP transport. Consumer CLIs bring their own binary entrypoint and domain modules. `cli-engine` owns the common execution pipeline so teams can add commands by copying a nearby command and filling in the @@ -83,6 +81,12 @@ manifest that produces the same `(path, bytes)` pairs. - [Basic example](examples/basic.rs) - [Agent instructions](AGENTS.md) +## Cargo Features + +| Feature | Description | +| --- | --- | +| `pkce-auth` | Enables `auth::pkce::PkceAuthProvider`, a built-in browser-based OAuth 2.0 PKCE flow with system keychain storage via the `keyring` crate. Adds optional dependencies: `keyring`, `open`, `rand`, `sha2`, `url`. | + ## Verification ```sh diff --git a/docs/auth.md b/docs/auth.md index d86c440..0214aef 100644 --- a/docs/auth.md +++ b/docs/auth.md @@ -173,6 +173,65 @@ For compatibility with existing provider binaries, `ExecProvider` also sends `re value as `env`, accepts credential responses containing `realm`, and can fall back to the `list-realms` action when `list-environments` is not available. +## PkceAuthProvider + +`PkceAuthProvider` is a built-in OAuth 2.0 Authorization Code + PKCE provider (RFC 7636). It +requires the `pkce-auth` Cargo feature. + +```toml +[dependencies] +cli-engine = { path = "...", features = ["pkce-auth"] } +``` + +The provider manages the full browser-based login flow: + +1. Generates a random PKCE code verifier and SHA-256 challenge. +2. Starts a local TCP server on `127.0.0.1` to receive the OAuth callback. +3. Opens the system browser to the authorization endpoint. +4. Exchanges the returned code for tokens at the token endpoint. +5. Stores tokens in the system keychain via the `keyring` crate (macOS Keychain, Linux Secret + Service, or the platform's encrypted file fallback). +6. Refreshes expired tokens automatically using the stored refresh token. + +```rust +use std::sync::Arc; +use cli_engine::{CliConfig, auth::pkce::PkceAuthProvider}; + +let provider = Arc::new( + PkceAuthProvider::new( + "primary", + "https://auth.example.com/oauth/authorize", + "https://auth.example.com/oauth/token", + "my-client-id", + &["openid", "profile"], + ) + .with_app_id("my-cli"), // keychain service prefix +); + +let config = CliConfig::new("my-cli", "My CLI", "my-cli") + .with_default_auth_provider("primary") + .with_auth_provider(provider); +``` + +Tokens are stored per `(app_id, provider_name, env)` tuple. The in-process cache avoids +redundant keychain reads. A 30-second expiry buffer triggers proactive refresh. + +The redirect port defaults to `7443`. Override it with `PkceAuthProvider::with_redirect_port`. + +### Environment Variable Overrides + +At runtime, the provider checks for environment variable overrides before using its compiled-in +values. The prefix is the provider name uppercased with hyphens replaced by underscores: + +| Variable | Purpose | +| --- | --- | +| `_OAUTH_CLIENT_ID` | OAuth client ID. | +| `_OAUTH_AUTH_URL` | Authorization endpoint URL. | +| `_OAUTH_TOKEN_URL` | Token endpoint URL. | + +For a provider named `"primary"`, the variables are `PRIMARY_OAUTH_CLIENT_ID`, +`PRIMARY_OAUTH_AUTH_URL`, and `PRIMARY_OAUTH_TOKEN_URL`. + ## Dispatcher [`Dispatcher`](../src/auth/dispatcher.rs) routes auth calls by provider name: diff --git a/docs/concepts.md b/docs/concepts.md index f29e74e..3c3e819 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -125,6 +125,37 @@ let command = RuntimeCommandSpec::new( Command definitions should stay close to business logic. Use the builder methods to set optional metadata: +### Streaming Commands + +For commands that emit a sequence of progress events rather than a single result — for example, a +deploy command that streams build logs — use `RuntimeCommandSpec::new_streaming`. The handler +receives a `StreamSender` and writes individual `serde_json::Value` events. Each value is written +to stdout as a newline-delimited JSON (NDJSON) line as it arrives. + +```rust +use cli_engine::{CommandSpec, RuntimeCommandSpec, StreamSender, Tier}; +use serde_json::json; + +RuntimeCommandSpec::new_streaming( + CommandSpec::new("deploy", "Deploy and stream progress") + .with_system("deploy-api") + .with_tier(Tier::Mutate), + |_ctx, sender: StreamSender| async move { + sender.send(json!({ "status": "building" })).await; + sender.send(json!({ "status": "deploying", "progress": 42 })).await; + sender.send(json!({ "status": "done" })).await; + Ok(()) + }, +) +``` + +Streaming commands do not go through the normal output pipeline (filtering, field selection, `--output`). +Each event is written verbatim to the process stdout (`tokio::io::stdout()`), bypassing any custom +writer injection from `execute_from` variants. The handler and the NDJSON writer run concurrently +so the handler can keep sending while the writer flushes to stdout. If stdout is under backpressure +the bounded channel can fill and the handler will wait on `send` until the writer catches up. + + - `with_long` for expanded help. - `with_alias` for alternate group or command names. - `hidden(true)` for groups or commands that remain runnable but are omitted from help, tree, and @@ -277,6 +308,10 @@ Auth providers implement the `AuthProvider` trait. Providers expose credential r status, logout, and environment-listing behavior. The framework includes: - `ExecProvider`, which invokes an external provider command using JSON stdin/stdout. +- `PkceAuthProvider` (requires the `pkce-auth` feature), a built-in browser-based OAuth 2.0 PKCE + flow that manages the local callback server, opens the system browser, and persists tokens in the + system keychain via the `keyring` crate. Auth URL, token URL, and client ID can be overridden via + environment variables at runtime. - A `Dispatcher` that routes auth calls by provider name. Single-provider facades created from the dispatcher remain live views of the dispatcher, so transport injectors observe later provider registration or replacement. @@ -322,6 +357,32 @@ Risk tiers classify command impact: Handlers return JSON-serializable data and a system id. Middleware wraps the result in an envelope with data, metadata, errors, and warnings. +### next_actions + +Command handlers can attach a list of follow-on command suggestions to any result using +`CommandResult::with_next_actions`. The framework includes these suggestions in the output +envelope under the `next_actions` key in JSON and TOON output formats. Human output does not +display `next_actions`. + +```rust +use cli_engine::{CommandResult, NextAction, NextActionParam}; +use serde_json::json; +use std::collections::HashMap; + +Ok(CommandResult::new(json!({ "id": "app-1", "name": "my-app" })) + .with_next_actions(vec![ + NextAction { + command: "application info --name my-app".to_owned(), + description: "Get full application details".to_owned(), + params: HashMap::new(), + }, + ])) +``` + +`NextAction` parameters are optional and carry `value`, `enum`, `required`, `default`, and +`description` fields. This is the primary mechanism for agent-first CLIs to tell callers what +command to run next. + The output pipeline runs in this order: 1. **Filtering**: `--filter` evaluates a JMESPath predicate against each item in list data. diff --git a/docs/design.md b/docs/design.md index cc81527..392250b 100644 --- a/docs/design.md +++ b/docs/design.md @@ -44,6 +44,10 @@ src/ tier.rs error.rs auth/ + mod.rs + exec.rs + pkce.rs + ... output/ transport/ tests/ @@ -173,6 +177,8 @@ fn list_projects() -> RuntimeCommandSpec { Use `RuntimeCommandSpec::new_with_context` only when a handler needs the colon command path, user-supplied args, or a middleware snapshot. +Use `RuntimeCommandSpec::new_streaming` for commands that emit a sequence of events rather than a single result. The handler receives a `StreamSender` and writes individual `serde_json::Value` events. Each event is written to stdout as a newline-delimited JSON line as it arrives. The handler and the NDJSON writer run concurrently so the handler can keep sending while the writer flushes to stdout. If stdout is under backpressure the bounded channel can fill and the handler will wait on `send` until the writer catches up. + ### Typed Arguments When commands have many flags or already use `#[derive(clap::Args)]` structs, the typed path avoids @@ -271,6 +277,8 @@ Auth providers implement `AuthProvider` and are registered with the CLI or durin initialization. The dispatcher routes credential operations by provider name and supports the built-in `auth login`, `auth status`, and `auth logout` commands. +`PkceAuthProvider` (behind the `pkce-auth` feature) is a built-in provider that implements the full browser-based OAuth 2.0 PKCE flow. It stores tokens in the system keychain and refreshes them automatically. Consumer CLIs that need a first-party browser login flow can use it directly without writing a provider binary. + Credential fields are serialized as provider-contract JSON and are used by transport injectors, authorization, audit, and activity. @@ -291,6 +299,7 @@ Handlers return JSON-serializable data and a system id. Middleware wraps the res - `metadata` - `error` - `warnings` +- `next_actions` Metadata is omitted unless `--verbose` is requested. Selective metadata is supported with comma-separated verbose fields. diff --git a/src/auth/mod.rs b/src/auth/mod.rs index 6c790ec..15dc0d9 100644 --- a/src/auth/mod.rs +++ b/src/auth/mod.rs @@ -6,6 +6,9 @@ //! //! The module also contains an [`crate::auth::exec::ExecProvider`] for provider //! binaries that speak the JSON stdin/stdout contract. +//! +//! When the `pkce-auth` feature is enabled, `pkce::PkceAuthProvider` adds a +//! built-in browser-based OAuth 2.0 PKCE flow with system keychain storage. /// Built-in `auth login`, `auth status`, and `auth logout` command helpers. pub mod commands; @@ -13,6 +16,9 @@ mod credential; mod dispatcher; /// External process auth provider implementation. pub mod exec; +/// OAuth 2.0 PKCE auth provider (requires the `pkce-auth` feature). +#[cfg(feature = "pkce-auth")] +pub mod pkce; use async_trait::async_trait; diff --git a/src/auth/pkce.rs b/src/auth/pkce.rs new file mode 100644 index 0000000..52544db --- /dev/null +++ b/src/auth/pkce.rs @@ -0,0 +1,597 @@ +//! OAuth 2.0 PKCE authentication provider. +//! +//! Implements the browser-based Authorization Code + PKCE flow (RFC 7636). +//! Tokens are stored in the system keychain via the `keyring` crate. +//! +//! # Setup +//! +//! ```no_run +//! use std::sync::Arc; +//! use cli_engine::{CliConfig, auth::pkce::PkceAuthProvider}; +//! +//! let provider = Arc::new(PkceAuthProvider::new( +//! "my-provider", +//! "https://auth.example.com/oauth/authorize", +//! "https://auth.example.com/oauth/token", +//! "my-client-id", +//! &["openid", "profile"], +//! )); +//! +//! let config = CliConfig::new("mycli", "My CLI", "mycli") +//! .with_default_auth_provider("my-provider") +//! .with_auth_provider(provider); +//! ``` +//! +//! Override endpoints and client ID via environment variables: +//! - `_OAUTH_CLIENT_ID` +//! - `_OAUTH_AUTH_URL` +//! - `_OAUTH_TOKEN_URL` +//! +//! where `` is the provider name uppercased and with `-` replaced by `_`. + +use std::{ + collections::HashMap, + net::{SocketAddr, TcpListener}, + sync::Arc, + time::Duration, +}; + +use async_trait::async_trait; +use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; +use chrono::{SecondsFormat, Utc}; +use rand::Rng; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use tokio::sync::RwLock; +use zeroize::{Zeroize, ZeroizeOnDrop}; + +use crate::{Credential, Result, auth::AuthProvider, error::CliCoreError}; + +const REDIRECT_PORT_DEFAULT: u16 = 7443; +const TOKEN_EXPIRY_BUFFER_SECS: i64 = 30; + +/// Stored token with expiry tracking. +/// +/// Token fields are zeroized on drop to limit in-memory exposure. +#[derive(Clone, Serialize, Deserialize, Zeroize, ZeroizeOnDrop)] +struct StoredToken { + access_token: String, + expires_at: i64, + refresh_token: Option, +} + +impl std::fmt::Debug for StoredToken { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("StoredToken") + .field("access_token", &"[redacted]") + .field("expires_at", &self.expires_at) + .field( + "refresh_token", + if self.refresh_token.is_some() { + &"Some([redacted])" + } else { + &"None" + }, + ) + .finish() + } +} + +impl StoredToken { + fn is_valid(&self) -> bool { + let now = Utc::now().timestamp(); + self.expires_at - TOKEN_EXPIRY_BUFFER_SECS > now + } +} + +/// OAuth 2.0 PKCE authentication provider. +/// +/// Stores one token per `(env, provider)` pair in the system keychain. +/// The keychain service name is `//`. +#[derive(Debug)] +pub struct PkceAuthProvider { + name: String, + auth_url: String, + token_url: String, + client_id: String, + scopes: Vec, + redirect_port: u16, + app_id: String, + env_prefix: String, + /// In-process token cache keyed by env. + cache: Arc>>, +} + +impl PkceAuthProvider { + /// Creates a new PKCE provider. + /// + /// - `name`: Provider registration name (e.g. `"primary"`) + /// - `auth_url`: Authorization endpoint + /// - `token_url`: Token endpoint + /// - `client_id`: OAuth client ID + /// - `scopes`: Default OAuth scopes + #[must_use] + pub fn new( + name: impl Into, + auth_url: impl Into, + token_url: impl Into, + client_id: impl Into, + scopes: &[impl AsRef], + ) -> Self { + let name = name.into(); + let env_prefix = name.to_uppercase().replace('-', "_"); + Self { + name, + auth_url: auth_url.into(), + token_url: token_url.into(), + client_id: client_id.into(), + scopes: scopes.iter().map(|s| s.as_ref().to_owned()).collect(), + redirect_port: REDIRECT_PORT_DEFAULT, + app_id: String::new(), + env_prefix, + cache: Arc::new(RwLock::new(HashMap::new())), + } + } + + /// Sets the local redirect server port (default: 7443). + #[must_use] + pub fn with_redirect_port(mut self, port: u16) -> Self { + self.redirect_port = port; + self + } + + /// Sets the application id used as the keychain service prefix. + #[must_use] + pub fn with_app_id(mut self, app_id: impl Into) -> Self { + self.app_id = app_id.into(); + self + } + + /// Adds extra scopes beyond the default set. + #[must_use] + pub fn with_extra_scopes(mut self, scopes: &[impl AsRef]) -> Self { + self.scopes + .extend(scopes.iter().map(|s| s.as_ref().to_owned())); + self + } + + fn effective_client_id(&self) -> String { + let key = format!("{}_OAUTH_CLIENT_ID", self.env_prefix); + std::env::var(&key).unwrap_or_else(|_| self.client_id.clone()) + } + + fn effective_auth_url(&self) -> String { + let key = format!("{}_OAUTH_AUTH_URL", self.env_prefix); + std::env::var(&key).unwrap_or_else(|_| self.auth_url.clone()) + } + + fn effective_token_url(&self) -> String { + let key = format!("{}_OAUTH_TOKEN_URL", self.env_prefix); + std::env::var(&key).unwrap_or_else(|_| self.token_url.clone()) + } + + fn keychain_service(&self, env: &str) -> String { + if self.app_id.is_empty() { + format!("{}/{}", self.name, env) + } else { + format!("{}/{}/{}", self.app_id, self.name, env) + } + } + + fn keychain_user(&self) -> &str { + "token" + } + + fn load_token_from_keychain(&self, env: &str) -> Option { + let entry = keyring::Entry::new(&self.keychain_service(env), self.keychain_user()).ok()?; + let json = entry.get_password().ok()?; + serde_json::from_str(&json).ok() + } + + fn save_token_to_keychain(&self, env: &str, token: &StoredToken) -> Result<()> { + let entry = keyring::Entry::new(&self.keychain_service(env), self.keychain_user()) + .map_err(|err| CliCoreError::message(format!("keychain access failed: {err}")))?; + let json = serde_json::to_string(token).map_err(CliCoreError::from)?; + entry + .set_password(&json) + .map_err(|err| CliCoreError::message(format!("keychain write failed: {err}")))?; + Ok(()) + } + + fn delete_token_from_keychain(&self, env: &str) { + if let Ok(entry) = keyring::Entry::new(&self.keychain_service(env), self.keychain_user()) { + drop(entry.delete_credential()); + } + } + + async fn cached_token(&self, env: &str) -> Option { + let cache = self.cache.read().await; + cache.get(env).filter(|t| t.is_valid()).cloned() + } + + async fn store_cached_token(&self, env: &str, token: StoredToken) { + let mut cache = self.cache.write().await; + cache.insert(env.to_owned(), token); + } + + async fn resolve_token(&self, env: &str) -> Result { + if let Some(token) = self.cached_token(env).await { + return Ok(token); + } + if let Some(token) = self.load_token_from_keychain(env) { + if token.is_valid() { + self.store_cached_token(env, token.clone()).await; + return Ok(token); + } + if let Some(refresh_token) = token.refresh_token.as_deref() + && let Ok(mut refreshed) = self.refresh_access_token(refresh_token).await + { + if refreshed.refresh_token.is_none() { + refreshed.refresh_token = Some(refresh_token.to_owned()); + } + self.save_token_to_keychain(env, &refreshed)?; + self.store_cached_token(env, refreshed.clone()).await; + return Ok(refreshed); + } + } + let token = self.run_pkce_flow(env).await?; + self.save_token_to_keychain(env, &token)?; + self.store_cached_token(env, token.clone()).await; + Ok(token) + } + + async fn run_pkce_flow(&self, env: &str) -> Result { + let (code_verifier, code_challenge) = pkce_challenge(); + let state = random_state(); + let client_id = self.effective_client_id(); + let auth_url = self.effective_auth_url(); + let redirect_uri = format!("http://127.0.0.1:{}/callback", self.redirect_port); + let scope = self.scopes.join(" "); + + let auth_params = [ + ("response_type", "code"), + ("client_id", &client_id), + ("redirect_uri", &redirect_uri), + ("scope", &scope), + ("state", &state), + ("code_challenge", &code_challenge), + ("code_challenge_method", "S256"), + ]; + let url = url::Url::parse_with_params(&auth_url, &auth_params) + .map_err(|err| CliCoreError::message(format!("invalid auth URL: {err}")))?; + + // Start the local callback server before opening the browser so the + // redirect lands as soon as the user approves. + let listener = TcpListener::bind(SocketAddr::from(([127, 0, 0, 1], self.redirect_port))) + .map_err(|err| { + CliCoreError::message(format!( + "failed to bind callback server on port {}: {err}", + self.redirect_port + )) + })?; + + tracing::info!("Opening browser for authentication…"); + tracing::info!("If the browser does not open, visit:\n {url}"); + drop(open::that(url.as_str())); + + let code = wait_for_callback(listener, &state, Duration::from_secs(120)).await?; + self.exchange_code_for_token(&code, &code_verifier, env) + .await + } + + async fn exchange_code_for_token( + &self, + code: &str, + code_verifier: &str, + env: &str, + ) -> Result { + let redirect_uri = format!("http://127.0.0.1:{}/callback", self.redirect_port); + let client_id = self.effective_client_id(); + let token_url = self.effective_token_url(); + + let params = [ + ("grant_type", "authorization_code"), + ("client_id", &client_id), + ("redirect_uri", &redirect_uri), + ("code", code), + ("code_verifier", code_verifier), + ]; + let response = reqwest::Client::new() + .post(&token_url) + .form(¶ms) + .send() + .await + .map_err(|err| CliCoreError::message(format!("token request failed: {err}")))?; + + if !response.status().is_success() { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(CliCoreError::message(format!( + "token endpoint returned {status}: {body}" + ))); + } + + parse_token_response(response, env).await + } + + async fn refresh_access_token(&self, refresh_token: &str) -> Result { + let client_id = self.effective_client_id(); + let token_url = self.effective_token_url(); + let params = [ + ("grant_type", "refresh_token"), + ("client_id", &client_id), + ("refresh_token", refresh_token), + ]; + let response = reqwest::Client::new() + .post(&token_url) + .form(¶ms) + .send() + .await + .map_err(|err| CliCoreError::message(format!("token refresh failed: {err}")))?; + + if !response.status().is_success() { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(CliCoreError::message(format!( + "refresh endpoint returned {status}: {body}" + ))); + } + + parse_token_response(response, "").await + } +} + +#[async_trait] +impl AuthProvider for PkceAuthProvider { + fn name(&self) -> &str { + &self.name + } + + async fn get_credential(&self, env: &str, _command: &str, _tier: &str) -> Result { + let token = self.resolve_token(env).await?; + Ok(Credential { + token: token.access_token.clone(), + env: env.to_owned(), + provider: self.name.clone(), + expires_at: chrono::DateTime::from_timestamp(token.expires_at, 0) + .map(|dt| dt.to_rfc3339_opts(SecondsFormat::Secs, true)) + .unwrap_or_default(), + ..Credential::default() + }) + } + + async fn status(&self, env: &str) -> Result { + let Some(token) = self.load_token_from_keychain(env) else { + return Err(CliCoreError::message(format!( + "not logged in for environment {env:?}" + ))); + }; + Ok(Credential { + token: token.access_token.clone(), + env: env.to_owned(), + provider: self.name.clone(), + expires_at: chrono::DateTime::from_timestamp(token.expires_at, 0) + .map(|dt| dt.to_rfc3339_opts(SecondsFormat::Secs, true)) + .unwrap_or_default(), + ..Credential::default() + }) + } + + async fn logout(&self, env: &str) -> Result<()> { + self.delete_token_from_keychain(env); + let mut cache = self.cache.write().await; + cache.remove(env); + Ok(()) + } + + async fn list_environments(&self) -> Result> { + // Keyring doesn't support listing; return cache keys as a hint. + let cache = self.cache.read().await; + Ok(cache.keys().cloned().collect()) + } +} + +/// Generates a PKCE code verifier and SHA-256 code challenge. +fn pkce_challenge() -> (String, String) { + let bytes: [u8; 32] = rand::rng().random(); + let verifier = URL_SAFE_NO_PAD.encode(bytes); + let hash = Sha256::digest(verifier.as_bytes()); + let challenge = URL_SAFE_NO_PAD.encode(hash); + (verifier, challenge) +} + +/// Generates a random OAuth state parameter. +fn random_state() -> String { + let bytes: [u8; 16] = rand::rng().random(); + URL_SAFE_NO_PAD.encode(bytes) +} + +/// Waits for the OAuth callback on the given listener, validates state. +/// +/// Accepts connections in a loop so that stray connections (port scanners, +/// browser preflight requests) do not consume the single callback attempt. +async fn wait_for_callback( + listener: TcpListener, + expected_state: &str, + timeout: Duration, +) -> Result { + use std::io::{Read, Write}; + + let expected_state = expected_state.to_owned(); + let result = tokio::time::timeout(timeout, async move { + tokio::task::spawn_blocking(move || { + loop { + let (mut stream, _) = listener.accept().map_err(|err| { + CliCoreError::message(format!("callback server accept failed: {err}")) + })?; + let mut buf = [0_u8; 4096]; + let n = match stream.read(&mut buf) { + Ok(n) => n, + Err(_) => continue, + }; + let request = String::from_utf8_lossy(&buf[..n]); + + let code = extract_query_param(&request, "code"); + let state = extract_query_param(&request, "state"); + + let html_response = + if state.as_deref() == Some(&expected_state) && code.is_some() { + "HTTP/1.1 200 OK\r\nContent-Type: text/html\r\n\r\n\ + Authentication successful. You may close this window." + } else { + "HTTP/1.1 400 Bad Request\r\nContent-Type: text/html\r\n\r\n\ + Authentication failed. Please try again." + }; + drop(stream.write_all(html_response.as_bytes())); + + if state.as_deref() == Some(expected_state.as_str()) { + return code.ok_or_else(|| { + CliCoreError::message("no authorization code in callback") + }); + } + } + }) + .await + .map_err(|err| CliCoreError::message(format!("callback task failed: {err}")))? + }) + .await; + + match result { + Ok(inner) => inner, + Err(_) => Err(CliCoreError::message( + "timed out waiting for OAuth callback", + )), + } +} + +/// Extracts a query parameter value from an HTTP request line. +fn extract_query_param(request: &str, name: &str) -> Option { + let line = request.lines().next()?; + let path = line.split_whitespace().nth(1)?; + let query = path.split_once('?')?.1; + url::form_urlencoded::parse(query.as_bytes()) + .find(|(key, _)| key == name) + .map(|(_, value)| value.into_owned()) +} + +#[derive(Debug, Deserialize)] +struct TokenResponse { + access_token: String, + expires_in: Option, + refresh_token: Option, +} + +async fn parse_token_response(response: reqwest::Response, _env: &str) -> Result { + let body: TokenResponse = response + .json() + .await + .map_err(|err| CliCoreError::message(format!("failed to parse token response: {err}")))?; + let expires_in = body.expires_in.unwrap_or(3600); + let expires_at = Utc::now().timestamp() + expires_in; + Ok(StoredToken { + access_token: body.access_token, + expires_at, + refresh_token: body.refresh_token, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_provider() -> PkceAuthProvider { + PkceAuthProvider::new( + "test", + "https://example.com/auth", + "https://example.com/token", + "client-id", + &["openid"], + ) + } + + fn valid_token(access_token: &str) -> StoredToken { + StoredToken { + access_token: access_token.to_owned(), + expires_at: Utc::now().timestamp() + 3600, + refresh_token: None, + } + } + + fn expired_token() -> StoredToken { + StoredToken { + access_token: "old-token".to_owned(), + // Older than the expiry buffer so is_valid() returns false. + expires_at: Utc::now().timestamp() - TOKEN_EXPIRY_BUFFER_SECS - 1, + refresh_token: None, + } + } + + /// store_cached_token + cached_token round-trip: the mechanism used by + /// the persistence fix must reliably write and read tokens from the cache. + #[tokio::test] + async fn cache_stores_and_retrieves_valid_token() { + let provider = test_provider(); + let token = valid_token("access-abc"); + + provider.store_cached_token("dev", token.clone()).await; + + let cached = provider.cached_token("dev").await; + assert!(cached.is_some(), "expected cached token to be present"); + assert_eq!( + cached.expect("token must be present").access_token, + "access-abc" + ); + } + + /// Expired tokens must not be returned from the cache; the caller would + /// then proceed to the keychain or PKCE flow. + #[tokio::test] + async fn cached_token_ignores_expired_tokens() { + let provider = test_provider(); + provider.store_cached_token("dev", expired_token()).await; + + assert!( + provider.cached_token("dev").await.is_none(), + "expired token should not be returned from cache" + ); + } + + #[test] + fn extract_query_param_skips_malformed_pairs() { + let request = "GET /callback?foo&code=abc123&state=xyz HTTP/1.1\r\nHost: localhost\r\n"; + assert_eq!( + extract_query_param(request, "code"), + Some("abc123".to_owned()), + ); + assert_eq!( + extract_query_param(request, "state"), + Some("xyz".to_owned()), + ); + } + + #[test] + fn extract_query_param_decodes_percent_encoding() { + let request = "GET /callback?code=a%20b%2Bc&state=ok HTTP/1.1\r\n"; + assert_eq!( + extract_query_param(request, "code"), + Some("a b+c".to_owned()), + ); + } + + /// resolve_token must return a pre-seeded in-memory token without + /// triggering the PKCE browser flow (which would require a port and browser). + /// This also exercises the cache-hit path that follows token persistence. + #[tokio::test] + async fn resolve_token_returns_cached_token_without_pkce_flow() { + let provider = test_provider(); + provider + .store_cached_token("dev", valid_token("cached-token")) + .await; + + let resolved = provider + .resolve_token("dev") + .await + .expect("resolve from cache"); + assert_eq!(resolved.access_token, "cached-token"); + } +} diff --git a/src/cli.rs b/src/cli.rs index a5cf29b..8154379 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -14,10 +14,14 @@ mod tree_render; use clap::{ArgMatches, Command}; use crate::{ - AuthProvider, CliCoreError, CommandMeta, CommandSpec, GroupSpec, GuideEntry, Middleware, - MiddlewareRequest, Result, RuntimeCommandSpec, RuntimeGroupSpec, + ActivityEmitter, Auditor, AuthProvider, Authorizer, CliCoreError, CommandMeta, CommandSpec, + GroupSpec, GuideEntry, Middleware, MiddlewareRequest, Result, RuntimeCommandSpec, + RuntimeGroupSpec, auth::commands::auth_command_group, - command::{CommandContext, command_args_from_matches, command_path_from_matches, leaf_matches}, + command::{ + CommandContext, StreamSender, command_args_from_matches, command_path_from_matches, + leaf_matches, + }, error::exit_code_for_error, flags::{ GlobalFlags, derive_bool_flags, derive_value_flags, extract_command_path, @@ -131,6 +135,12 @@ pub struct CliConfig { pub views: Vec, /// Providers registered before command execution starts. pub auth_providers: Vec>, + /// Optional authorization gatekeeper injected into middleware. + pub authz: Option>, + /// Optional audit recorder injected into middleware. + pub auditor: Option>, + /// Optional activity event sink injected into middleware. + pub activity: Option>, /// Optional late initializer for runtime dependencies. pub init_deps: Option, /// Optional hook for adding application-specific global flags. @@ -233,6 +243,27 @@ impl CliConfig { self } + /// Sets the authorization gatekeeper. + #[must_use] + pub fn with_authz(mut self, authz: Arc) -> Self { + self.authz = Some(authz); + self + } + + /// Sets the audit recorder. + #[must_use] + pub fn with_auditor(mut self, auditor: Arc) -> Self { + self.auditor = Some(auditor); + self + } + + /// Sets the activity event sink. + #[must_use] + pub fn with_activity(mut self, activity: Arc) -> Self { + self.activity = Some(activity); + self + } + /// Sets the late dependency initializer. #[must_use] pub fn with_init_deps(mut self, init_deps: InitDeps) -> Self { @@ -298,6 +329,9 @@ impl std::fmt::Debug for CliConfig { .field("guides", &self.guides) .field("views", &self.views) .field("auth_providers_len", &self.auth_providers.len()) + .field("has_authz", &self.authz.is_some()) + .field("has_auditor", &self.auditor.is_some()) + .field("has_activity", &self.activity.is_some()) .field("has_init_deps", &self.init_deps.is_some()) .field("has_register_flags", &self.register_flags.is_some()) .field("has_apply_flags", &self.apply_flags.is_some()) @@ -318,6 +352,15 @@ pub struct CliRunOutput { pub rendered: String, } +impl From for CliRunOutput { + fn from(o: crate::middleware::MiddlewareOutput) -> Self { + Self { + exit_code: o.exit_code, + rendered: o.rendered, + } + } +} + /// Configured CLI application. /// /// A `Cli` owns the `clap` command tree, middleware, registered runtime @@ -424,6 +467,7 @@ impl Cli { } root = register_global_flags(root) .subcommand(help_command()) + .subcommand(guide_command()) .subcommand(Command::new("tree").about("Display full command tree")); if let Some(register_flags) = &config.register_flags { root = register_flags(root); @@ -438,6 +482,9 @@ impl Cli { let mut middleware = Middleware::new(); middleware.app_id = config.app_id.clone(); middleware.default_auth_provider = config.default_auth_provider.clone().unwrap_or_default(); + middleware.authz = config.authz.clone(); + middleware.auditor = config.auditor.clone(); + middleware.activity = config.activity.clone(); middleware .schema_registry .merge(&global_schema_registry_snapshot()); @@ -810,6 +857,33 @@ impl Cli { let meta = self.resolve_meta(&command_path, command.spec.metadata()); let default_fields = command.spec.default_fields.clone().unwrap_or_default(); let system = command.spec.system.clone().unwrap_or_default(); + + if let Some(streaming_handler) = command.streaming_handler.clone() { + let result = run_with_timeout( + command_timeout, + &flags.timeout, + run_streaming_command( + &middleware, + MiddlewareRequest { + meta, + command_path: &command_path, + system: &system, + user_args, + args, + default_fields: &default_fields, + no_auth: command.spec.no_auth, + }, + Arc::new(leaf.clone()), + streaming_handler, + ), + ) + .await; + return self.finish_run(match result { + Ok(output) => output, + Err(err) => render_cli_error(&middleware, &err, &self.config.app_id), + }); + } + let handler = command.handler.clone(); let args_for_handler = args.clone(); let user_args_for_handler = user_args.clone(); @@ -845,14 +919,8 @@ impl Cli { .await; match result { - Ok(output) => self.finish_run(CliRunOutput { - exit_code: output.exit_code, - rendered: output.rendered, - }), - Err(err) => self.finish_run(CliRunOutput { - exit_code: exit_code_for_error(&err), - rendered: render_cli_error(&middleware, &err, &self.config.app_id).rendered, - }), + Ok(output) => self.finish_run(output.into()), + Err(err) => self.finish_run(render_cli_error(&middleware, &err, &self.config.app_id)), } } @@ -1425,7 +1493,10 @@ fn has_root_version_flag(args: &[String], root: &Command, root_name: &str) -> bo let bool_flags = derive_bool_flags(root); let value_flags = derive_value_flags(root); let mut iter = args.iter().peekable(); - if iter.peek().is_some_and(|arg| arg.as_str() == root_name) { + if iter + .peek() + .is_some_and(|arg| arg_matches_root_name(arg, root_name)) + { iter.next(); } @@ -1455,7 +1526,7 @@ fn normalize_optional_global_flags_before_command(root: &Command, args: &[String let mut current = root; while index < args.len() { let arg = &args[index]; - if index == 0 && arg == root.get_name() { + if index == 0 && arg_matches_root_name(arg, root.get_name()) { normalized.push(arg.clone()); index += 1; continue; @@ -1551,7 +1622,10 @@ fn positional_command_tokens( ) -> Vec { let mut tokens = Vec::new(); let mut iter = args.iter().peekable(); - if iter.peek().is_some_and(|arg| arg.as_str() == root_name) { + if iter + .peek() + .is_some_and(|arg| arg_matches_root_name(arg, root_name)) + { iter.next(); } @@ -1582,6 +1656,14 @@ fn unknown_flag_consumes_value(arg: &str, next: Option<&&String>) -> bool { arg.starts_with('-') && next.is_some_and(|value| !value.starts_with('-')) } +fn arg_matches_root_name(arg: &str, root_name: &str) -> bool { + arg == root_name + || std::path::Path::new(arg) + .file_stem() + .and_then(|n| n.to_str()) + .is_some_and(|n| n == root_name) +} + fn register_runtime_group_schemas( group: &RuntimeGroupSpec, prefix: &mut Vec, @@ -1686,3 +1768,73 @@ fn process_exit_code(code: i32) -> ExitCode { Ok(_) | Err(_) => ExitCode::from(1), } } + +async fn run_streaming_command( + middleware: &Middleware, + request: MiddlewareRequest<'_>, + raw_matches: Arc, + streaming_handler: crate::command::StreamingCommandHandler, +) -> Result { + use tokio::{io::AsyncWriteExt, sync::mpsc}; + + let args_for_handler = request.args.clone(); + let user_args_for_handler = request.user_args.clone(); + let handler_path = request.command_path.to_owned(); + let middleware_for_handler = middleware.clone(); + let raw_matches_for_handler = raw_matches; + + let (tx, mut rx) = mpsc::channel::(64); + let sender = StreamSender(tx); + + // Drain the channel concurrently so the handler's sends don't stall + // while the writer flushes to stdout. If stdout is under backpressure + // the bounded channel can still fill and the handler will await send. + let writer = tokio::spawn(async move { + let mut stdout = tokio::io::stdout(); + while let Some(event) = rx.recv().await { + let Ok(line) = serde_json::to_string(&event) else { + continue; + }; + if stdout.write_all(line.as_bytes()).await.is_err() + || stdout.write_all(b"\n").await.is_err() + || stdout.flush().await.is_err() + { + break; + } + } + }); + + let output = middleware + .run(request, async move |credential| { + streaming_handler( + CommandContext { + credential, + args: args_for_handler, + user_args: user_args_for_handler, + command_path: handler_path, + middleware: middleware_for_handler, + raw_matches: raw_matches_for_handler, + }, + sender, + ) + .await?; + Ok(crate::CommandResult::new(serde_json::Value::Null)) + }) + .await; + + // Handler has completed; its sender is dropped, which closes the channel. + // Wait for the writer task to flush all remaining events. + let _write_result = writer.await; + + match output { + Ok(out) if out.exit_code == 0 => Ok(CliRunOutput { + exit_code: 0, + rendered: String::new(), + }), + Ok(out) => Ok(out.into()), + Err(err) => Ok(CliRunOutput { + exit_code: exit_code_for_error(&err), + rendered: render_cli_error(middleware, &err, middleware.app_id.as_str()).rendered, + }), + } +} diff --git a/src/command.rs b/src/command.rs index 03b6d9a..92464b6 100644 --- a/src/command.rs +++ b/src/command.rs @@ -3,17 +3,38 @@ use std::{collections::BTreeMap, future::Future, pin::Pin, sync::Arc}; use clap::{Arg, ArgAction, ArgMatches, Command}; use schemars::JsonSchema; use serde_json::{Number, Value}; +use tokio::sync::mpsc; use crate::{ CommandMeta, Credential, Middleware, OutputSchema, Result, SchemaInfo, Tier, - middleware::ValueMap, + middleware::ValueMap, output::NextAction, }; +/// Sender half for streaming command output. +/// +/// Streaming handlers call [`StreamSender::send`] for each progress event. +/// The engine drains the channel and writes each event as an NDJSON line. +#[derive(Clone, Debug)] +pub struct StreamSender(pub(crate) mpsc::Sender); + +impl StreamSender { + /// Sends one event. Silently drops the event if the receiver is gone. + pub async fn send(&self, event: Value) { + drop(self.0.send(event).await); + } +} + /// Boxed future returned by runtime command handlers. pub type CommandFuture = Pin> + Send>>; /// Shared command handler used by [`RuntimeCommandSpec`]. pub type CommandHandler = Arc CommandFuture + Send + Sync>; +/// Boxed future returned by streaming command handlers. +pub type StreamingCommandFuture = Pin> + Send>>; +/// Shared streaming handler: receives context and an event sender; returns when the stream ends. +pub type StreamingCommandHandler = + Arc StreamingCommandFuture + Send + Sync>; + /// Data returned by a command handler. /// /// Command handlers should return renderable data and keep output metadata on @@ -36,6 +57,13 @@ impl CommandResult { metadata: CommandResultMetadata::default(), } } + + /// Attaches suggested follow-up actions to this result. + #[must_use] + pub fn with_next_actions(mut self, actions: Vec) -> Self { + self.metadata.next_actions = actions; + self + } } impl From for CommandResult { @@ -47,7 +75,10 @@ impl From for CommandResult { /// Optional metadata a command can attach to its result. #[non_exhaustive] #[derive(Clone, Debug, Default, Eq, PartialEq)] -pub struct CommandResultMetadata {} +pub struct CommandResultMetadata { + /// Suggested follow-up actions for the caller. + pub next_actions: Vec, +} /// Runtime context passed to advanced command handlers. /// @@ -411,12 +442,18 @@ impl GroupSpec { /// `RuntimeCommandSpec` pairs a [`CommandSpec`] with async business logic. /// This split keeps metadata inspectable for help/search/schema generation /// before the handler ever runs. +/// +/// Use [`RuntimeCommandSpec::new_streaming`] for commands that emit incremental +/// NDJSON progress events (e.g. long-running deployments with `--follow`). #[derive(Clone)] pub struct RuntimeCommandSpec { /// Declarative command metadata. pub spec: CommandSpec, /// Async command implementation. pub handler: CommandHandler, + /// Optional streaming handler. When set, the engine writes NDJSON events + /// to stdout as they arrive instead of collecting a single envelope. + pub streaming_handler: Option, } impl std::fmt::Debug for RuntimeCommandSpec { @@ -424,6 +461,7 @@ impl std::fmt::Debug for RuntimeCommandSpec { formatter .debug_struct("RuntimeCommandSpec") .field("spec", &self.spec) + .field("is_streaming", &self.streaming_handler.is_some()) .finish_non_exhaustive() } } @@ -442,6 +480,7 @@ impl RuntimeCommandSpec { { Self { spec, + streaming_handler: None, handler: Arc::new(move |context| { let future = handler(context.credential, context.args); Box::pin(async move { future.await.map(Into::into) }) @@ -459,6 +498,7 @@ impl RuntimeCommandSpec { { Self { spec, + streaming_handler: None, handler: Arc::new(move |context| { let future = handler(context); Box::pin(async move { future.await.map(Into::into) }) @@ -466,6 +506,28 @@ impl RuntimeCommandSpec { } } + /// Creates a streaming command that emits NDJSON events to stdout. + /// + /// The handler receives context and a [`StreamSender`]. It should call + /// `sender.send(event).await` for each progress event, then return `Ok(())`. + /// The engine writes each event as a JSON line; stdout is flushed after each. + #[must_use] + pub fn new_streaming(spec: CommandSpec, handler: F) -> Self + where + F: Fn(CommandContext, StreamSender) -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, + { + let streaming: StreamingCommandHandler = Arc::new(move |context, sender| { + let future = handler(context, sender); + Box::pin(future) + }); + Self { + spec, + streaming_handler: Some(streaming), + handler: Arc::new(|_context| Box::pin(async { Ok(CommandResult::new(Value::Null)) })), + } + } + /// Creates a runtime command with typed argument deserialization. /// /// The handler receives the optional credential and the deserialized args @@ -497,6 +559,7 @@ impl RuntimeCommandSpec { handler(credential, args).await.map(Into::into) }) }), + streaming_handler: None, } } } diff --git a/src/lib.rs b/src/lib.rs index 2901c67..e400257 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -98,7 +98,8 @@ pub use cli::{ }; pub use command::{ CommandContext, CommandFuture, CommandHandler, CommandResult, CommandResultMetadata, - CommandSpec, GroupSpec, RuntimeCommandSpec, RuntimeGroupSpec, command_args_from_matches, + CommandSpec, GroupSpec, RuntimeCommandSpec, RuntimeGroupSpec, StreamSender, + StreamingCommandFuture, StreamingCommandHandler, command_args_from_matches, command_path_from_matches, command_path_from_parts, leaf_matches, }; pub use error::{ @@ -117,10 +118,10 @@ pub use middleware::{ pub use module::{CommandModule, Module, ModuleContext, ModuleRegister}; pub use output::{ Envelope, ErrorEnvelope, FieldInfo, HumanViewDef, HumanViewFn, HumanViewRegistry, - HumanViewRenderer, Metadata, OutputField, OutputFormat, OutputSchema, PaginationMeta, - PipelineOpts, RendererFactory, SchemaInfo, SchemaRegistry, TableColumn, apply_pipeline, - build_detailed_error_envelope, build_error_envelope, fields_for, fields_from_json_schema, - filter_fields, format_help_section, get_global_schema_by_path, + HumanViewRenderer, Metadata, NextAction, NextActionParam, OutputField, OutputFormat, + OutputSchema, PaginationMeta, PipelineOpts, RendererFactory, SchemaInfo, SchemaRegistry, + TableColumn, apply_pipeline, build_detailed_error_envelope, build_error_envelope, fields_for, + fields_from_json_schema, filter_fields, format_help_section, get_global_schema_by_path, global_human_view_registry_snapshot, global_schema_registry_snapshot, is_valid_output_format, json_schema_for, json_schema_info, lookup_global_human_view_columns, lookup_global_human_view_func, parse_fields, register_global_human_view, diff --git a/src/middleware.rs b/src/middleware.rs index 0d7323f..c828f0d 100644 --- a/src/middleware.rs +++ b/src/middleware.rs @@ -380,8 +380,9 @@ impl Middleware { ) .await; + let CommandResult { data, metadata } = result; self.render_envelope( - Envelope::success(result.data, command_system), + Envelope::success(data, command_system).with_next_actions(metadata.next_actions), default_fields, command_path, start, diff --git a/src/output/envelope.rs b/src/output/envelope.rs index df7d9ea..4d63892 100644 --- a/src/output/envelope.rs +++ b/src/output/envelope.rs @@ -1,4 +1,4 @@ -use std::time::Duration; +use std::{collections::HashMap, time::Duration}; use chrono::{SecondsFormat, Utc}; use serde::{Deserialize, Serialize}; @@ -21,10 +21,84 @@ pub struct Envelope { /// Non-fatal warnings. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub warnings: Vec, + /// Suggested follow-up actions for the caller (agent or human). + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub next_actions: Vec, #[serde(default, skip)] serialization_error: Option, } +/// A suggested follow-up command the caller can run next. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct NextAction { + /// Executable command template, e.g. `"application info --name {{name}}"`. + pub command: String, + /// Human-readable description of what this action does. + pub description: String, + /// Optional parameter hints for agent-driven invocation. + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + pub params: HashMap, +} + +impl NextAction { + /// Creates a next action with a command template and description. + #[must_use] + pub fn new(command: impl Into, description: impl Into) -> Self { + Self { + command: command.into(), + description: description.into(), + params: HashMap::new(), + } + } + + /// Adds a parameter hint. + #[must_use] + pub fn with_param(mut self, name: impl Into, param: NextActionParam) -> Self { + self.params.insert(name.into(), param); + self + } +} + +/// Metadata hint for a parameter in a [`NextAction`] command template. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize, Default)] +pub struct NextActionParam { + /// Concrete value to substitute, if known. + #[serde(skip_serializing_if = "Option::is_none")] + pub value: Option, + /// Allowed values for enumeration parameters. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub r#enum: Vec, + /// Whether the parameter is required. + #[serde(default, skip_serializing_if = "is_false")] + pub required: bool, + /// Default value when none is supplied. + #[serde(skip_serializing_if = "Option::is_none")] + pub default: Option, + /// Human-readable description of this parameter. + #[serde(skip_serializing_if = "Option::is_none")] + pub description: Option, +} + +impl NextActionParam { + /// Creates a parameter hint with a known concrete value. + #[must_use] + pub fn value(value: impl Into) -> Self { + Self { + value: Some(value.into()), + ..Self::default() + } + } + + /// Creates a required parameter hint. + #[must_use] + pub fn required() -> Self { + Self { + required: true, + ..Self::default() + } + } +} + /// Execution metadata attached to an [`Envelope`]. #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct Metadata { @@ -102,6 +176,7 @@ impl Envelope { metadata: Some(Metadata::new(system)), error: None, warnings: Vec::new(), + next_actions: Vec::new(), serialization_error, } } @@ -124,6 +199,7 @@ impl Envelope { request_id: String::new(), }), warnings: Vec::new(), + next_actions: Vec::new(), serialization_error: None, } } @@ -151,10 +227,18 @@ impl Envelope { request_id, }), warnings: Vec::new(), + next_actions: Vec::new(), serialization_error: None, } } + /// Attaches suggested follow-up actions. + #[must_use] + pub fn with_next_actions(mut self, actions: Vec) -> Self { + self.next_actions = actions; + self + } + /// Marks the envelope as a dry-run response. #[must_use] pub fn with_dry_run(mut self) -> Self { @@ -288,6 +372,7 @@ pub fn build_error_envelope(err: &(dyn std::error::Error + 'static), system: &st request_id, }), warnings: Vec::new(), + next_actions: Vec::new(), serialization_error: None, }; } @@ -397,6 +482,7 @@ pub fn build_detailed_error_envelope(err: &dyn DetailedError, system: &str) -> E request_id, }), warnings: Vec::new(), + next_actions: Vec::new(), serialization_error: None, } } @@ -447,3 +533,70 @@ fn is_absent_null_or_empty_object(value: &Option) -> bool { Some(Value::Array(_) | Value::Bool(_) | Value::Number(_) | Value::String(_)) => false, } } + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + #[test] + fn next_actions_appear_in_serialized_envelope() { + let envelope = + Envelope::success(json!({"id": "p1"}), "projects-api").with_next_actions(vec![ + NextAction::new("project get --id {{id}}", "Get project details"), + ]); + + let serialized = serde_json::to_string(&envelope).expect("envelope serializes to JSON"); + let parsed: Value = + serde_json::from_str(&serialized).expect("serialized envelope is valid JSON"); + + assert_eq!( + parsed["next_actions"][0]["command"], + "project get --id {{id}}" + ); + assert_eq!( + parsed["next_actions"][0]["description"], + "Get project details" + ); + } + + #[test] + fn next_actions_omitted_from_json_when_empty() { + let envelope = Envelope::success(json!({"id": "p1"}), "projects-api"); + + let serialized = serde_json::to_string(&envelope).expect("envelope serializes to JSON"); + let parsed: Value = + serde_json::from_str(&serialized).expect("serialized envelope is valid JSON"); + + assert!( + parsed.get("next_actions").is_none(), + "empty next_actions must not appear in JSON output" + ); + } + + #[test] + fn next_action_params_serialize_when_present() { + let action = NextAction::new("deploy run --app {{app}}", "Deploy the app").with_param( + "app", + NextActionParam { + description: Some("Application name".to_owned()), + required: true, + value: None, + r#enum: Vec::new(), + default: None, + }, + ); + let envelope = Envelope::success(json!(null), "deploy-api").with_next_actions(vec![action]); + + let serialized = serde_json::to_string(&envelope).expect("envelope serializes to JSON"); + let parsed: Value = + serde_json::from_str(&serialized).expect("serialized envelope is valid JSON"); + + assert_eq!( + parsed["next_actions"][0]["params"]["app"]["description"], + "Application name" + ); + assert_eq!(parsed["next_actions"][0]["params"]["app"]["required"], true); + } +} diff --git a/src/output/mod.rs b/src/output/mod.rs index 6e13e6c..ebfd305 100644 --- a/src/output/mod.rs +++ b/src/output/mod.rs @@ -19,8 +19,8 @@ mod toon; pub use crate::error::{DetailedError, ExitCoder, exit_code_for_error, exit_code_for_exit_coder}; pub use envelope::{ - Envelope, ErrorEnvelope, Metadata, PaginationMeta, build_detailed_error_envelope, - build_error_envelope, + Envelope, ErrorEnvelope, Metadata, NextAction, NextActionParam, PaginationMeta, + build_detailed_error_envelope, build_error_envelope, }; pub use fields::{FieldTree, filter_fields, parse_fields}; pub use human::{ diff --git a/src/transport/client.rs b/src/transport/client.rs index 5db74d1..8a2c42a 100644 --- a/src/transport/client.rs +++ b/src/transport/client.rs @@ -303,6 +303,19 @@ impl HttpClient { Ok(()) } + /// Sends GET and returns the raw response body as bytes. + pub async fn get_bytes(&self, path: &str) -> Result> { + let response = self.send_get_raw_status_only_retry(path).await?; + if response.status().is_client_error() || response.status().is_server_error() { + return Err(parse_error_response(response, "GET", path).await.into()); + } + let bytes = response + .bytes() + .await + .map_err(|err| CliCoreError::message(format!("transport: stream response: {err}")))?; + Ok(bytes.to_vec()) + } + /// Sends POST and streams the raw response body into a writer. pub async fn post_raw( &self, diff --git a/tests/streaming.rs b/tests/streaming.rs new file mode 100644 index 0000000..375dedd --- /dev/null +++ b/tests/streaming.rs @@ -0,0 +1,142 @@ +use std::sync::Arc; + +use async_trait::async_trait; +use cli_engine::{ + AuthProvider, Cli, CliConfig, CliCoreError, CommandSpec, Credential, GroupSpec, Module, Result, + RuntimeCommandSpec, RuntimeGroupSpec, StreamSender, +}; +use serde_json::json; + +#[derive(Debug)] +struct AlwaysFailAuth; + +#[async_trait] +impl AuthProvider for AlwaysFailAuth { + fn name(&self) -> &str { + "always-fail" + } + + async fn get_credential(&self, _env: &str, _command: &str, _tier: &str) -> Result { + Err(CliCoreError::message( + "auth failed: no credentials configured", + )) + } + + async fn status(&self, _env: &str) -> Result { + Err(CliCoreError::message("not logged in")) + } + + async fn logout(&self, _env: &str) -> Result<()> { + Ok(()) + } + + async fn list_environments(&self) -> Result> { + Ok(vec![]) + } +} + +#[allow(closure_returning_async_block)] +fn cli_with_erroring_handler(msg: &'static str) -> Cli { + let module = Module::new("Streaming Tests", move |_| { + RuntimeGroupSpec::new(GroupSpec::new("deploy", "Deploy commands")).with_command( + RuntimeCommandSpec::new_streaming( + CommandSpec::new("run", "Run a deploy").no_auth(true), + move |_ctx, _sender: StreamSender| async move { + Err::<(), _>(CliCoreError::message(msg)) + }, + ), + ) + }); + Cli::new(CliConfig::new("test-cli", "Test CLI", "test").with_module(module)) +} + +fn cli_with_failing_auth() -> Cli { + let module = Module::new("Streaming Tests", |_| { + RuntimeGroupSpec::new(GroupSpec::new("deploy", "Deploy commands")).with_command( + RuntimeCommandSpec::new_streaming( + CommandSpec::new("run", "Run a deploy"), + async |_ctx, sender: StreamSender| { + sender.send(json!({"status": "deploying"})).await; + Ok(()) + }, + ), + ) + }); + Cli::new( + CliConfig::new("test-cli", "Test CLI", "test") + .with_auth_provider(Arc::new(AlwaysFailAuth)) + .with_default_auth_provider("always-fail") + .with_module(module), + ) +} + +/// Streaming handler errors must propagate as a non-zero exit code. +/// +/// Before the fix, middleware.run rendered errors into Ok(MiddlewareOutput{exit_code:non-zero}) +/// but the streaming match arm always returned exit_code:0. +#[tokio::test] +async fn streaming_handler_error_exits_nonzero() { + let result = cli_with_erroring_handler("deployment failed") + .run(["test-cli", "deploy", "run"]) + .await; + + assert_ne!(result.exit_code, 0, "handler error should exit non-zero"); + assert!( + result.rendered.contains("deployment failed"), + "expected error message in rendered output, got: {}", + result.rendered + ); +} + +fn cli_with_successful_handler() -> Cli { + let module = Module::new("Streaming Tests", |_| { + RuntimeGroupSpec::new(GroupSpec::new("deploy", "Deploy commands")).with_command( + RuntimeCommandSpec::new_streaming( + CommandSpec::new("run", "Run a deploy").no_auth(true), + async |_ctx, sender: StreamSender| { + sender.send(json!({"status": "building"})).await; + sender.send(json!({"status": "done"})).await; + Ok(()) + }, + ), + ) + }); + Cli::new(CliConfig::new("test-cli", "Test CLI", "test").with_module(module)) +} + +/// Successful streaming commands must exit 0 with no trailing rendered output. +/// +/// NDJSON events are written directly to stdout by the writer task. The +/// framework must not append a rendered envelope (e.g. `{"data": null}`) +/// after the events. +#[tokio::test] +async fn streaming_success_exits_zero_with_no_rendered_output() { + let result = cli_with_successful_handler() + .run(["test-cli", "deploy", "run"]) + .await; + + assert_eq!(result.exit_code, 0, "successful stream should exit 0"); + assert!( + result.rendered.is_empty(), + "successful stream should produce no rendered output, got: {:?}", + result.rendered + ); +} + +/// Auth failures on streaming commands must exit non-zero. +/// +/// Middleware renders auth errors as Ok(MiddlewareOutput{exit_code:non-zero}), which +/// the streaming match arm must preserve rather than discard. +#[tokio::test] +async fn streaming_auth_failure_exits_nonzero() { + let result = cli_with_failing_auth() + .run(["test-cli", "deploy", "run"]) + .await; + + assert_ne!(result.exit_code, 0, "auth failure should exit non-zero"); + assert!( + result.rendered.contains("auth failed"), + "expected auth error in rendered output, got: {}", + result.rendered + ); +}