From d11241019507f7c42cf7b5fe133d6bd0185a8cd4 Mon Sep 17 00:00:00 2001 From: Abraham Ingersoll <586805+aberoham@users.noreply.github.com> Date: Sun, 11 Oct 2026 16:44:41 +0100 Subject: [PATCH 1/7] Add --lifecycle-url to subscribe create Graph rejects Teams subscriptions longer than one hour unless the create request carries lifecycleNotificationUrl, and subscribe create had no way to set it, so any --expiration past one hour failed. Add --lifecycle-url, sent as lifecycleNotificationUrl and omitted when not given. The Subscription model now reads the field back so create output shows it. No client-side expiration limit: Graph's 4xx body already reaches the user verbatim through ApiError. Tests: serialization with and without the field, and a help test for the flag. Both failed before the change. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Hd5XXx9UkpLZpAAkvPPyUg --- CHANGELOG.md | 1 + README.md | 1 + docs/command-reference.md | 2 +- docs/man/teams.1 | 2 +- src/cli/subscribe.rs | 5 +++++ src/models/subscription.rs | 24 ++++++++++++++++++++++++ tests/cli.rs | 9 +++++++++ 7 files changed, 42 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c48b53..e9b21e8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ ### 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. ### Changed diff --git a/README.md b/README.md index a3ec034..667507c 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 diff --git a/docs/command-reference.md b/docs/command-reference.md index 7053881..85be50c 100644 --- a/docs/command-reference.md +++ b/docs/command-reference.md @@ -282,7 +282,7 @@ 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 diff --git a/docs/man/teams.1 b/docs/man/teams.1 index 0eb5a93..9ef34c4 100644 --- a/docs/man/teams.1 +++ b/docs/man/teams.1 @@ -375,7 +375,7 @@ 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 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/models/subscription.rs b/src/models/subscription.rs index a3032f8..d1802f3 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. @@ -75,6 +79,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 +95,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] 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() From c1461183ff170fdea2ea9e782cfb951d592d2ace Mon Sep 17 00:00:00 2001 From: Abraham Ingersoll <586805+aberoham@users.noreply.github.com> Date: Sun, 11 Oct 2026 16:56:40 +0100 Subject: [PATCH 2/7] Harden teams listen: loopback bind, clientState, limits teams listen bound 0.0.0.0 and printed any POSTed JSON, so anyone who could reach the port could inject notifications. It had no body size or read deadline, unlimited connections, and wrote to stdout before answering, so a slow reader stalled Graph's request. Output dropped fields Graph sends (lifecycleEvent, subscriptionExpirationDateTime) but printed clientState. A port in use exited 3 as an auth error. - Bind 127.0.0.1 by default; --bind chooses another address, and --port 0 picks a free one. - --client-state / TEAMS_CLI_CLIENT_STATE drops notifications with another or no clientState (SHA-256 then constant-time compare), with a warning that names neither value. Dropped items still get 202: an error only makes Graph resend them. Without it, warn at startup. - 1 MiB body cap (413), 10s header and body deadlines (408), at most 64 connections; more wait in the backlog. - Output goes through an 8 MiB queue to a writer thread. A batch is queued whole and answered 202, or 503 when full. Validation never touches the queue. Ctrl+C waits up to 2s for queued output. - Keep unknown fields via serde flatten, add kind (lifecycle/change), strip clientState. - Bind failure is ConfigError (exit 10) naming the address. - README "Running behind a relay", man page, changelog. Tests: unit tests for the handler, queue and model; tests/listen.rs runs the binary on 127.0.0.1:0 over real TCP: forged/missing clientState not printed, lifecycle kinds, 5 MiB body 413 then alive, stalled connections not blocking requests, a stalled stdout giving 503 while validation still works, port in use exiting 10. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Hd5XXx9UkpLZpAAkvPPyUg --- CHANGELOG.md | 14 ++ CLAUDE.md | 7 +- README.md | 29 ++- docs/command-reference.md | 4 +- docs/man/teams.1 | 37 +++- src/cli/listen.rs | 7 +- src/cli/mod.rs | 15 +- src/listen/handler.rs | 275 +++++++++++++++++++++---- src/listen/mod.rs | 63 ++++-- src/listen/queue.rs | 98 +++++++++ src/models/subscription.rs | 57 ++++++ tests/listen.rs | 398 +++++++++++++++++++++++++++++++++++++ 12 files changed, 940 insertions(+), 64 deletions(-) create mode 100644 src/listen/queue.rs create mode 100644 tests/listen.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index e9b21e8..53883fd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,20 @@ - `--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. +- `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. +- `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`. + +### Fixed + +- `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). ### Changed diff --git a/CLAUDE.md b/CLAUDE.md index 4611c1e..2e4632c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -85,9 +85,11 @@ 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 +- `teams listen --port 8080` starts a hyper HTTP server on 127.0.0.1 (`--bind` to change) - Handles Graph subscription validation (echoes `?validationToken`) -- Outputs change notifications as NDJSON to stdout +- 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 - Multi-connection, graceful Ctrl+C shutdown - Requires HTTPS via reverse proxy (ngrok) for production use @@ -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 667507c..b459016 100644 --- a/README.md +++ b/README.md @@ -626,15 +626,41 @@ 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). 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 `/` or `/webhook`. +- 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) | Retry with backoff | + +At most 64 connections are served at once; others wait. Headers must arrive within 10 seconds. + ### User Lookup ```bash @@ -732,6 +758,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 85be50c..1f09a8b 100644 --- a/docs/command-reference.md +++ b/docs/command-reference.md @@ -286,10 +286,10 @@ teams subscribe create --resource RESOURCE --change-type created,updated,deleted 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 9ef34c4..59a0f45 100644 --- a/docs/man/teams.1 +++ b/docs/man/teams.1 @@ -379,12 +379,39 @@ teams subscribe create --resource RESOURCE --change-type TYPES --webhook-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 / or /webhook, 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 (retry with backoff). At most 64 connections are +served at once, and headers must arrive within 10 seconds. +A port already in use exits with code 10. .SH COMPLETIONS .nf teams completions bash @@ -406,6 +433,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/listen/handler.rs b/src/listen/handler.rs index f691196..baf424f 100644 --- a/src/listen/handler.rs +++ b/src/listen/handler.rs @@ -1,58 +1,144 @@ -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::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() +} -pub async fn handle_request(req: Request) -> HandlerResult { +/// 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, 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 + // 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); - } - Ok(ok_response("")) + 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()); } - Err(e) => { - tracing::warn!("Failed to parse notification: {e}"); - Ok(plain_response( + Ok(Err(e)) => { + return Ok(plain_response( StatusCode::BAD_REQUEST, - &format!("Invalid notification body: {e}"), + &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}"), + )); + } + }; + + 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; + } + 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() && !state.queue.try_push(lines.join("\n")) { + tracing::warn!("Output queue full; answering 503"); + return Ok(plain_response( + StatusCode::SERVICE_UNAVAILABLE, + "Output queue full", + )); + } + Ok(plain_response(StatusCode::ACCEPTED, "")) } _ => Ok(plain_response(StatusCode::NOT_FOUND, "Not Found")), } } -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 +152,129 @@ 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()); + }); + (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 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 full_queue_answers_503() { + // Smaller than one line, so the push is refused. + let queue = Queue::new(4, |_| {}); + 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..27d8254 100644 --- a/src/listen/mod.rs +++ b/src/listen/mod.rs @@ -1,39 +1,77 @@ 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); +/// Time allowed on Ctrl+C to write output already accepted. +const SHUTDOWN_FLUSH: Duration = Duration::from_secs(2); + +/// 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 with JSON body — change notifications, printed as NDJSON to stdout +/// by a writer thread; 202 once queued, 503 when the queue is full +/// +/// 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::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"); 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; }); } @@ -44,6 +82,7 @@ pub async fn run_listener(port: u16) -> Result<()> { } _ = tokio::signal::ctrl_c() => { eprintln!("\nShutting down webhook listener"); + state.queue().flush(SHUTDOWN_FLUSH).await; return Ok(()); } } diff --git a/src/listen/queue.rs b/src/listen/queue.rs new file mode 100644 index 0000000..49836b3 --- /dev/null +++ b/src/listen/queue.rs @@ -0,0 +1,98 @@ +use std::sync::Arc; +use std::time::Duration; + +use tokio::sync::{mpsc, OwnedSemaphorePermit, Semaphore}; + +/// Bytes of output that may wait for stdout before requests get 503. +pub const OUTPUT_BUDGET: usize = 8 * 1024 * 1024; + +/// Output waiting for stdout, bounded in bytes, so a slow reader never +/// blocks an HTTP response. A dedicated thread writes entries in order. +pub struct Queue { + tx: mpsc::UnboundedSender<(String, OwnedSemaphorePermit)>, + budget: Arc, + capacity: usize, +} + +impl Queue { + pub fn new(capacity: usize, mut sink: impl FnMut(&str) + Send + 'static) -> Self { + let (tx, mut rx) = mpsc::unbounded_channel::<(String, OwnedSemaphorePermit)>(); + std::thread::Builder::new() + .name("teams-listen-output".into()) + .spawn(move || { + // The permit is held until the write returns. + while let Some((text, _permit)) = rx.blocking_recv() { + sink(&text); + } + }) + .expect("spawn output thread"); + Self { + tx, + budget: Arc::new(Semaphore::new(capacity)), + capacity, + } + } + + /// Enqueue `text` whole, or return false when it does not fit. + pub fn try_push(&self, text: String) -> bool { + let Ok(n) = u32::try_from(text.len().max(1)) else { + return false; + }; + match self.budget.clone().try_acquire_many_owned(n) { + Ok(permit) => self.tx.send((text, permit)).is_ok(), + Err(_) => false, + } + } + + /// Wait up to `limit` for queued output to be written. + pub async fn flush(&self, limit: Duration) { + let _ = tokio::time::timeout(limit, async { + while self.budget.available_permits() < self.capacity { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::mpsc as std_mpsc; + + #[tokio::test] + async fn full_queue_refuses_until_the_writer_catches_up() { + let (release_tx, release_rx) = std_mpsc::channel::<()>(); + let (out_tx, out_rx) = std_mpsc::channel::(); + let queue = Queue::new(10, move |text| { + release_rx.recv().unwrap(); + out_tx.send(text.to_string()).unwrap(); + }); + + // The writer takes the first entry and blocks in the sink holding its + // bytes; the second fills the budget. + assert!(queue.try_push("aaaaa".into())); + assert!(queue.try_push("bbbbb".into())); + assert!(!queue.try_push("c".into())); + + release_tx.send(()).unwrap(); + assert_eq!(out_rx.recv().unwrap(), "aaaaa"); + // Space frees once the write returns. + let deadline = std::time::Instant::now() + Duration::from_secs(10); + while !queue.try_push("c".into()) { + assert!(std::time::Instant::now() < deadline, "space never freed"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + release_tx.send(()).unwrap(); + release_tx.send(()).unwrap(); + assert_eq!(out_rx.recv().unwrap(), "bbbbb"); + assert_eq!(out_rx.recv().unwrap(), "c"); + } + + #[test] + fn entry_larger_than_the_budget_is_refused() { + let queue = Queue::new(4, |_| {}); + assert!(!queue.try_push("12345".into())); + assert!(queue.try_push("1234".into())); + } +} diff --git a/src/models/subscription.rs b/src/models/subscription.rs index d1802f3..b7f0f1f 100644 --- a/src/models/subscription.rs +++ b/src/models/subscription.rs @@ -64,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)] @@ -135,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/tests/listen.rs b/tests/listen.rs new file mode 100644 index 0000000..316f2ca --- /dev/null +++ b/tests/listen.rs @@ -0,0 +1,398 @@ +//! 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. + _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: 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(); + stream.write_all(request).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}"); + 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 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()); +} From f2e53ae2acb788eb0f83bf9f90594ad68b0a91c4 Mon Sep 17 00:00:00 2001 From: Abraham Ingersoll <586805+aberoham@users.noreply.github.com> Date: Sun, 11 Oct 2026 17:10:03 +0100 Subject: [PATCH 3/7] Write out acknowledged listen output before exiting A notification answered 202 could still be waiting in the stdout queue. On Ctrl+C the listener waited 2s, ignored whether the queue had drained, and exited 0, so with a slow reader acknowledged notifications were lost silently; handlers could also keep enqueueing meanwhile. A closed stdout exited 0 from the writer thread, losing the rest too. On Ctrl+C or SIGTERM the listener now stops accepting connections and closes queue admission under the queue's lock, so handlers past admission finish and later ones answer 503. It then waits for the writer with no time limit. A second signal abandons the queue. The writer counts lines written; if any acknowledged line is not written (second signal, or a stdout write error, which also closes admission), the count goes to stderr and the process exits 1 directly. Exit 1 (general) because no documented category fits a local output failure, and the JSON error envelope would go to the stdout that failed. Tests: queue unit tests for closed admission, full drain, aborted drain and stdout failure counts; a handler test for 503 while shutting down. Unix-only binary tests: SIGINT with a stalled reader, then reading prints every acknowledged notification and exits 0; SIGINT then SIGTERM exits 1 with the lost count. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Hd5XXx9UkpLZpAAkvPPyUg --- CHANGELOG.md | 2 +- CLAUDE.md | 2 +- README.md | 4 +- docs/man/teams.1 | 9 +- src/listen/handler.rs | 41 +++++-- src/listen/mod.rs | 63 +++++++++-- src/listen/queue.rs | 244 ++++++++++++++++++++++++++++++++++-------- src/output/mod.rs | 2 +- tests/listen.rs | 98 ++++++++++++++++- 9 files changed, 399 insertions(+), 66 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 53883fd..f0ece2c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,7 +13,7 @@ ### 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. +- `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`. diff --git a/CLAUDE.md b/CLAUDE.md index 2e4632c..a7ae018 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -90,7 +90,7 @@ CLI flags > env vars (TEAMS_CLI_CLIENT_ID, TEAMS_CLI_CLIENT_SECRET, TEAMS_CLI_TE - 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 -- Multi-connection, graceful Ctrl+C shutdown +- 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 diff --git a/README.md b/README.md index b459016..e7ed00b 100644 --- a/README.md +++ b/README.md @@ -657,10 +657,12 @@ Responses: | 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) | Retry with backoff | +| 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 diff --git a/docs/man/teams.1 b/docs/man/teams.1 index 59a0f45..55a51c0 100644 --- a/docs/man/teams.1 +++ b/docs/man/teams.1 @@ -409,8 +409,13 @@ to / or /webhook, 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 (retry with backoff). At most 64 connections are -served at once, and headers must arrive within 10 seconds. +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 diff --git a/src/listen/handler.rs b/src/listen/handler.rs index baf424f..331582a 100644 --- a/src/listen/handler.rs +++ b/src/listen/handler.rs @@ -7,7 +7,7 @@ use hyper::body::{Body, Bytes}; use hyper::{Method, Request, Response, StatusCode}; use sha2::{Digest, Sha256}; -use super::queue::Queue; +use super::queue::{Push, Queue}; use crate::models::subscription::ChangeNotificationCollection; /// Largest notification body accepted. @@ -123,14 +123,23 @@ where // 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() && !state.queue.try_push(lines.join("\n")) { - tracing::warn!("Output queue full; answering 503"); - return Ok(plain_response( + 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::SERVICE_UNAVAILABLE, + "Output queue full", + )) + } + Push::Closed => Ok(plain_response( StatusCode::SERVICE_UNAVAILABLE, - "Output queue full", - )); + "Shutting down", + )), } - Ok(plain_response(StatusCode::ACCEPTED, "")) } _ => Ok(plain_response(StatusCode::NOT_FOUND, "Not Found")), @@ -159,6 +168,7 @@ mod tests { 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) } @@ -256,10 +266,25 @@ mod tests { 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, |_| {}); + 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 diff --git a/src/listen/mod.rs b/src/listen/mod.rs index 27d8254..34957bf 100644 --- a/src/listen/mod.rs +++ b/src/listen/mod.rs @@ -17,8 +17,6 @@ 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); -/// Time allowed on Ctrl+C to write output already accepted. -const SHUTDOWN_FLUSH: Duration = Duration::from_secs(2); /// Run the webhook listener HTTP server on `bind:port`. /// @@ -28,6 +26,11 @@ const SHUTDOWN_FLUSH: Duration = Duration::from_secs(2); /// - POST with JSON body — change 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. /// @@ -40,7 +43,7 @@ pub async fn run_listener(bind: IpAddr, port: u16, client_state: Option) .map_err(|e| TeamsError::ConfigError(format!("cannot listen on {addr}: {e}")))?; let local = listener.local_addr().unwrap_or(addr); - let queue = queue::Queue::new(queue::OUTPUT_BUDGET, output::write_stdout_line); + 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)); @@ -52,6 +55,7 @@ pub async fn run_listener(bind: IpAddr, port: u16, client_state: Option) } eprintln!("Press Ctrl+C to stop"); + let mut signals = Signals::new()?; loop { tokio::select! { (slot, accept) = async { @@ -80,11 +84,56 @@ pub async fn run_listener(bind: IpAddr, port: u16, client_state: Option) } } } - _ = tokio::signal::ctrl_c() => { - eprintln!("\nShutting down webhook listener"); - state.queue().flush(SHUTDOWN_FLUSH).await; - 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 index 49836b3..f0e58fb 100644 --- a/src/listen/queue.rs +++ b/src/listen/queue.rs @@ -1,57 +1,148 @@ -use std::sync::Arc; +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, OwnedSemaphorePermit, Semaphore}; +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 { - tx: mpsc::UnboundedSender<(String, OwnedSemaphorePermit)>, + /// `None` once admission is closed. The lock makes closing atomic with + /// respect to a push in progress. + tx: Mutex>>, budget: Arc, - capacity: usize, + 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) + Send + 'static) -> Self { - let (tx, mut rx) = mpsc::unbounded_channel::<(String, OwnedSemaphorePermit)>(); + 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 write returns. - while let Some((text, _permit)) = rx.blocking_recv() { - sink(&text); + // 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, + tx: Mutex::new(Some(tx)), budget: Arc::new(Semaphore::new(capacity)), - capacity, + shared, } } - /// Enqueue `text` whole, or return false when it does not fit. - pub fn try_push(&self, text: String) -> bool { - let Ok(n) = u32::try_from(text.len().max(1)) else { - return false; + /// 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; }; - match self.budget.clone().try_acquire_many_owned(n) { - Ok(permit) => self.tx.send((text, permit)).is_ok(), - Err(_) => false, + 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(); } - /// Wait up to `limit` for queued output to be written. - pub async fn flush(&self, limit: Duration) { - let _ = tokio::time::timeout(limit, async { - while self.budget.available_permits() < self.capacity { - tokio::time::sleep(Duration::from_millis(20)).await; - } - }) - .await; + /// 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() } } @@ -60,39 +151,106 @@ mod tests { use super::*; use std::sync::mpsc as std_mpsc; - #[tokio::test] - async fn full_queue_refuses_until_the_writer_catches_up() { + 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(10, move |text| { - release_rx.recv().unwrap(); + 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!(queue.try_push("aaaaa".into())); - assert!(queue.try_push("bbbbb".into())); - assert!(!queue.try_push("c".into())); + 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_tx.send(()).unwrap(); - assert_eq!(out_rx.recv().unwrap(), "aaaaa"); + 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("c".into()) { + 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_tx.send(()).unwrap(); - release_tx.send(()).unwrap(); - assert_eq!(out_rx.recv().unwrap(), "bbbbb"); - assert_eq!(out_rx.recv().unwrap(), "c"); + 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(4, |_| {}); - assert!(!queue.try_push("12345".into())); - assert!(queue.try_push("1234".into())); + 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/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/listen.rs b/tests/listen.rs index 316f2ca..59a2178 100644 --- a/tests/listen.rs +++ b/tests/listen.rs @@ -32,7 +32,8 @@ struct Listener { port: u16, stdout: Option>, /// Held unread when the test stalls the consumer. - _raw_stdout: Option, + #[cfg_attr(not(unix), allow(dead_code))] + raw_stdout: Option, stderr: Arc>, _home: tempfile::TempDir, } @@ -108,7 +109,7 @@ impl Listener { child, port, stdout, - _raw_stdout: raw_stdout, + raw_stdout, stderr, _home: home, } @@ -396,3 +397,96 @@ fn listen_help_documents_flags_without_leaking_client_state() { .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}"); +} From 2f3b9c441e76b29086a1ecfbda57ec2d82968c77 Mon Sep 17 00:00:00 2001 From: Abraham Ingersoll <586805+aberoham@users.noreply.github.com> Date: Sun, 11 Oct 2026 19:18:48 +0100 Subject: [PATCH 4/7] Colour log lines only on a terminal Logging coloured stderr unless --no-color was given, so a redirected log, `teams listen 2>listen.log` above all, filled with ANSI escape codes. Colour now also requires stderr to be a terminal and NO_COLOR to be unset or empty (no-color.org); --no-color still turns it off. Tests: the rule as a unit test; the listen test that reads a dropped notification's warning from piped stderr now finds no escape codes, which fails against the previous setting. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Hd5XXx9UkpLZpAAkvPPyUg --- CHANGELOG.md | 1 + src/main.rs | 32 +++++++++++++++++++++++++++++++- tests/listen.rs | 2 ++ 3 files changed, 34 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f0ece2c..7620c7c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,7 @@ ### 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). ### Changed 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/tests/listen.rs b/tests/listen.rs index 59a2178..6166da6 100644 --- a/tests/listen.rs +++ b/tests/listen.rs @@ -213,6 +213,8 @@ fn client_state_filters_forged_notifications() { 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}"); From bd78cbef15d81eaf8d35419938a4784e2df649c3 Mon Sep 17 00:00:00 2001 From: Sion Smith Date: Mon, 26 Oct 2026 21:25:13 +0000 Subject: [PATCH 5/7] Serve /lifecycle so a --lifecycle-url there validates Graph validates lifecycleNotificationUrl as well as notificationUrl when a subscription is created, and the README suggests https://your-endpoint/lifecycle. The listener answered that path 404, so following the README failed subscription creation. Answer validation and accept notifications on /lifecycle like / and /webhook, with a unit and a black-box test. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 2 +- CLAUDE.md | 2 +- README.md | 4 ++-- docs/man/teams.1 | 6 +++--- src/listen/handler.rs | 23 ++++++++++++++++++++++- src/listen/mod.rs | 7 ++++--- tests/listen.rs | 17 +++++++++++++++++ 7 files changed, 50 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7620c7c..c74b9ae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,7 @@ ### 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. +- `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. diff --git a/CLAUDE.md b/CLAUDE.md index a7ae018..5f88d2e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -86,7 +86,7 @@ CLI flags > env vars (TEAMS_CLI_CLIENT_ID, TEAMS_CLI_CLIENT_SECRET, TEAMS_CLI_TE ### Webhook Listener - `teams listen --port 8080` starts a hyper HTTP server on 127.0.0.1 (`--bind` to change) -- Handles Graph subscription validation (echoes `?validationToken`) +- 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 diff --git a/README.md b/README.md index e7ed00b..f8061e2 100644 --- a/README.md +++ b/README.md @@ -626,7 +626,7 @@ 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). Each line is one notification with every field Graph sent except `clientState`, plus `"kind": "change"` or, for lifecycle events (`lifecycleEvent`), `"kind": "lifecycle"`. +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. @@ -645,7 +645,7 @@ teams listen --port 8080 A relay can forward notifications to `teams listen` instead of Graph calling it directly: -- POST the raw Graph body, `{"value":[...]}`, to `/` or `/webhook`. +- 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. diff --git a/docs/man/teams.1 b/docs/man/teams.1 index 55a51c0..0599607 100644 --- a/docs/man/teams.1 +++ b/docs/man/teams.1 @@ -405,9 +405,9 @@ without it, the listener warns that notifications are not authenticated. .PP Behind a relay, POST the raw Graph body .B {"value":[...]} -to / or /webhook, 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 +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. diff --git a/src/listen/handler.rs b/src/listen/handler.rs index 331582a..b424a8d 100644 --- a/src/listen/handler.rs +++ b/src/listen/handler.rs @@ -62,7 +62,9 @@ where match (req.method(), req.uri().path()) { (&Method::GET, "/") => Ok(plain_response(StatusCode::OK, "OK")), - (&Method::POST, "/") | (&Method::POST, "/webhook") => { + // `/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() { @@ -203,6 +205,25 @@ mod tests { assert_eq!(body_text(resp).await, "a b+c"); } + #[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); diff --git a/src/listen/mod.rs b/src/listen/mod.rs index 34957bf..dbdbec3 100644 --- a/src/listen/mod.rs +++ b/src/listen/mod.rs @@ -22,9 +22,10 @@ const HEADER_TIMEOUT: Duration = Duration::from_secs(10); /// /// The server handles: /// - GET `/` — health check (200 OK) -/// - POST with `?validationToken=xxx` — Graph subscription validation (echoes token) -/// - POST with JSON body — change notifications, printed as NDJSON to stdout -/// by a writer thread; 202 once queued, 503 when the queue is full +/// - 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 diff --git a/tests/listen.rs b/tests/listen.rs index 6166da6..4a9560d 100644 --- a/tests/listen.rs +++ b/tests/listen.rs @@ -241,6 +241,23 @@ fn lifecycle_notifications_are_printed_and_unauthenticated_mode_warns() { } } +#[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(&[], &[]); From 7fac8a7596c3037f118ec95d7df6178a2b9da3a8 Mon Sep 17 00:00:00 2001 From: Sion Smith Date: Mon, 26 Oct 2026 22:59:15 +0000 Subject: [PATCH 6/7] docs: merge the duplicated Unreleased sections after rebase Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 10 ++-------- 1 file changed, 2 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c74b9ae..9961f09 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,20 +16,14 @@ - `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`. - -### 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). - -### Changed - - 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). From 770b375939081f60166cc0e5bd95b9ffd2bc01c2 Mon Sep 17 00:00:00 2001 From: Sion Smith Date: Mon, 26 Oct 2026 23:36:24 +0000 Subject: [PATCH 7/7] test: report the listener's stderr when a request cannot be sent client_state_filters_forged_notifications failed three times locally, under heavy concurrent load, with ENOTCONN on the first write after a successful connect. About 650 stress runs and 300 instant-POST startups did not reproduce it, so the panic now names the port and carries the listener's stderr, so a recurrence shows what the listener was doing. Co-Authored-By: Claude Opus 5.5 --- tests/listen.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/tests/listen.rs b/tests/listen.rs index 4a9560d..e2a58cd 100644 --- a/tests/listen.rs +++ b/tests/listen.rs @@ -158,7 +158,15 @@ impl Listener { /// Send a raw request and return (status, body). fn raw(&self, request: &[u8]) -> (u16, String) { let mut stream = self.connect(); - stream.write_all(request).unwrap(); + 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) }