diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c48b53..9961f09 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,15 +5,25 @@ ### Added - `--redirect-uri`, `TEAMS_CLI_REDIRECT_URI` and the profile's `redirect_uri` select the browser login's redirect URI for `auth login`, `auth consent-url` and `auth doctor`. The login listens on the URI's host and port (both loopback addresses for `localhost`) for a request to the URI's path carrying the login's `state`; the flag is saved to the profile with the IDs. A redirect URI without a port gets a free port from the OS at each login, as MSAL does; the identity platform allows this because it ignores the port of a localhost redirect URI, so `--redirect-uri http://localhost` works for an app registered that way. +- `teams subscribe create --lifecycle-url URL` sets the subscription's `lifecycleNotificationUrl`, which Graph requires for Teams subscriptions longer than one hour. Graph validates that URL too, so `teams listen` answers validation and notifications on `/lifecycle` as well as `/` and `/webhook`, and one tunnel serves both. +- `teams listen --client-state SECRET` (or `TEAMS_CLI_CLIENT_STATE`) drops notifications whose `clientState` is missing or different, with a warning on stderr that does not include either value. Without it, the listener warns at startup that notifications are not authenticated. +- `teams listen --bind ADDR` chooses the listening address, and `--port 0` picks a free port; the startup line names the address actually bound. +- Each `teams listen` line now carries every field Graph sent (`subscriptionExpirationDateTime`, `lifecycleEvent`, `encryptedContent` and so on) and `kind`: `lifecycle` when `lifecycleEvent` is present, else `change`. The README's "Running behind a relay" says what a relay should POST and which statuses to retry. ### Changed +- **`teams listen` binds `127.0.0.1` by default instead of `0.0.0.0`.** Pass `--bind 0.0.0.0` to accept connections from other hosts. +- `teams listen` answers notifications with 202 once they are queued for stdout, instead of 200 after printing, so a slow stdout reader no longer holds the HTTP response; when the queue (8 MiB) is full it answers 503 so the sender retries. Validation requests are answered whatever the queue state. On Ctrl+C or SIGTERM it stops admitting notifications (503) and writes out every one it already acknowledged before exiting; a second signal abandons them, and any acknowledged notification left unwritten, or lost to a closed stdout, is counted on stderr with exit code 1 (previously a closed stdout exited 0). +- `teams listen` limits bodies to 1 MiB (413), allows 10 seconds for headers and for the body (408), and serves at most 64 connections at once. +- `teams listen` no longer prints `clientState`. - The default redirect URI is now `http://localhost/callback` on a free port instead of `http://localhost:8400/callback`, so a busy port 8400 no longer blocks login. App registrations with `http://localhost:8400/callback` keep working because the port is ignored. - Updated Rust dependencies (`tokio` 1.54.1, `serde_json` 1.0.154, `uuid` 1.28.0, `webbrowser` 1.2.5) and the CI and release workflows' `Swatinem/rust-cache` pin to v2.9.3. `rust-version` in `Cargo.toml` is now 1.89, the oldest Rust the dependencies build with; it said 1.75. - Updated `sha2` to 0.11.1 (it computes the PKCE code challenge for browser login). The challenge is now unit-tested against the RFC 7636 example, so a hashing or encoding change cannot slip through as failed logins. ### Fixed +- Log lines on stderr are coloured only when stderr is a terminal and `NO_COLOR` is unset, so `teams listen 2>listen.log` and other redirected logs no longer contain ANSI escape codes. `--no-color` still turns colour off. +- `teams listen` on a port already in use exits with code 10 (`CONFIG_ERROR`), naming the address and the OS error, instead of reporting an authentication failure (exit code 3). - The browser login keeps serving its loopback redirect URI until the callback arrives. It served one connection, so a browser's speculative connection or a second request could leave it waiting forever or failing with "Failed to receive auth code from callback". A redirect carrying the identity platform's `error` (a declined consent, say) now fails the login at once with `error` and `error_description` (exit code 3) instead of that generic message, requests to other paths or with another `state` are ignored, and the login gives up after 10 minutes with no callback. - Updated `rustls-webpki` to 0.103.16 and `webpki-roots` to 1.0.11 for RUSTSEC-2026-0382 and RUSTSEC-2026-0381, under which certificates issued by distrusted intermediates could still be accepted. The `tokio` update also clears RUSTSEC-2026-0386 (unsound `AsyncFd` constructors). diff --git a/CLAUDE.md b/CLAUDE.md index 4611c1e..5f88d2e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -85,10 +85,12 @@ CLI flags > env vars (TEAMS_CLI_CLIENT_ID, TEAMS_CLI_CLIENT_SECRET, TEAMS_CLI_TE - `$top` parameter for page size control ### Webhook Listener -- `teams listen --port 8080` starts a hyper HTTP server -- Handles Graph subscription validation (echoes `?validationToken`) -- Outputs change notifications as NDJSON to stdout -- Multi-connection, graceful Ctrl+C shutdown +- `teams listen --port 8080` starts a hyper HTTP server on 127.0.0.1 (`--bind` to change) +- Handles Graph subscription validation (echoes `?validationToken`) on `/`, `/webhook` and `/lifecycle` +- Outputs change notifications as NDJSON to stdout via a bounded queue and writer thread: 202 once queued, 503 when full +- `--client-state` / `TEAMS_CLI_CLIENT_STATE` drops notifications with another or no `clientState`; output strips `clientState` and adds `kind` +- Limits: 1 MiB body (413), 10s header and body deadlines, 64 connections +- Ctrl+C/SIGTERM: stop admitting (503), write out everything acknowledged; a second signal abandons it; unwritten acknowledged lines → count on stderr, exit 1 - Requires HTTPS via reverse proxy (ngrok) for production use ## Environment Variables @@ -96,6 +98,7 @@ CLI flags > env vars (TEAMS_CLI_CLIENT_ID, TEAMS_CLI_CLIENT_SECRET, TEAMS_CLI_TE - `TEAMS_CLI_CLIENT_SECRET` — Azure AD client secret - `TEAMS_CLI_TENANT_ID` — Azure AD tenant ID - `TEAMS_CLI_ACCESS_TOKEN` — Pre-obtained access token +- `TEAMS_CLI_CLIENT_STATE` — clientState `teams listen` requires on notifications - `TEAMS_CLI_TOKEN_STORE` — `keyring` (default) or `file` (0600 files under the config dir's `tokens/`) - `RUST_LOG` — Tracing filter level diff --git a/README.md b/README.md index a3ec034..f8061e2 100644 --- a/README.md +++ b/README.md @@ -616,6 +616,7 @@ teams subscribe create \ --resource "/teams/{team-id}/channels/{channel-id}/messages" \ --change-type created,updated \ --webhook-url https://your-endpoint/webhook +# Over one hour, Graph also needs --lifecycle-url https://your-endpoint/lifecycle teams subscribe list teams subscribe renew [--expiration ] teams subscribe delete @@ -625,15 +626,43 @@ teams listen --port 8080 teams listen --port 8080 | jq '.changeType, .resource' ``` +The listener binds `127.0.0.1` unless `--bind` says otherwise (`--bind 0.0.0.0` for all interfaces). It answers Graph's validation request and accepts notifications on `/`, `/webhook` and `/lifecycle`, so one tunnel can serve both `--webhook-url` and `--lifecycle-url` (Graph validates both when the subscription is created). Each line is one notification with every field Graph sent except `clientState`, plus `"kind": "change"` or, for lifecycle events (`lifecycleEvent`), `"kind": "lifecycle"`. + +Pass the subscription's `--client-state` to `listen` too (or set `TEAMS_CLI_CLIENT_STATE`) and notifications without that `clientState` are dropped with a warning on stderr. Without it, anyone who can reach the port can inject lines, and the listener says so at startup. + **Note:** Microsoft Graph requires HTTPS for webhook URLs. Use [ngrok](https://ngrok.com) or similar: ```bash +export TEAMS_CLI_CLIENT_STATE="$(openssl rand -hex 16)" ngrok http 8080 teams subscribe create --resource "/teams/all/messages" \ - --change-type created --webhook-url https://xxxx.ngrok.io/webhook + --change-type created --webhook-url https://xxxx.ngrok.io/webhook \ + --client-state "$TEAMS_CLI_CLIENT_STATE" teams listen --port 8080 ``` +#### Running behind a relay + +A relay can forward notifications to `teams listen` instead of Graph calling it directly: + +- POST the raw Graph body, `{"value":[...]}`, to `/`, `/webhook` or `/lifecycle`. +- Graph's validation request (`?validationToken=`) needs a synchronous reply within 10 seconds, so answer it at the public edge (echo the decoded token as `text/plain`) rather than forwarding it. +- Bind to the interface the relay uses with `--bind`, and set `--client-state` so forged posts are dropped. + +Responses: + +| Status | Meaning | Relay should | +|--------|---------|--------------| +| 202 | Queued for stdout, or dropped for a wrong `clientState` | Not retry | +| 400 | Body is not a notification collection | Not retry | +| 408 | Body not received within 10 seconds | Retry | +| 413 | Body over 1 MiB | Not retry | +| 503 | Output queue full (stdout reader is behind), or shutting down | Retry with backoff | + +At most 64 connections are served at once; others wait. Headers must arrive within 10 seconds. + +On Ctrl+C or SIGTERM the listener stops accepting, answers 503, and writes out every notification it already answered 202, however long stdout takes. A second signal abandons the rest. If any acknowledged notification is not written (second signal, or stdout closed), it prints how many on stderr and exits 1. + ### User Lookup ```bash @@ -731,6 +760,7 @@ auth_flow = "device-code" | `TEAMS_CLI_PROFILE` | Named credential profile (overridden by `--profile`; an explicit value beats the config's `default.profile`) | | `TEAMS_CLI_SCOPES` | Delegated OAuth scopes for login (same precedence as `--scopes`; ignored by client credentials) | | `TEAMS_CLI_ACCESS_TOKEN` | Pre-obtained Microsoft Graph access token (skips login entirely; `--profile`/`TEAMS_CLI_PROFILE` do not affect which token is used) | +| `TEAMS_CLI_CLIENT_STATE` | Secret `teams listen` requires in each notification's `clientState` (same as `--client-state`) | | `RUST_LOG` | Tracing filter (e.g., `debug`, `teams=trace`) | ## Verbosity diff --git a/docs/command-reference.md b/docs/command-reference.md index 7053881..1f09a8b 100644 --- a/docs/command-reference.md +++ b/docs/command-reference.md @@ -282,14 +282,14 @@ If `file download` runs without `--path`, raw bytes go to stdout. Use `--path` i ## Subscriptions and listener ```bash -teams subscribe create --resource RESOURCE --change-type created,updated,deleted --webhook-url URL [--expiration ISO8601] [--client-state STATE] +teams subscribe create --resource RESOURCE --change-type created,updated,deleted --webhook-url URL [--expiration ISO8601] [--client-state STATE] [--lifecycle-url URL] teams subscribe list teams subscribe renew SUBSCRIPTION_ID [--expiration ISO8601] teams subscribe delete SUBSCRIPTION_ID -teams listen [--port PORT] +teams listen [--port PORT] [--bind ADDR] [--client-state SECRET] ``` -Microsoft Graph requires an HTTPS notification URL. `teams listen` is a local HTTP listener for testing behind a tunnel or reverse proxy. +Microsoft Graph requires an HTTPS notification URL. `teams listen` is a local HTTP listener for testing behind a tunnel or reverse proxy. It binds `127.0.0.1` unless `--bind` says otherwise. With `--client-state` (or `TEAMS_CLI_CLIENT_STATE`), notifications without that `clientState` are dropped. Output lines omit `clientState` and add `kind` (`change` or `lifecycle`). For relays and status codes, see "Running behind a relay" in the README. ## Completions diff --git a/docs/man/teams.1 b/docs/man/teams.1 index 0eb5a93..0599607 100644 --- a/docs/man/teams.1 +++ b/docs/man/teams.1 @@ -375,16 +375,48 @@ for a destination file because global is reserved for json, human, and plain output formats. .SH SUBSCRIPTIONS AND LISTENER .nf -teams subscribe create --resource RESOURCE --change-type TYPES --webhook-url URL [--expiration ISO8601] [--client-state STATE] +teams subscribe create --resource RESOURCE --change-type TYPES --webhook-url URL [--expiration ISO8601] [--client-state STATE] [--lifecycle-url URL] teams subscribe list teams subscribe renew SUBSCRIPTION_ID [--expiration ISO8601] teams subscribe delete SUBSCRIPTION_ID -teams listen [--port PORT] +teams listen [--port PORT] [--bind ADDR] [--client-state SECRET] .fi .PP Microsoft Graph requires HTTPS notification URLs. The local listener is plain HTTP and is intended to sit behind a reverse proxy such as ngrok for testing. -It emits change notifications as newline-delimited JSON. +It emits change notifications as newline-delimited JSON: every field Graph +sent except +.B clientState, +plus +.B kind +(\fBlifecycle\fR when +.B lifecycleEvent +is present, else \fBchange\fR). +.PP +The listener binds 127.0.0.1 unless +.B --bind +names another address, such as 0.0.0.0. +With +.B --client-state +(or +.B TEAMS_CLI_CLIENT_STATE), +notifications without that clientState are dropped with a warning on stderr; +without it, the listener warns that notifications are not authenticated. +.PP +Behind a relay, POST the raw Graph body +.B {"value":[...]} +to /, /webhook or /lifecycle, and answer Graph's validationToken request +at the public edge. Responses: 202 queued, or dropped for a wrong clientState +(do not retry); 400 malformed (do not retry); 408 body not received within 10 +seconds; 413 body over 1 MiB (do not retry); 503 output queue full because +the stdout reader is behind, or shutting down (retry with backoff). At most 64 +connections are served at once, and headers must arrive within 10 seconds. +.PP +On Ctrl+C or SIGTERM the listener stops accepting, answers 503, and writes out +every notification it already answered 202. A second signal abandons the rest. +If any acknowledged notification is not written, the count goes to stderr and +the exit code is 1. +A port already in use exits with code 10. .SH COMPLETIONS .nf teams completions bash @@ -406,6 +438,12 @@ Prefer it to .B --client-secret, whose value shows in process listings and shell history. .TP +.B TEAMS_CLI_CLIENT_STATE +Secret +.B teams listen +requires in each notification's clientState. Same as +.B --client-state. +.TP .B TEAMS_CLI_TENANT_ID Azure AD tenant ID. .TP diff --git a/src/cli/listen.rs b/src/cli/listen.rs index deed6d0..63ff7b1 100644 --- a/src/cli/listen.rs +++ b/src/cli/listen.rs @@ -1,5 +1,8 @@ +use std::net::IpAddr; + use crate::error::Result; -pub async fn run(port: u16) -> Result<()> { - crate::listen::run_listener(port).await +pub async fn run(bind: IpAddr, port: u16, client_state: Option) -> Result<()> { + let client_state = client_state.filter(|s| !s.is_empty()); + crate::listen::run_listener(bind, port, client_state).await } diff --git a/src/cli/mod.rs b/src/cli/mod.rs index ab26cb4..7ff39c1 100644 --- a/src/cli/mod.rs +++ b/src/cli/mod.rs @@ -164,9 +164,16 @@ pub enum Commands { }, /// Start a webhook listener for change notifications Listen { - /// Port to listen on + /// Port to listen on (0 picks a free port) #[arg(long, default_value = "8080")] port: u16, + /// Address to bind. Use 0.0.0.0 to accept connections from other hosts + #[arg(long, default_value = "127.0.0.1")] + bind: std::net::IpAddr, + /// Print only notifications carrying this clientState. Prefer the + /// environment variable: a flag value shows in process listings + #[arg(long, env = "TEAMS_CLI_CLIENT_STATE", hide_env_values = true)] + client_state: Option, }, } @@ -234,7 +241,11 @@ pub async fn run(cli: Cli, config: &ConfigFile) -> Result<()> { Commands::Subscribe { command } => { subscribe::run(command, &runtime_config, &profile, format, &pagination).await } - Commands::Listen { port } => listen::run(port).await, + Commands::Listen { + port, + bind, + client_state, + } => listen::run(bind, port, client_state).await, } } diff --git a/src/cli/subscribe.rs b/src/cli/subscribe.rs index d70bf94..2973baa 100644 --- a/src/cli/subscribe.rs +++ b/src/cli/subscribe.rs @@ -27,6 +27,9 @@ pub enum SubscribeCommand { /// Client state string for validation #[arg(long)] client_state: Option, + /// Lifecycle notification URL; Graph requires it for subscriptions over one hour + #[arg(long)] + lifecycle_url: Option, }, /// List active subscriptions List, @@ -62,6 +65,7 @@ pub async fn run( webhook_url, expiration, client_state, + lifecycle_url, } => { let start = Instant::now(); let expiration_date_time = expiration.unwrap_or_else(|| { @@ -74,6 +78,7 @@ pub async fn run( resource, expiration_date_time, client_state, + lifecycle_notification_url: lifecycle_url, }; let sub = api::subscriptions::create_subscription(&client, &req).await?; output::print_success(format, &sub, start); diff --git a/src/listen/handler.rs b/src/listen/handler.rs index f691196..b424a8d 100644 --- a/src/listen/handler.rs +++ b/src/listen/handler.rs @@ -1,45 +1,146 @@ -use http_body_util::{BodyExt, Full}; -use hyper::body::{Bytes, Incoming}; +use std::convert::Infallible; +use std::sync::Arc; +use std::time::Duration; + +use http_body_util::{BodyExt, Full, LengthLimitError, Limited}; +use hyper::body::{Body, Bytes}; use hyper::{Method, Request, Response, StatusCode}; +use sha2::{Digest, Sha256}; +use super::queue::{Push, Queue}; use crate::models::subscription::ChangeNotificationCollection; -use crate::output; -type HandlerResult = std::result::Result>, hyper::Error>; +/// Largest notification body accepted. +pub const MAX_BODY: usize = 1024 * 1024; +/// Time allowed to receive a request body. +pub const BODY_TIMEOUT: Duration = Duration::from_secs(10); + +type HandlerResult = std::result::Result>, Infallible>; + +pub struct State { + /// SHA-256 of the expected clientState, when one is set. + client_state: Option<[u8; 32]>, + queue: Queue, +} + +impl State { + pub fn new(client_state: Option<&str>, queue: Queue) -> Self { + Self { + client_state: client_state.map(digest), + queue, + } + } + + pub fn queue(&self) -> &Queue { + &self.queue + } + + fn accepts(&self, client_state: Option<&str>) -> bool { + match (&self.client_state, client_state) { + (None, _) => true, + (Some(want), Some(got)) => constant_time_eq(want, &digest(got)), + (Some(_), None) => false, + } + } +} + +fn digest(s: &str) -> [u8; 32] { + Sha256::digest(s.as_bytes()).into() +} + +/// Compare digests without an early exit, so timing does not reveal a prefix. +fn constant_time_eq(a: &[u8; 32], b: &[u8; 32]) -> bool { + let diff = a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)); + std::hint::black_box(diff) == 0 +} -pub async fn handle_request(req: Request) -> HandlerResult { +pub async fn handle_request(req: Request, state: Arc) -> HandlerResult +where + B: Body, + B::Error: Into>, +{ match (req.method(), req.uri().path()) { - (&Method::GET, "/") => Ok(ok_response("OK")), + (&Method::GET, "/") => Ok(plain_response(StatusCode::OK, "OK")), - (&Method::POST, "/") | (&Method::POST, "/webhook") => { - // Check for validation token in query string + // `/lifecycle` is the path the README suggests for --lifecycle-url; + // Graph validates that URL too, so it must answer like the others. + (&Method::POST, "/" | "/webhook" | "/lifecycle") => { + // Validation is answered before the body is read or the output + // queue is consulted, so it works whatever stdout is doing. if let Some(query) = req.uri().query() { - let params: Vec<(String, String)> = url::form_urlencoded::parse(query.as_bytes()) - .into_owned() - .collect(); - - if let Some((_, token)) = params.iter().find(|(k, _)| k == "validationToken") { - return Ok(plain_response(StatusCode::OK, token)); + if let Some((_, token)) = url::form_urlencoded::parse(query.as_bytes()) + .find(|(k, _)| k == "validationToken") + { + return Ok(plain_response(StatusCode::OK, &token)); } } - // Parse notification body - let body_bytes = req.into_body().collect().await?.to_bytes(); - match serde_json::from_slice::(&body_bytes) { - Ok(collection) => { - for notification in &collection.value { - let json = serde_json::to_string(notification).unwrap_or_default(); - output::write_stdout_line(&json); + if req.body().size_hint().lower() > MAX_BODY as u64 { + return Ok(too_large()); + } + let body = Limited::new(req.into_body(), MAX_BODY); + let body_bytes = match tokio::time::timeout(BODY_TIMEOUT, body.collect()).await { + Ok(Ok(collected)) => collected.to_bytes(), + Ok(Err(e)) if e.downcast_ref::().is_some() => { + return Ok(too_large()); + } + Ok(Err(e)) => { + return Ok(plain_response( + StatusCode::BAD_REQUEST, + &format!("Failed to read body: {e}"), + )); + } + Err(_) => { + return Ok(plain_response( + StatusCode::REQUEST_TIMEOUT, + "Timed out reading body", + )) + } + }; + + let collection = + match serde_json::from_slice::(&body_bytes) { + Ok(collection) => collection, + Err(e) => { + tracing::warn!("Failed to parse notification: {e}"); + return Ok(plain_response( + StatusCode::BAD_REQUEST, + &format!("Invalid notification body: {e}"), + )); } - Ok(ok_response("")) + }; + + let mut lines = Vec::with_capacity(collection.value.len()); + for notification in &collection.value { + if !state.accepts(notification.client_state.as_deref()) { + tracing::warn!( + subscription_id = ?notification.subscription_id, + "Dropped notification: clientState missing or wrong" + ); + continue; } - Err(e) => { - tracing::warn!("Failed to parse notification: {e}"); + lines.push(notification.to_output().to_string()); + } + + // Rejected items still get 202: an error only makes Graph resend + // them. A batch is queued whole or not at all, so a retry after + // 503 does not duplicate lines. + if lines.is_empty() { + return Ok(plain_response(StatusCode::ACCEPTED, "")); + } + match state.queue.try_push(lines) { + Push::Queued => Ok(plain_response(StatusCode::ACCEPTED, "")), + Push::Full => { + tracing::warn!("Output queue full; answering 503"); Ok(plain_response( - StatusCode::BAD_REQUEST, - &format!("Invalid notification body: {e}"), + StatusCode::SERVICE_UNAVAILABLE, + "Output queue full", )) } + Push::Closed => Ok(plain_response( + StatusCode::SERVICE_UNAVAILABLE, + "Shutting down", + )), } } @@ -47,12 +148,8 @@ pub async fn handle_request(req: Request) -> HandlerResult { } } -fn ok_response(body: &str) -> Response> { - Response::builder() - .status(StatusCode::OK) - .header("Content-Type", "text/plain") - .body(Full::new(Bytes::from(body.to_string()))) - .unwrap() +fn too_large() -> Response> { + plain_response(StatusCode::PAYLOAD_TOO_LARGE, "Body exceeds 1 MiB") } fn plain_response(status: StatusCode, body: &str) -> Response> { @@ -66,22 +163,164 @@ fn plain_response(status: StatusCode, body: &str) -> Response> { #[cfg(test)] mod tests { use super::*; + use hyper::body::Frame; + use std::sync::mpsc as std_mpsc; - #[test] - fn health_check_returns_ok() { - let resp = ok_response("OK"); + fn state(client_state: Option<&str>) -> (Arc, std_mpsc::Receiver) { + let (tx, rx) = std_mpsc::channel(); + let queue = Queue::new(1024 * 1024, move |text| { + let _ = tx.send(text.to_string()); + Ok(()) + }); + (Arc::new(State::new(client_state, queue)), rx) + } + + fn post(uri: &str, body: &str) -> Request> { + Request::post(uri) + .body(Full::new(Bytes::from(body.to_string()))) + .unwrap() + } + + async fn body_text(resp: Response>) -> String { + let bytes = resp.into_body().collect().await.unwrap().to_bytes(); + String::from_utf8(bytes.to_vec()).unwrap() + } + + #[tokio::test] + async fn health_check_returns_ok() { + let (state, _) = state(None); + let req = Request::get("/").body(Full::new(Bytes::new())).unwrap(); + let resp = handle_request(req, state).await.unwrap(); assert_eq!(resp.status(), StatusCode::OK); } - #[test] - fn validation_token_echo() { - let resp = plain_response(StatusCode::OK, "my-token-123"); + #[tokio::test] + async fn validation_token_is_echoed_decoded_as_text() { + let (state, _) = state(Some("s")); + let resp = handle_request(post("/webhook?validationToken=a%20b%2Bc", ""), state) + .await + .unwrap(); assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(resp.headers()["Content-Type"], "text/plain"); + assert_eq!(body_text(resp).await, "a b+c"); } - #[test] - fn not_found_response() { - let resp = plain_response(StatusCode::NOT_FOUND, "Not Found"); + #[tokio::test] + async fn lifecycle_path_is_validated_and_accepts_notifications() { + let (state, rx) = state(None); + let resp = handle_request(post("/lifecycle?validationToken=tok", ""), state.clone()) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + assert_eq!(body_text(resp).await, "tok"); + + let body = r#"{"value":[{"subscriptionId":"s","lifecycleEvent":"missed"}]}"#; + let resp = handle_request(post("/lifecycle", body), state) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::ACCEPTED); + let line: serde_json::Value = + serde_json::from_str(rx.recv_timeout(Duration::from_secs(10)).unwrap().trim()).unwrap(); + assert_eq!(line["kind"], "lifecycle"); + } + + #[tokio::test] + async fn not_found_response() { + let (state, _) = state(None); + let resp = handle_request(post("/other", ""), state).await.unwrap(); assert_eq!(resp.status(), StatusCode::NOT_FOUND); } + + #[tokio::test] + async fn notifications_with_wrong_client_state_are_dropped() { + let (state, rx) = state(Some("right")); + let body = r#"{"value":[ + {"subscriptionId":"forged","clientState":"wrong"}, + {"subscriptionId":"missing"}, + {"subscriptionId":"ok","clientState":"right","changeType":"created"} + ]}"#; + let resp = handle_request(post("/", body), state).await.unwrap(); + assert_eq!(resp.status(), StatusCode::ACCEPTED); + let out = rx.recv_timeout(Duration::from_secs(10)).unwrap(); + let lines: Vec = out + .lines() + .map(|l| serde_json::from_str(l).unwrap()) + .collect(); + assert_eq!(lines.len(), 1); + assert_eq!(lines[0]["subscriptionId"], "ok"); + assert_eq!(lines[0]["kind"], "change"); + assert!(lines[0].get("clientState").is_none()); + } + + #[tokio::test] + async fn invalid_json_is_bad_request() { + let (state, _) = state(None); + let resp = handle_request(post("/", "not json"), state).await.unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + } + + /// A body of `n` 64 KiB frames with no size hint, like a chunked request. + struct Chunks(usize); + + impl Body for Chunks { + type Data = Bytes; + type Error = Infallible; + + fn poll_frame( + mut self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll, Infallible>>> { + if self.0 == 0 { + return std::task::Poll::Ready(None); + } + self.0 -= 1; + let frame = Frame::data(Bytes::from(vec![b' '; 64 * 1024])); + std::task::Poll::Ready(Some(Ok(frame))) + } + } + + #[tokio::test] + async fn oversized_body_without_length_is_rejected() { + let (state, _) = state(None); + let req = Request::post("/").body(Chunks(20)).unwrap(); + let resp = handle_request(req, state).await.unwrap(); + assert_eq!(resp.status(), StatusCode::PAYLOAD_TOO_LARGE); + } + + #[tokio::test] + async fn shutting_down_answers_503_but_still_validates() { + let (state, rx) = state(None); + state.queue().close(); + let resp = handle_request(post("/", r#"{"value":[{"id":"1"}]}"#), state.clone()) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE); + assert!(rx.try_recv().is_err()); + let resp = handle_request(post("/?validationToken=t", ""), state) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + } + + #[tokio::test] + async fn full_queue_answers_503() { + // Smaller than one line, so the push is refused. + let queue = Queue::new(4, |_| Ok(())); + let state = Arc::new(State::new(None, queue)); + let resp = handle_request(post("/", r#"{"value":[{"id":"1"}]}"#), state) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE); + } + + #[test] + fn client_state_comparison() { + let (state, _) = state(Some("secret")); + assert!(state.accepts(Some("secret"))); + assert!(!state.accepts(Some("secreT"))); + assert!(!state.accepts(Some(""))); + assert!(!state.accepts(None)); + let (open, _) = self::state(None); + assert!(open.accepts(None)); + } } diff --git a/src/listen/mod.rs b/src/listen/mod.rs index fdd24f1..dbdbec3 100644 --- a/src/listen/mod.rs +++ b/src/listen/mod.rs @@ -1,39 +1,82 @@ mod handler; +mod queue; -use std::net::SocketAddr; +use std::net::{IpAddr, SocketAddr}; +use std::sync::Arc; +use std::time::Duration; use hyper::service::service_fn; -use hyper_util::rt::TokioIo; +use hyper_util::rt::{TokioIo, TokioTimer}; +use tokio::sync::Semaphore; use crate::error::{Result, TeamsError}; +use crate::output; -/// Run the webhook listener HTTP server on the given port. +/// Connections served at once; further ones wait in the accept backlog. +const MAX_CONNECTIONS: usize = 64; +/// Time allowed to send request headers, including on an idle keep-alive +/// connection, so a stalled client releases its connection slot. +const HEADER_TIMEOUT: Duration = Duration::from_secs(10); + +/// Run the webhook listener HTTP server on `bind:port`. /// /// The server handles: /// - GET `/` — health check (200 OK) -/// - POST with `?validationToken=xxx` — Graph subscription validation (echoes token) -/// - POST with JSON body — change notification processing (prints NDJSON to stdout) +/// - POST to `/`, `/webhook` or `/lifecycle`: +/// - with `?validationToken=xxx` — Graph subscription validation (echoes token) +/// - with a JSON body — change and lifecycle notifications, printed as NDJSON +/// to stdout by a writer thread; 202 once queued, 503 when the queue is full +/// +/// On Ctrl+C or SIGTERM it stops admitting notifications (503) and writes out +/// everything already acknowledged; a second signal abandons the rest. If any +/// acknowledged notification is not written, the count goes to stderr and the +/// process exits 1. +/// +/// With `client_state`, notifications carrying another or no `clientState` +/// are dropped. /// /// Note: Microsoft Graph webhooks require HTTPS. Use a reverse proxy (e.g., ngrok) /// in front of this listener for production use. -pub async fn run_listener(port: u16) -> Result<()> { - let addr: SocketAddr = ([0, 0, 0, 0], port).into(); - let listener = tokio::net::TcpListener::bind(addr).await.map_err(|e| { - TeamsError::AuthError(format!("Failed to bind listener on port {port}: {e}")) - })?; +pub async fn run_listener(bind: IpAddr, port: u16, client_state: Option) -> Result<()> { + let addr = SocketAddr::new(bind, port); + let listener = tokio::net::TcpListener::bind(addr) + .await + .map_err(|e| TeamsError::ConfigError(format!("cannot listen on {addr}: {e}")))?; + let local = listener.local_addr().unwrap_or(addr); - eprintln!("Webhook listener started on http://0.0.0.0:{port}"); + let queue = queue::Queue::new(queue::OUTPUT_BUDGET, output::try_write_stdout_line); + let state = Arc::new(handler::State::new(client_state.as_deref(), queue)); + let slots = Arc::new(Semaphore::new(MAX_CONNECTIONS)); + + eprintln!("Webhook listener started on http://{local}"); + if client_state.is_none() { + eprintln!( + "Warning: notifications are not authenticated; set --client-state to verify them" + ); + } eprintln!("Press Ctrl+C to stop"); + let mut signals = Signals::new()?; loop { tokio::select! { - accept = listener.accept() => { + (slot, accept) = async { + let slot = slots.clone().acquire_owned().await; + (slot, listener.accept().await) + } => { + let Ok(slot) = slot else { continue }; match accept { Ok((stream, _remote)) => { let io = TokioIo::new(stream); + let state = state.clone(); tokio::spawn(async move { + let _slot = slot; + let service = service_fn(move |req| { + handler::handle_request(req, state.clone()) + }); let _ = hyper::server::conn::http1::Builder::new() - .serve_connection(io, service_fn(handler::handle_request)) + .timer(TokioTimer::new()) + .header_read_timeout(HEADER_TIMEOUT) + .serve_connection(io, service) .await; }); } @@ -42,10 +85,56 @@ pub async fn run_listener(port: u16) -> Result<()> { } } } - _ = tokio::signal::ctrl_c() => { - eprintln!("\nShutting down webhook listener"); - return Ok(()); + _ = signals.recv() => { + eprintln!("\nShutting down webhook listener; Ctrl+C again to abandon queued output"); + break; } + _ = state.queue().writer_failed() => break, + } + } + + drop(listener); + let lost = state.queue().shutdown(signals.recv()).await; + if lost > 0 { + // Exit here: the error envelope would go to the stdout that failed. + eprintln!("Error: {lost} acknowledged notifications were not written to stdout"); + std::process::exit(1); + } + Ok(()) +} + +/// Ctrl+C, and SIGTERM on Unix. Kept for the whole run so a signal between +/// polls, or a second one during shutdown, is not missed. +struct Signals { + #[cfg(unix)] + int: tokio::signal::unix::Signal, + #[cfg(unix)] + term: tokio::signal::unix::Signal, +} + +impl Signals { + #[cfg(unix)] + fn new() -> Result { + use tokio::signal::unix::{signal, SignalKind}; + let register = |kind| signal(kind).map_err(|e| TeamsError::Other(anyhow::Error::from(e))); + Ok(Self { + int: register(SignalKind::interrupt())?, + term: register(SignalKind::terminate())?, + }) + } + + #[cfg(not(unix))] + fn new() -> Result { + Ok(Self {}) + } + + async fn recv(&mut self) { + #[cfg(unix)] + tokio::select! { + _ = self.int.recv() => {} + _ = self.term.recv() => {} } + #[cfg(not(unix))] + let _ = tokio::signal::ctrl_c().await; } } diff --git a/src/listen/queue.rs b/src/listen/queue.rs new file mode 100644 index 0000000..f0e58fb --- /dev/null +++ b/src/listen/queue.rs @@ -0,0 +1,256 @@ +use std::future::Future; +use std::io; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use tokio::sync::{mpsc, Notify, OwnedSemaphorePermit, Semaphore}; + +/// Bytes of output that may wait for stdout before requests get 503. +pub const OUTPUT_BUDGET: usize = 8 * 1024 * 1024; + +type Entry = (Vec, OwnedSemaphorePermit); + +/// Output waiting for stdout, bounded in bytes, so a slow reader never +/// blocks an HTTP response. A dedicated thread writes entries in order. +/// +/// Every line admitted has been acknowledged to the sender, so shutdown +/// closes admission first and then waits for the writer to catch up. +pub struct Queue { + /// `None` once admission is closed. The lock makes closing atomic with + /// respect to a push in progress. + tx: Mutex>>, + budget: Arc, + shared: Arc, +} + +#[derive(Default)] +struct Shared { + accepted: AtomicU64, + written: AtomicU64, + failed: AtomicBool, + /// Signalled after each entry and when the writer stops. + progress: Notify, +} + +#[derive(Debug, PartialEq, Eq)] +pub enum Push { + Queued, + Full, + Closed, +} + +impl Queue { + pub fn new( + capacity: usize, + mut sink: impl FnMut(&str) -> io::Result<()> + Send + 'static, + ) -> Self { + let (tx, mut rx) = mpsc::unbounded_channel::(); + let shared = Arc::new(Shared::default()); + let writer = shared.clone(); + std::thread::Builder::new() + .name("teams-listen-output".into()) + .spawn(move || { + // The permit is held until the writes return. + while let Some((lines, _permit)) = rx.blocking_recv() { + for line in &lines { + if let Err(e) = sink(line) { + eprintln!("Error: Failed to write to stdout: {e}"); + writer.failed.store(true, Ordering::SeqCst); + writer.progress.notify_one(); + return; + } + writer.written.fetch_add(1, Ordering::SeqCst); + } + writer.progress.notify_one(); + } + }) + .expect("spawn output thread"); + Self { + tx: Mutex::new(Some(tx)), + budget: Arc::new(Semaphore::new(capacity)), + shared, + } + } + + /// Enqueue `lines` whole, or refuse them. + pub fn try_push(&self, lines: Vec) -> Push { + let tx = self.tx.lock().unwrap(); + let Some(tx) = tx.as_ref().filter(|_| !self.failed()) else { + return Push::Closed; + }; + let bytes: usize = lines.iter().map(|l| l.len() + 1).sum(); + let Ok(n) = u32::try_from(bytes) else { + return Push::Full; + }; + let Ok(permit) = self.budget.clone().try_acquire_many_owned(n) else { + return Push::Full; + }; + // Counted before sending so `written` never runs ahead of it. + let count = lines.len() as u64; + self.shared.accepted.fetch_add(count, Ordering::SeqCst); + if tx.send((lines, permit)).is_err() { + self.shared.accepted.fetch_sub(count, Ordering::SeqCst); + return Push::Closed; + } + Push::Queued + } + + /// Refuse further pushes. Lines already queued are still written. + pub fn close(&self) { + self.tx.lock().unwrap().take(); + } + + /// Lines acknowledged but not (yet) written. + pub fn pending(&self) -> u64 { + let written = self.shared.written.load(Ordering::SeqCst); + self.shared + .accepted + .load(Ordering::SeqCst) + .saturating_sub(written) + } + + fn failed(&self) -> bool { + self.shared.failed.load(Ordering::SeqCst) + } + + /// Resolves once stdout has failed. + pub async fn writer_failed(&self) { + while !self.failed() { + self.wait_progress().await; + } + } + + /// Resolves once everything queued is written or stdout has failed. + pub async fn drained(&self) { + while self.pending() > 0 && !self.failed() { + self.wait_progress().await; + } + } + + async fn wait_progress(&self) { + // The timeout covers a notification consumed by another waiter. + let _ = + tokio::time::timeout(Duration::from_millis(100), self.shared.progress.notified()).await; + } + + /// Close admission and write out the queue, unless `abort` resolves + /// first. Returns the number of acknowledged lines not written. + pub async fn shutdown(&self, abort: impl Future) -> u64 { + self.close(); + tokio::select! { + _ = self.drained() => {} + _ = abort => {} + } + self.pending() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::mpsc as std_mpsc; + + fn lines(items: &[&str]) -> Vec { + items.iter().map(|s| s.to_string()).collect() + } + + /// A queue whose sink writes one line per message on `release`. + fn gated(capacity: usize) -> (Queue, std_mpsc::Sender<()>, std_mpsc::Receiver) { + let (release_tx, release_rx) = std_mpsc::channel::<()>(); + let (out_tx, out_rx) = std_mpsc::channel::(); + let queue = Queue::new(capacity, move |text| { + release_rx + .recv() + .map_err(|_| io::Error::from(io::ErrorKind::BrokenPipe))?; + out_tx.send(text.to_string()).unwrap(); + Ok(()) + }); + (queue, release_tx, out_rx) + } + + #[tokio::test] + async fn full_queue_refuses_until_the_writer_catches_up() { + let (queue, release, out) = gated(12); + + // The writer takes the first entry and blocks in the sink holding its + // bytes; the second fills the budget. + assert_eq!(queue.try_push(lines(&["aaaaa"])), Push::Queued); + assert_eq!(queue.try_push(lines(&["bbbbb"])), Push::Queued); + assert_eq!(queue.try_push(lines(&["c"])), Push::Full); + + release.send(()).unwrap(); + assert_eq!(out.recv().unwrap(), "aaaaa"); + // Space frees once the write returns. + let deadline = std::time::Instant::now() + Duration::from_secs(10); + while queue.try_push(lines(&["c"])) != Push::Queued { + assert!(std::time::Instant::now() < deadline, "space never freed"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + release.send(()).unwrap(); + release.send(()).unwrap(); + assert_eq!(out.recv().unwrap(), "bbbbb"); + assert_eq!(out.recv().unwrap(), "c"); + } + + #[test] + fn entry_larger_than_the_budget_is_refused() { + let queue = Queue::new(5, |_| Ok(())); + assert_eq!(queue.try_push(lines(&["12345"])), Push::Full); + assert_eq!(queue.try_push(lines(&["1234"])), Push::Queued); + } + + #[tokio::test] + async fn closed_queue_refuses_pushes() { + let (queue, _release, _out) = gated(1024); + queue.close(); + assert_eq!(queue.try_push(lines(&["a"])), Push::Closed); + assert_eq!(queue.pending(), 0); + } + + #[tokio::test] + async fn shutdown_writes_everything_acknowledged() { + let (queue, release, out) = gated(1024); + assert_eq!(queue.try_push(lines(&["a", "b"])), Push::Queued); + assert_eq!(queue.try_push(lines(&["c"])), Push::Queued); + for _ in 0..3 { + release.send(()).unwrap(); + } + let lost = tokio::time::timeout( + Duration::from_secs(10), + queue.shutdown(std::future::pending()), + ) + .await + .expect("drain hung"); + assert_eq!(lost, 0); + let written: Vec = out.try_iter().collect(); + assert_eq!(written, ["a", "b", "c"]); + assert_eq!(queue.try_push(lines(&["d"])), Push::Closed); + } + + #[tokio::test] + async fn aborted_shutdown_reports_lines_not_written() { + let (queue, release, _out) = gated(1024); + assert_eq!(queue.try_push(lines(&["a", "b", "c"])), Push::Queued); + release.send(()).unwrap(); + while queue.pending() > 2 { + tokio::time::sleep(Duration::from_millis(10)).await; + } + // The sink stays blocked on "b", as with a stalled stdout reader. + assert_eq!(queue.shutdown(std::future::ready(())).await, 2); + drop(release); + } + + #[tokio::test] + async fn failed_stdout_closes_admission_and_counts_the_loss() { + let (queue, release, _out) = gated(1024); + assert_eq!(queue.try_push(lines(&["a", "b"])), Push::Queued); + release.send(()).unwrap(); + drop(release); // the next write fails + tokio::time::timeout(Duration::from_secs(10), queue.writer_failed()) + .await + .expect("failure not reported"); + assert_eq!(queue.try_push(lines(&["c"])), Push::Closed); + assert_eq!(queue.shutdown(std::future::pending()).await, 1); + } +} diff --git a/src/main.rs b/src/main.rs index b3f48fc..7b2cca5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -46,7 +46,11 @@ async fn main() { tracing_subscriber::fmt() .with_env_filter(filter) .with_writer(std::io::stderr) - .with_ansi(!cli.no_color) + .with_ansi(log_colors( + cli.no_color, + std::env::var_os("NO_COLOR"), + std::io::IsTerminal::is_terminal(&std::io::stderr()), + )) .init(); // Load config @@ -73,3 +77,29 @@ async fn main() { std::process::exit(e.exit_code()); } } + +/// Whether log lines on stderr carry ANSI colour: not when `--no-color` or a +/// non-empty `NO_COLOR` asks otherwise, nor when stderr is a file or pipe, +/// where the codes would land in a log as text. +fn log_colors( + no_color: bool, + no_color_env: Option, + stderr_is_terminal: bool, +) -> bool { + let no_color_env = no_color_env.is_some_and(|v| !v.is_empty()); + !no_color && !no_color_env && stderr_is_terminal +} + +#[cfg(test)] +mod tests { + use super::log_colors; + + #[test] + fn log_colors_only_on_a_terminal_without_a_no_color_request() { + assert!(log_colors(false, None, true)); + assert!(!log_colors(false, None, false)); + assert!(!log_colors(true, None, true)); + assert!(!log_colors(false, Some("1".into()), true)); + assert!(log_colors(false, Some("".into()), true)); + } +} diff --git a/src/models/subscription.rs b/src/models/subscription.rs index a3032f8..b7f0f1f 100644 --- a/src/models/subscription.rs +++ b/src/models/subscription.rs @@ -16,6 +16,8 @@ pub struct Subscription { pub expiration_date_time: Option, #[serde(skip_serializing_if = "Option::is_none")] pub client_state: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub lifecycle_notification_url: Option, } /// Request body for creating a new subscription. @@ -28,6 +30,8 @@ pub struct CreateSubscriptionRequest { pub expiration_date_time: String, #[serde(skip_serializing_if = "Option::is_none")] pub client_state: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub lifecycle_notification_url: Option, } /// Request body for renewing a subscription. @@ -60,6 +64,29 @@ pub struct ChangeNotification { pub tenant_id: Option, #[serde(skip_serializing_if = "Option::is_none")] pub client_state: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub lifecycle_event: Option, + /// Fields not modelled above, kept so output carries everything Graph sent. + #[serde(flatten)] + pub extra: serde_json::Map, +} + +impl ChangeNotification { + /// Output form: every field Graph sent except `clientState`, plus `kind` + /// (`lifecycle` when `lifecycleEvent` is present, else `change`). + pub fn to_output(&self) -> serde_json::Value { + let mut value = serde_json::to_value(self).unwrap_or_default(); + if let Some(map) = value.as_object_mut() { + map.remove("clientState"); + let kind = if self.lifecycle_event.is_some() { + "lifecycle" + } else { + "change" + }; + map.insert("kind".into(), kind.into()); + } + value + } } #[cfg(test)] @@ -75,6 +102,7 @@ mod tests { notification_url: Some("https://example.com/webhook".into()), expiration_date_time: Some("2026-03-15T00:00:00Z".into()), client_state: Some("my-secret".into()), + lifecycle_notification_url: None, }; let json = serde_json::to_string(&sub).unwrap(); let parsed: Subscription = serde_json::from_str(&json).unwrap(); @@ -90,11 +118,30 @@ mod tests { resource: "/teams/all/messages".into(), expiration_date_time: "2026-03-15T00:00:00Z".into(), client_state: Some("secret".into()), + lifecycle_notification_url: None, }; let json = serde_json::to_value(&req).unwrap(); assert_eq!(json["changeType"], "created"); assert_eq!(json["notificationUrl"], "https://example.com/webhook"); assert_eq!(json["clientState"], "secret"); + assert!(json.get("lifecycleNotificationUrl").is_none()); + } + + #[test] + fn create_subscription_request_sends_lifecycle_url() { + let req = CreateSubscriptionRequest { + change_type: "created".into(), + notification_url: "https://example.com/webhook".into(), + resource: "/teams/all/messages".into(), + expiration_date_time: "2026-03-15T00:00:00Z".into(), + client_state: None, + lifecycle_notification_url: Some("https://example.com/lifecycle".into()), + }; + let json = serde_json::to_value(&req).unwrap(); + assert_eq!( + json["lifecycleNotificationUrl"], + "https://example.com/lifecycle" + ); } #[test] @@ -111,4 +158,38 @@ mod tests { assert_eq!(notif.change_type.as_deref(), Some("created")); assert_eq!(notif.tenant_id.as_deref(), Some("tenant-1")); } + + #[test] + fn change_notification_output_keeps_unknown_fields_and_drops_client_state() { + let notif: ChangeNotification = serde_json::from_str( + r#"{ + "subscriptionId": "sub-1", + "changeType": "created", + "clientState": "my-secret", + "subscriptionExpirationDateTime": "2026-03-15T00:00:00Z", + "encryptedContent": {"data": "abc"} + }"#, + ) + .unwrap(); + let out = notif.to_output(); + assert_eq!(out["kind"], "change"); + assert_eq!(out["subscriptionId"], "sub-1"); + assert_eq!( + out["subscriptionExpirationDateTime"], + "2026-03-15T00:00:00Z" + ); + assert_eq!(out["encryptedContent"]["data"], "abc"); + assert!(out.get("clientState").is_none()); + } + + #[test] + fn lifecycle_notification_output_is_kind_lifecycle() { + let notif: ChangeNotification = serde_json::from_str( + r#"{"subscriptionId": "sub-1", "lifecycleEvent": "reauthorizationRequired"}"#, + ) + .unwrap(); + let out = notif.to_output(); + assert_eq!(out["kind"], "lifecycle"); + assert_eq!(out["lifecycleEvent"], "reauthorizationRequired"); + } } diff --git a/src/output/mod.rs b/src/output/mod.rs index e9bbd8e..4bc1d87 100644 --- a/src/output/mod.rs +++ b/src/output/mod.rs @@ -47,7 +47,7 @@ fn try_write_stdout(bytes: &[u8]) -> io::Result<()> { io::stdout().lock().write_all(bytes) } -fn try_write_stdout_line(line: &str) -> io::Result<()> { +pub fn try_write_stdout_line(line: &str) -> io::Result<()> { let stdout = io::stdout(); let mut stdout = stdout.lock(); stdout.write_all(line.as_bytes())?; diff --git a/tests/cli.rs b/tests/cli.rs index 8a4474a..f4577c4 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -1505,6 +1505,15 @@ fn subscribe_help_shows_subcommands() { ); } +#[test] +fn subscribe_create_help_shows_lifecycle_url() { + teams() + .args(["subscribe", "create", "--help"]) + .assert() + .success() + .stdout(predicate::str::contains("--lifecycle-url")); +} + #[test] fn listen_help_shows_options() { teams() diff --git a/tests/listen.rs b/tests/listen.rs new file mode 100644 index 0000000..e2a58cd --- /dev/null +++ b/tests/listen.rs @@ -0,0 +1,519 @@ +//! Black-box tests of `teams listen` over real TCP on 127.0.0.1. + +use std::io::{BufRead, BufReader, Read, Write}; +use std::net::{TcpListener, TcpStream}; +use std::process::{Child, ChildStdout, Command, Stdio}; +use std::sync::{mpsc, Arc, Mutex}; +use std::thread; +use std::time::{Duration, Instant}; + +use assert_cmd::prelude::*; +use predicates::prelude::*; + +const WAIT: Duration = Duration::from_secs(20); + +fn teams(home: &std::path::Path) -> Command { + let mut cmd = Command::new(assert_cmd::cargo::cargo_bin!("teams")); + cmd.env("HOME", home) + .env("XDG_CONFIG_HOME", home) + .env("APPDATA", home) + .env("TEAMS_CLI_DISABLE_KEYRING", "1") + .env("TEAMS_CLI_TOKEN_STORE", "file") + .env_remove("TEAMS_CLI_CLIENT_STATE") + .env_remove("TEAMS_CLI_PROFILE") + .env_remove("TEAMS_CLI_ACCESS_TOKEN") + .env_remove("RUST_LOG"); + cmd +} + +/// A running `teams listen --port 0`, killed on drop. +struct Listener { + child: Child, + port: u16, + stdout: Option>, + /// Held unread when the test stalls the consumer. + #[cfg_attr(not(unix), allow(dead_code))] + raw_stdout: Option, + stderr: Arc>, + _home: tempfile::TempDir, +} + +impl Drop for Listener { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +impl Listener { + fn start(args: &[&str], envs: &[(&str, &str)]) -> Self { + Self::spawn(args, envs, true) + } + + fn spawn(args: &[&str], envs: &[(&str, &str)], read_stdout: bool) -> Self { + let home = tempfile::tempdir().unwrap(); + let mut cmd = teams(home.path()); + cmd.args(["listen", "--port", "0"]).args(args); + for (k, v) in envs { + cmd.env(k, v); + } + let mut child = cmd + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + + // Drain stderr continuously so the process never blocks on it. + let stderr = Arc::new(Mutex::new(String::new())); + let (port_tx, port_rx) = mpsc::channel(); + { + let stderr = stderr.clone(); + let pipe = child.stderr.take().unwrap(); + thread::spawn(move || { + for line in BufReader::new(pipe).lines() { + let Ok(line) = line else { break }; + { + let mut all = stderr.lock().unwrap(); + all.push_str(&line); + all.push('\n'); + } + if let Some(rest) = line.split("started on http://").nth(1) { + let port = rest.rsplit(':').next().unwrap().trim().parse::(); + let _ = port_tx.send(port.unwrap()); + } + } + }); + } + + let (stdout, raw_stdout) = if read_stdout { + let (tx, rx) = mpsc::channel(); + let pipe = child.stdout.take().unwrap(); + thread::spawn(move || { + for line in BufReader::new(pipe).lines() { + let Ok(line) = line else { break }; + if tx.send(line).is_err() { + break; + } + } + }); + (Some(rx), None) + } else { + (None, child.stdout.take()) + }; + + let port = port_rx + .recv_timeout(WAIT) + .expect("listener did not report its address"); + Listener { + child, + port, + stdout, + raw_stdout, + stderr, + _home: home, + } + } + + fn next_line(&self) -> serde_json::Value { + let line = self + .stdout + .as_ref() + .unwrap() + .recv_timeout(WAIT) + .expect("no notification printed"); + serde_json::from_str(&line).unwrap() + } + + fn no_more_lines(&self) { + let got = self + .stdout + .as_ref() + .unwrap() + .recv_timeout(Duration::from_millis(500)); + assert!(got.is_err(), "unexpected output: {got:?}"); + } + + fn stderr_eventually_contains(&self, needle: &str) -> String { + let deadline = Instant::now() + WAIT; + loop { + let text = self.stderr.lock().unwrap().clone(); + if text.contains(needle) || Instant::now() > deadline { + return text; + } + thread::sleep(Duration::from_millis(50)); + } + } + + fn assert_alive(&mut self) { + assert!(self.child.try_wait().unwrap().is_none(), "listener exited"); + } + + fn connect(&self) -> TcpStream { + let stream = TcpStream::connect(("127.0.0.1", self.port)).unwrap(); + stream.set_read_timeout(Some(WAIT)).unwrap(); + stream + } + + /// Send a raw request and return (status, body). + fn raw(&self, request: &[u8]) -> (u16, String) { + let mut stream = self.connect(); + if let Err(e) = stream.write_all(request) { + // Seen a handful of times locally under heavy load (ENOTCONN right + // after connect) and never reproduced; say what the listener said. + panic!( + "sending to the listener on port {} failed: {e}; its stderr:\n{}", + self.port, + self.stderr.lock().unwrap() + ); + } + read_response(&mut stream) + } + + fn post(&self, target: &str, body: &str) -> (u16, String) { + self.raw(post_request(target, body).as_bytes()) + } +} + +fn post_request(target: &str, body: &str) -> String { + format!( + "POST {target} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\n\ + Content-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ) +} + +fn read_response(stream: &mut TcpStream) -> (u16, String) { + let mut buf = Vec::new(); + let _ = stream.read_to_end(&mut buf); + let text = String::from_utf8_lossy(&buf).into_owned(); + let status = text + .split(' ') + .nth(1) + .and_then(|s| s.parse().ok()) + .unwrap_or_else(|| panic!("no HTTP response: {text:?}")); + let body = text.split("\r\n\r\n").nth(1).unwrap_or("").to_string(); + (status, body) +} + +const VALID: &str = r#"{"value":[{"subscriptionId":"sub-ok","changeType":"created","resource":"chats/1/messages/2","clientState":"s3cret-value","tenantId":"t","subscriptionExpirationDateTime":"2026-10-12T00:00:00Z"}]}"#; + +#[test] +fn client_state_filters_forged_notifications() { + let listener = Listener::start(&[], &[("TEAMS_CLI_CLIENT_STATE", "s3cret-value")]); + + let forged = r#"{"value":[{"subscriptionId":"sub-forged","changeType":"created","clientState":"guessed-value"}]}"#; + let missing = r#"{"value":[{"subscriptionId":"sub-missing","changeType":"created"}]}"#; + assert_eq!(listener.post("/", forged).0, 202); + assert_eq!(listener.post("/webhook", missing).0, 202); + assert_eq!(listener.post("/", VALID).0, 202); + + // Only the valid notification is printed, with kind and without clientState. + let line = listener.next_line(); + assert_eq!(line["subscriptionId"], "sub-ok"); + assert_eq!(line["kind"], "change"); + assert_eq!( + line["subscriptionExpirationDateTime"], + "2026-10-12T00:00:00Z" + ); + assert!(line.get("clientState").is_none(), "{line}"); + listener.no_more_lines(); + + let stderr = listener.stderr_eventually_contains("clientState missing or wrong"); + assert!(stderr.contains("clientState missing or wrong"), "{stderr}"); + // stderr is a pipe here, so log lines carry no colour codes. + assert!(!stderr.contains('\u{1b}'), "{stderr:?}"); + assert!(!stderr.contains("s3cret-value"), "{stderr}"); + assert!(!stderr.contains("guessed-value"), "{stderr}"); + assert!(!stderr.contains("not authenticated"), "{stderr}"); +} + +#[test] +fn lifecycle_notifications_are_printed_and_unauthenticated_mode_warns() { + let listener = Listener::start(&[], &[]); + let stderr = listener.stderr_eventually_contains("not authenticated"); + assert!(stderr.contains("http://127.0.0.1:"), "{stderr}"); + assert!(stderr.contains("not authenticated"), "{stderr}"); + + let body = r#"{"value":[ + {"subscriptionId":"s","lifecycleEvent":"reauthorizationRequired","clientState":"x","organizationId":"o"}, + {"subscriptionId":"s","lifecycleEvent":"subscriptionRemoved"}, + {"subscriptionId":"s","lifecycleEvent":"missed"} + ]}"#; + assert_eq!(listener.post("/", body).0, 202); + for event in ["reauthorizationRequired", "subscriptionRemoved", "missed"] { + let line = listener.next_line(); + assert_eq!(line["kind"], "lifecycle"); + assert_eq!(line["lifecycleEvent"], event); + assert!(line.get("clientState").is_none(), "{line}"); + } +} + +#[test] +fn lifecycle_url_path_is_validated_and_served() { + // Graph validates --lifecycle-url as well as --webhook-url when a + // subscription is created, and the README suggests a /lifecycle path. + let listener = Listener::start(&[], &[("TEAMS_CLI_CLIENT_STATE", "s3cret-value")]); + + let (status, body) = listener.post("/lifecycle?validationToken=a%20b%2Bc", ""); + assert_eq!((status, body.as_str()), (200, "a b+c")); + + let event = r#"{"value":[{"subscriptionId":"s","lifecycleEvent":"reauthorizationRequired","clientState":"s3cret-value"}]}"#; + assert_eq!(listener.post("/lifecycle", event).0, 202); + let line = listener.next_line(); + assert_eq!(line["kind"], "lifecycle"); + assert_eq!(line["lifecycleEvent"], "reauthorizationRequired"); + listener.no_more_lines(); +} + +#[test] +fn oversized_body_is_413_and_listener_survives() { + let mut listener = Listener::start(&[], &[]); + let size = 5 * 1024 * 1024; + + // Announced size: answered before any body byte is sent. + let head = format!( + "POST / HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: {size}\r\n\ + Expect: 100-continue\r\nConnection: close\r\n\r\n" + ); + assert_eq!(listener.raw(head.as_bytes()).0, 413); + + // A real 5 MiB body. The server may reset the connection once it has + // answered, so a reset is accepted here; what matters is what follows. + let mut stream = listener.connect(); + let mut writer = stream.try_clone().unwrap(); + let request = post_request("/", &format!(r#"{{"value":["{}"]}}"#, "x".repeat(size))); + let sender = thread::spawn(move || { + let _ = writer.write_all(request.as_bytes()); + }); + let mut buf = [0u8; 12]; + if stream.read_exact(&mut buf).is_ok() { + assert_eq!(&buf, b"HTTP/1.1 413"); + } + drop(stream); + sender.join().unwrap(); + + listener.assert_alive(); + assert_eq!(listener.post("/", VALID).0, 202); + assert_eq!(listener.next_line()["subscriptionId"], "sub-ok"); +} + +#[test] +fn stalled_connections_do_not_block_requests() { + let mut listener = Listener::start(&[], &[]); + + // Fewer than the 64 connection slots: idle sockets, unfinished headers, + // and complete headers with an unfinished body. + let mut stalled = Vec::new(); + for _ in 0..30 { + stalled.push(listener.connect()); + } + let mut partial_head = Vec::new(); + for _ in 0..10 { + let mut s = listener.connect(); + s.write_all(b"POST / HTTP/1.1\r\nHost: 127.0.0.1\r\n") + .unwrap(); + partial_head.push(s); + } + let mut partial_body = Vec::new(); + for _ in 0..10 { + let mut s = listener.connect(); + s.write_all( + b"POST / HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: 1000\r\n\r\n{\"value\"", + ) + .unwrap(); + partial_body.push(s); + } + + let started = Instant::now(); + assert_eq!(listener.post("/", VALID).0, 202); + assert_eq!( + listener.post("/webhook?validationToken=abc%20123", ""), + (200, "abc 123".to_string()) + ); + assert!( + started.elapsed() < Duration::from_secs(5), + "{:?}", + started.elapsed() + ); + assert_eq!(listener.next_line()["subscriptionId"], "sub-ok"); + + // The deadlines then close the stalled ones. + let (status, _) = read_response(&mut partial_body[0]); + assert_eq!(status, 408); + let mut rest = Vec::new(); + let closed = partial_head[0].read_to_end(&mut rest); + assert!(closed.is_ok(), "header deadline did not close: {closed:?}"); + listener.assert_alive(); +} + +#[test] +fn more_stalled_connections_than_slots_only_delay_requests() { + let listener = Listener::start(&[], &[]); + let stalled: Vec<_> = (0..70).map(|_| listener.connect()).collect(); + + // Slots free when the header deadline closes the idle sockets. + let mut stream = TcpStream::connect(("127.0.0.1", listener.port)).unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(40))) + .unwrap(); + stream + .write_all(post_request("/?validationToken=t", "").as_bytes()) + .unwrap(); + assert_eq!(read_response(&mut stream), (200, "t".to_string())); + drop(stalled); +} + +#[test] +fn stalled_stdout_answers_503_and_validation_still_works() { + let listener = Listener::spawn(&[], &[], false); + let pad = "x".repeat(900 * 1024); + let body = + format!(r#"{{"value":[{{"subscriptionId":"s","resourceData":{{"pad":"{pad}"}}}}]}}"#); + + // Pipe buffer plus the 8 MiB queue hold at most about ten of these. + let mut statuses = Vec::new(); + for _ in 0..40 { + let started = Instant::now(); + let (status, _) = listener.post("/", &body); + assert!(started.elapsed() < Duration::from_secs(10), "request hung"); + statuses.push(status); + if status == 503 { + break; + } + } + assert_eq!(statuses.last(), Some(&503), "{statuses:?}"); + assert!(statuses[..statuses.len() - 1].iter().all(|s| *s == 202)); + + assert_eq!( + listener.post("/?validationToken=still-here", ""), + (200, "still-here".to_string()) + ); +} + +#[test] +fn port_in_use_is_a_config_error_naming_the_address() { + let taken = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = taken.local_addr().unwrap().port(); + let home = tempfile::tempdir().unwrap(); + let output = assert_cmd::Command::from_std(teams(home.path())) + .args(["listen", "--port", &port.to_string()]) + .timeout(WAIT) + .output() + .unwrap(); + let text = format!( + "{}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert_eq!(output.status.code(), Some(10), "{text}"); + assert!(text.contains(&format!("127.0.0.1:{port}")), "{text}"); + assert!(text.contains("CONFIG_ERROR"), "{text}"); +} + +#[test] +fn listen_help_documents_flags_without_leaking_client_state() { + let home = tempfile::tempdir().unwrap(); + teams(home.path()) + .args(["listen", "--help"]) + .env("TEAMS_CLI_CLIENT_STATE", "sentinel-client-state") + .assert() + .success() + .stdout(predicate::str::contains("--bind")) + .stdout(predicate::str::contains("127.0.0.1")) + .stdout(predicate::str::contains("TEAMS_CLI_CLIENT_STATE")) + .stdout(predicate::str::contains("sentinel-client-state").not()); +} + +/// Fill the stdout pipe and queue more behind it; returns the ids acknowledged. +#[cfg(unix)] +fn post_backlog(listener: &Listener) -> Vec { + let pad = "x".repeat(200 * 1024); + (0..20) + .map(|i| { + let id = format!("n{i}"); + let body = format!( + r#"{{"value":[{{"subscriptionId":"{id}","resourceData":{{"pad":"{pad}"}}}}]}}"# + ); + assert_eq!(listener.post("/", &body).0, 202); + id + }) + .collect() +} + +#[cfg(unix)] +fn send_signal(listener: &Listener, signal: &str) { + let status = Command::new("kill") + .args([signal, &listener.child.id().to_string()]) + .status() + .unwrap(); + assert!(status.success()); +} + +#[cfg(unix)] +fn wait_exit(listener: &mut Listener) -> std::process::ExitStatus { + let deadline = Instant::now() + WAIT; + loop { + if let Some(status) = listener.child.try_wait().unwrap() { + return status; + } + assert!(Instant::now() < deadline, "listener did not exit"); + thread::sleep(Duration::from_millis(50)); + } +} + +// Signals: Unix only. The queue logic itself is unit-tested on every OS. +#[cfg(unix)] +#[test] +fn ctrl_c_writes_out_every_acknowledged_notification() { + let mut listener = Listener::spawn(&[], &[], false); + let acked = post_backlog(&listener); + + send_signal(&listener, "-INT"); + listener.stderr_eventually_contains("Shutting down"); + // No longer admitting: the port closes. + let deadline = Instant::now() + WAIT; + while TcpStream::connect(("127.0.0.1", listener.port)).is_ok() { + assert!(Instant::now() < deadline, "still accepting after Ctrl+C"); + thread::sleep(Duration::from_millis(50)); + } + + // The reader resumes and gets everything that was acknowledged. + let stdout = listener.raw_stdout.take().unwrap(); + let printed: Vec = BufReader::new(stdout) + .lines() + .map(|l| { + let v: serde_json::Value = serde_json::from_str(&l.unwrap()).unwrap(); + v["subscriptionId"].as_str().unwrap().to_string() + }) + .collect(); + assert_eq!(printed, acked); + let status = wait_exit(&mut listener); + assert_eq!( + status.code(), + Some(0), + "{}", + listener.stderr.lock().unwrap() + ); +} + +#[cfg(unix)] +#[test] +fn second_signal_abandons_the_queue_and_reports_the_loss() { + let mut listener = Listener::spawn(&[], &[], false); + post_backlog(&listener); + + send_signal(&listener, "-INT"); + listener.stderr_eventually_contains("Shutting down"); + send_signal(&listener, "-TERM"); + let status = wait_exit(&mut listener); + let stderr = listener.stderr_eventually_contains("were not written"); + assert_eq!(status.code(), Some(1), "{stderr}"); + let lost: u64 = stderr + .split("Error: ") + .nth(1) + .and_then(|s| s.split(' ').next()) + .and_then(|n| n.parse().ok()) + .unwrap_or_else(|| panic!("no lost count: {stderr}")); + assert!(lost > 0 && lost <= 20, "{stderr}"); +}