diff --git a/.github/workflows/voltage-agent-images.yml b/.github/workflows/voltage-agent-images.yml index b28027f087e..b494bbc323c 100644 --- a/.github/workflows/voltage-agent-images.yml +++ b/.github/workflows/voltage-agent-images.yml @@ -257,6 +257,7 @@ jobs: echo "==> test-$test.py" python3 "test-$test.py" "$IMAGE" done + python3 test-codex-login.py "$IMAGE" python3 -O test-profile.py "$IMAGE" python3 test-cluster.py "$IMAGE" diff --git a/.github/workflows/voltage-buzz-gpui.yml b/.github/workflows/voltage-buzz-gpui.yml index 6e6d4c3c355..b830e20dab5 100644 --- a/.github/workflows/voltage-buzz-gpui.yml +++ b/.github/workflows/voltage-buzz-gpui.yml @@ -50,6 +50,8 @@ jobs: run: nix develop --command cargo fmt --all --check - name: Clippy run: nix develop --command cargo clippy --workspace --all-targets --locked -- -D warnings + - name: Voltage agent Clippy + run: nix develop --command cargo clippy -p buzz-gpui --all-targets --locked --features voltage-sandbox -- -D warnings test: name: GPUI Test @@ -76,6 +78,8 @@ jobs: # Tests that need the local test relay skip themselves without it. - name: Test run: nix develop --command cargo test --workspace --locked + - name: Voltage agent tests + run: nix develop --command cargo test -p buzz-gpui -p buzz-identity --locked --features buzz-gpui/voltage-sandbox,buzz-identity/session-control plugins: name: GPUI Node plugins @@ -169,3 +173,8 @@ jobs: working-directory: apps/buzz-gpui env: RUSTUP_TOOLCHAIN: 1.98.1 + - name: Voltage agent Clippy + run: cargo clippy -p buzz-gpui --locked --features voltage-sandbox -- -D warnings + working-directory: apps/buzz-gpui + env: + RUSTUP_TOOLCHAIN: 1.98.1 diff --git a/Cargo.lock b/Cargo.lock index 13507ebf0ab..2aa36b62736 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -653,6 +653,32 @@ dependencies = [ "uuid", ] +[[package]] +name = "aws-sdk-dynamodb" +version = "1.131.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18d4e5ed975d72bce813432317e0cf63d4b37b5f62e9bc0a4922b2fe2b2c4347" +dependencies = [ + "arc-swap", + "aws-credential-types", + "aws-runtime", + "aws-smithy-async", + "aws-smithy-http", + "aws-smithy-json", + "aws-smithy-observability", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", + "aws-types", + "bytes", + "fastrand", + "http 1.4.2", + "regex-lite", + "tracing", + "url", +] + [[package]] name = "aws-sdk-s3" version = "1.143.0" @@ -1069,6 +1095,28 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "axum-server" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1df331683d982a0b9492b38127151e6453639cd34926eb9c07d4cd8c6d22bfc" +dependencies = [ + "arc-swap", + "bytes", + "either", + "fs-err", + "http 1.4.2", + "http-body 1.0.1", + "hyper", + "hyper-util", + "pin-project-lite", + "rustls", + "rustls-pki-types", + "tokio", + "tokio-rustls", + "tower-service", +] + [[package]] name = "backon" version = "1.6.0" @@ -2001,9 +2049,12 @@ version = "0.1.0" dependencies = [ "anyhow", "aws-config", + "aws-sdk-dynamodb", "aws-sdk-s3", "aws-sdk-secretsmanager", "axum", + "axum-server", + "base64 0.22.1", "buzz-backend-kubernetes", "buzz-backend-relay", "buzz-sdk", @@ -2025,15 +2076,18 @@ dependencies = [ "serde", "serde_json", "sha2 0.11.0", + "subtle", "thiserror 2.0.18", "tokio", "tokio-tungstenite 0.29.0", "tokio-util", "toml", "tower 0.5.3", + "tower-http", "tracing", "tracing-subscriber", "url", + "uuid", "zeroize", ] @@ -3912,6 +3966,16 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs-err" +version = "3.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5c95b673b8f6f7235229ae11c5642d81b04c2e64c1e2fb417bc0cf73ca45f29" +dependencies = [ + "autocfg", + "tokio", +] + [[package]] name = "fs2" version = "0.4.3" diff --git a/apps/buzz-gpui/Cargo.lock b/apps/buzz-gpui/Cargo.lock index b5b174efa48..0ac2cb5e87f 100644 --- a/apps/buzz-gpui/Cargo.lock +++ b/apps/buzz-gpui/Cargo.lock @@ -1092,6 +1092,7 @@ dependencies = [ "libc", "markdown", "markup5ever_rcdom", + "nix 0.31.3", "nostr 0.45.5", "regex", "reqwest", diff --git a/apps/buzz-gpui/crates/app/Cargo.toml b/apps/buzz-gpui/crates/app/Cargo.toml index ddde7496100..be0c2997f2d 100644 --- a/apps/buzz-gpui/crates/app/Cargo.toml +++ b/apps/buzz-gpui/crates/app/Cargo.toml @@ -67,7 +67,7 @@ sha2 = "0.10" # The voice-note encoder writes into a private directory of its own. tempfile = "3" serde_yaml = "0.9" -tokio = { version = "1", features = ["rt-multi-thread", "sync", "time", "macros", "net", "io-util", "process"] } +tokio = { version = "1", features = ["rt-multi-thread", "sync", "time", "macros", "net", "io-util", "process", "fs"] } toml = "0.9" toml_edit = "0.25" unicode-normalization = "0.1" @@ -115,6 +115,10 @@ tray-icon = { version = "0.26", default-features = false } # Only to turn on its runtime_shaders feature (see [features]). gpui-pre-macos = { version = "=0.3.8", optional = true } +[target.'cfg(unix)'.dependencies] +# Stop the device-login CLI and its npm/native child as one owned process group. +nix = { version = "0.31", features = ["signal", "process"] } + [target.'cfg(windows)'.dependencies] # The user's `buzz://` registration; GPUI already uses it. windows-registry = "0.6" diff --git a/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/codex_login.rs b/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/codex_login.rs new file mode 100644 index 00000000000..e4d7666794e --- /dev/null +++ b/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/codex_login.rs @@ -0,0 +1,336 @@ +//! A dedicated device login, separate from the user's normal Codex cache. +//! Only the one-time device code is shown. The private cache is removed +//! when the login completes, fails, times out, or is cancelled. + +use std::process::Stdio; +use std::time::Duration; + +#[cfg(windows)] +use buzz_desktop::process::windows_job::WindowsJob; +use buzz_desktop::process::{effective_path, resolve_program}; +#[cfg(unix)] +use nix::{ + sys::signal::{Signal, killpg}, + unistd::Pid, +}; +use regex::Regex; +use serde_json::{Value, from_str}; +use tempfile::Builder as TempDirBuilder; +use tokio::fs::{metadata, read_to_string}; +use tokio::io::AsyncReadExt; +use tokio::process::Command; +use tokio::sync::mpsc::Sender; +use tokio::task::JoinHandle; +use tokio::time::{interval, sleep}; +use tokio::{pin, select}; +use zeroize::Zeroizing; + +const MAX_BYTES: usize = 16_384; +const LOGIN_TIMEOUT: Duration = Duration::from_secs(16 * 60); + +/// A completed sign-in remains bound to the Buzz identity and community +/// that requested it, including after the asynchronous task has completed. +pub struct PendingLogin { + cache: Zeroizing, + owner: String, + relay: String, +} + +impl PendingLogin { + /// Bind a dedicated cache to its enrollment scope. + pub fn new(cache: Zeroizing, owner: String, relay: String) -> Self { + Self { + cache, + owner, + relay, + } + } + + /// Consume only for the same identity and community. + pub fn into_cache(self, owner: &str, relay: &str) -> Result, String> { + if self.owner != owner || self.relay != relay { + return Err("Buzz identity or community changed; sign in to ChatGPT again.".into()); + } + Ok(self.cache) + } +} + +#[cfg(unix)] +struct LoginGroup(Pid); + +#[cfg(unix)] +impl Drop for LoginGroup { + fn drop(&mut self) { + // The group is created before spawn and contains the npm wrapper + // and its native child. Killing only the wrapper leaks that child. + let _ = killpg(self.0, Signal::SIGKILL); + } +} + +/// Cancelling the view's future also cancels its runtime login task. +pub struct LoginTask(pub JoinHandle, String>>); + +impl Drop for LoginTask { + fn drop(&mut self) { + self.0.abort(); + } +} + +fn prompt(output: &[u8]) -> String { + let plain = String::from_utf8_lossy(output); + let plain = Regex::new(r"\x1b\[[0-9;]*m") + .map(|ansi| ansi.replace_all(&plain, "").into_owned()) + .unwrap_or_default(); + let code = Regex::new(r"\b[A-Z0-9]{4}-[A-Z0-9]{4,6}\b") + .ok() + .and_then(|code| code.find(&plain).map(|found| found.as_str().to_string())); + match code { + Some(code) => format!( + "Open https://auth.openai.com/codex/device and enter {code}. Complete only the sign-in you started here." + ), + None => "Requesting a ChatGPT sign-in code…".into(), + } +} + +/// Sign in with the installed CLI without inheriting other model credentials. +pub async fn run(progress: Sender) -> Result, String> { + let path = effective_path(); + let mut command = Command::new(resolve_program("codex", &path)); + command.env("PATH", path); + run_command(progress, command).await +} + +async fn run_command( + progress: Sender, + mut command: Command, +) -> Result, String> { + let directory = TempDirBuilder::new() + .prefix("buzz-codex-login-") + .tempdir() + .map_err(|_| "Could not create a private login directory.".to_string())?; + command + .args([ + "-c", + "cli_auth_credentials_store=\"file\"", + "login", + "--device-auth", + ]) + .env("CODEX_HOME", directory.path()) + .env_remove("CODEX_ACCESS_TOKEN") + .env_remove("CODEX_API_KEY") + .env_remove("OPENAI_API_KEY") + .env_remove("CODEX_REFRESH_TOKEN_URL_OVERRIDE") + .env_remove("CODEX_CA_CERTIFICATE") + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .kill_on_drop(true); + #[cfg(unix)] + command.process_group(0); + #[cfg(windows)] + let _job = WindowsJob::new().ok_or("Could not contain the Codex sign-in process.")?; + #[cfg(windows)] + let mut child = _job + .spawn_hidden(&mut command) + .ok_or("Could not start a contained Codex sign-in process.")?; + #[cfg(not(windows))] + let mut child = command + .spawn() + .map_err(|_| "Could not start Codex. Install the Codex CLI and try again.".to_string())?; + #[cfg(unix)] + let _group = LoginGroup( + child + .id() + .and_then(|id| i32::try_from(id).ok()) + .map(Pid::from_raw) + .ok_or("Could not contain the Codex sign-in process.")?, + ); + let mut output = child + .stdout + .take() + .ok_or("Could not read Codex's sign-in prompt.")?; + let mut captured = Zeroizing::new(Vec::new()); + let mut chunk = [0_u8; 1024]; + let mut closed = false; + let deadline = sleep(LOGIN_TIMEOUT); + pin!(deadline); + let mut polling = interval(Duration::from_millis(500)); + let status = loop { + select! { + _ = &mut deadline => { + let _ = child.kill().await; + let _ = child.wait().await; + return Err("ChatGPT sign-in timed out. Start a new sign-in.".into()); + } + _ = polling.tick() => { + if progress.is_closed() { + let _ = child.kill().await; + let _ = child.wait().await; + return Err("ChatGPT sign-in was cancelled.".into()); + } + let _ = progress.try_send(prompt(&captured)); + if let Some(status) = child.try_wait().map_err(|_| "Could not check ChatGPT sign-in.".to_string())? { + break status; + } + } + read = output.read(&mut chunk), if !closed => { + let count = read.map_err(|_| "Could not read ChatGPT sign-in progress.".to_string())?; + if count == 0 { + closed = true; + } else if captured.len().saturating_add(count) > MAX_BYTES { + return Err("Codex returned too much sign-in output.".into()); + } else { + captured.extend_from_slice(&chunk[..count]); + } + } + } + }; + if !status.success() { + return Err("ChatGPT sign-in did not finish. Check that device-code login is enabled for your workspace, then try again.".into()); + } + let path = directory.path().join("auth.json"); + let size = metadata(&path) + .await + .map_err(|_| "Codex did not save a login.".to_string())? + .len(); + if size > MAX_BYTES as u64 { + return Err("Codex returned an oversized login.".into()); + } + let cache = Zeroizing::new( + read_to_string(path) + .await + .map_err(|_| "Could not read the dedicated Codex login.".to_string())?, + ); + let value: Value = from_str(&cache).map_err(|_| "Codex saved an invalid login.".to_string())?; + if value["auth_mode"] != "chatgpt" + || !value["OPENAI_API_KEY"].is_null() + || value["tokens"]["refresh_token"] + .as_str() + .is_none_or(str::is_empty) + { + return Err("Codex did not save a ChatGPT subscription login.".into()); + } + Ok(cache) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[cfg(unix)] + mod process { + use super::super::*; + use std::fs::{read_to_string as read_file, set_permissions, write}; + use std::os::unix::fs::PermissionsExt; + use std::path::Path; + use tempfile::{TempDir, tempdir}; + use tokio::sync::mpsc::channel; + use tokio::time::timeout; + use tokio::{spawn, test as async_test}; + + fn executable(tail: &str) -> (TempDir, Command) { + let directory = tempdir().unwrap(); + let path = directory.path().join("codex-fixture"); + let script = format!( + r#"#!/bin/sh +printf '%s\n%s\n' "$CODEX_HOME" "$$" > "$(dirname "$0")/record" +{tail} +"# + ); + write(&path, script).unwrap(); + set_permissions(&path, PermissionsExt::from_mode(0o700)).unwrap(); + (directory, Command::new(path)) + } + + #[async_test] + async fn a_completed_login_uses_and_removes_a_dedicated_cache() { + let (directory, command) = executable( + r#" +printf '%s' '{"auth_mode":"chatgpt","OPENAI_API_KEY":null,"tokens":{"refresh_token":"private-fixture"}}' > "$CODEX_HOME/auth.json" +printf 'ABCD-EFGHJ\n' +"#, + ); + let (sender, _receiver) = channel(1); + let cache = run_command(sender, command).await.unwrap(); + assert!(cache.contains("private-fixture")); + let record = read_file(directory.path().join("record")).unwrap(); + assert!(!Path::new(record.lines().next().unwrap()).exists()); + } + + #[async_test] + async fn dropping_the_login_task_kills_the_cli_and_removes_its_cache() { + let (directory, command) = executable( + r#"sleep 60 & +printf '%s\n' "$!" >> "$(dirname "$0")/record" +wait"#, + ); + let (sender, _receiver) = channel(1); + let task = spawn(run_command(sender, command)); + let record = timeout(Duration::from_secs(5), async { + loop { + if let Ok(record) = read_file(directory.path().join("record")) + && record.lines().count() == 3 + { + break record; + } + sleep(Duration::from_millis(20)).await; + } + }) + .await + .unwrap(); + let mut lines = record.lines(); + let cache = lines.next().unwrap(); + let pids: Vec<_> = lines.collect(); + task.abort(); + assert!(task.await.unwrap_err().is_cancelled()); + assert!(!Path::new(cache).exists()); + timeout(Duration::from_secs(5), async { + loop { + let mut alive = false; + for pid in &pids { + alive |= Command::new("kill") + .args(["-0", *pid]) + .stderr(Stdio::null()) + .status() + .await + .unwrap() + .success(); + } + if !alive { + break; + } + sleep(Duration::from_millis(20)).await; + } + }) + .await + .unwrap(); + } + } + + #[test] + fn a_completed_login_cannot_be_enrolled_under_another_identity_or_community() { + for (owner, relay, accepted) in [ + ("alice", "community-a", true), + ("bob", "community-a", false), + ("alice", "community-b", false), + ] { + let login = PendingLogin::new( + Zeroizing::new("private-cache".into()), + "alice".into(), + "community-a".into(), + ); + assert_eq!(login.into_cache(owner, relay).is_ok(), accepted); + } + } + + #[test] + fn only_the_device_code_and_fixed_openai_url_reach_the_form() { + let text = prompt( + b"private-token-must-not-display\n\x1b[34mABCD-EFGHJ\x1b[0m\nhttps://attacker.invalid", + ); + assert!(text.contains("ABCD-EFGHJ")); + assert!(text.contains("https://auth.openai.com/codex/device")); + assert!(!text.contains("private-token")); + assert!(!text.contains("attacker")); + } +} diff --git a/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/enroll.rs b/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/enroll.rs index 857281fe70d..0b0fb06a6e2 100644 --- a/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/enroll.rs +++ b/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/enroll.rs @@ -142,6 +142,8 @@ pub struct Summary { /// The stored models entries, by key. #[serde(default)] pub models: Vec, + #[serde(default)] + pub codex_login: bool, } #[derive(Clone, Debug, Default, PartialEq, Eq, serde::Deserialize)] @@ -215,6 +217,9 @@ impl Summary { .iter() .map(|field| field.label().to_owned()), ); + if self.codex_login { + parts.push("ChatGPT login".into()); + } let updated = self .updated_at .as_deref() @@ -723,6 +728,14 @@ pub async fn done( Err("The community connection closed before the Sandbox manager answered.".into()) } +/// Model credentials to remove or replace in one enrollment. +pub struct ModelChanges { + /// Stored model fields to remove, including `codex_login` to disconnect it. + pub removed: Vec<&'static str>, + /// Dedicated native ChatGPT cache, encrypted only to the manager. + pub codex_login: Option>, +} + /// Enrolls the form's `values` as `identity`. /// /// With nothing on file, it reads the GitHub token's account, creates the @@ -737,11 +750,31 @@ pub async fn run( session: Arc, identity: Arc, manager: String, - values: Values, + mut values: Values, on_file: Option, replace_key: bool, - removed: Vec<&'static str>, + models: ModelChanges, ) -> Result<(Outcome, Option), String> { + let ModelChanges { + removed, + codex_login, + } = models; + if codex_login.is_some() { + values.set(Field::OpenaiApiKey, String::new()); + values.set(Field::CodexAccessToken, String::new()); + } + let chatgpt_login = match codex_login { + Some(cache) => Some( + serde_json::from_str::(&cache) + .map_err(|_| "The dedicated ChatGPT login is invalid.".to_string())?, + ), + None if removed.contains(&"codex_login") => Some(Value::Null), + None => None, + }; + let model_removals: Vec<_> = removed + .into_iter() + .filter(|key| *key != "codex_login") + .collect(); let token = values.get(Field::GithubToken).trim().to_owned(); let github = if !token.is_empty() { Some(signing::identity(&token).await?) @@ -775,11 +808,14 @@ pub async fn run( Changes { github: github.as_ref(), signing_key: key.as_ref().map(|key| key.private.as_str()), - removed: &removed, + removed: &model_removals, }, )?, }; - let outcome = enroll(session, identity, manager, profile).await?; + let outcome = match chatgpt_login { + Some(login) => enroll_with_login(session, identity, manager, profile, Some(login)).await?, + None => enroll(session, identity, manager, profile).await?, + }; let (Outcome::Enrolled { .. }, Some(key)) = (&outcome, key) else { return Ok((outcome, None)); }; @@ -799,11 +835,21 @@ pub async fn enroll( identity: Arc, manager: String, profile: Value, +) -> Result { + enroll_with_login(session, identity, manager, profile, None).await +} + +async fn enroll_with_login( + session: Arc, + identity: Arc, + manager: String, + profile: Value, + login: Option, ) -> Result { let manager = PublicKey::from_hex(&manager).map_err(|_| "The manager key is not valid.")?; let request_id = uuid::Uuid::new_v4().to_string(); let request = identity - .enroll_request(&manager, &request_id, profile, now()) + .enroll_request_with_codex_login(&manager, &request_id, profile, login, now()) .map_err(|error| format!("Could not sign the enrollment ({}).", error.label()))?; if json!(["EVENT", request]).to_string().len() > MAX_PUBLICATION_FRAME { return Err("The profile is too large to send. Shorten the via profile.".into()); diff --git a/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/enroll_form.rs b/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/enroll_form.rs index 1fe328c36d7..4911d8722a3 100644 --- a/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/enroll_form.rs +++ b/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/enroll_form.rs @@ -3,6 +3,7 @@ //! and once a profile is, sends only what changes. Its tokens are masked, //! cleared once sent, and never saved. +use super::codex_login::PendingLogin; use super::enroll::{Field, OnFile, Outcome, Values}; use super::settings::SettingsTab; use super::signing::Registration; @@ -14,10 +15,16 @@ use buzz_ui::{ Button, ButtonSize, Checkbox, Colors, Surface, TextArea, TextField, TypeStyled, tokens::space, }; use gpui_kit::component::input::{InputState, TextareaState}; +use gpui_kit::prelude::FluentBuilder; use gpui_kit::*; use std::sync::Arc; pub struct EnrollForm { + pub codex_login: Option, + pub logging_in: bool, + pub login_generation: u64, + pub login_abort: Option, + pub login_status: Option, /// Every field but the via profile. inputs: Vec<(Field, Entity)>, via: Entity, @@ -74,6 +81,11 @@ impl EnrollForm { TextareaState::new(window, cx).placeholder(placeholder(Field::ViaProfile, false)) }); Self { + codex_login: None, + logging_in: false, + login_generation: 0, + login_abort: None, + login_status: None, inputs, via, sending: false, @@ -103,8 +115,10 @@ impl EnrollForm { self.via.update(cx, |state, cx| { state.set_placeholder(placeholder(Field::ViaProfile, held), window, cx) }); - self.removed - .retain(|key| summary.models.iter().any(|stored| stored == key)); + self.removed.retain(|key| { + summary.models.iter().any(|stored| stored == key) + || (*key == "codex_login" && summary.codex_login) + }); if !summary.git.signing_key { self.replace_key = false; } @@ -229,6 +243,36 @@ impl EnrollForm { c, )) .child(row(&[Field::OpToken])) + .child( + div().flex().gap(space::S2) + .child(Button::new("remote-workspace-chatgpt-login", "Sign in with ChatGPT") + .size(ButtonSize::Sm).outline().disabled(disabled || self.logging_in) + .on_click(cx.listener(|this, _, window, cx| this.sign_in_chatgpt(window, cx)))) + .when(self.codex_login.is_some(), |row| row.child( + Button::new("remote-workspace-discard-chatgpt-login", "Discard sign-in") + .size(ButtonSize::Sm).outline().disabled(disabled) + .on_click(cx.listener(|this, _, _, cx| { + this.enroll.codex_login = None; + this.enroll.login_status = None; + cx.notify(); + })) + )) + .when(self.logging_in, |row| row.child( + Button::new("remote-workspace-cancel-chatgpt-login", "Cancel sign-in") + .size(ButtonSize::Sm).outline() + .on_click(cx.listener(|this, _, _, cx| { + this.enroll.login_generation = this.enroll.login_generation.wrapping_add(1); + if let Some(abort) = this.enroll.login_abort.take() { abort.abort(); } + this.enroll.logging_in = false; + this.enroll.login_status = Some("ChatGPT sign-in cancelled.".into()); + cx.notify(); + })) + )) + ) + .children(self.login_status.as_ref().map(|status| div().text_body_sm().child(status.clone()))) + .child(div().text_body_sm().text_color(c.text_subtle).child( + "Sign in once for all remote Codex agents. Enrolling this login replaces an OpenAI API key or Codex access token." + )) // Subscriptions first: both can run in parallel in every Sandbox. .child(row(&[Field::CodexAccessToken, Field::ClaudeOauthToken])) .child(row(&[ @@ -309,6 +353,29 @@ impl EnrollForm { .flex_col() .gap(space::S2) .child(div().flex().flex_wrap().gap(space::S2).children(removals)) + .when(summary.codex_login, |view| { + view.child( + Button::new( + "remote-workspace-remove-chatgpt-login", + if self.removed.contains(&"codex_login") { + "Keep ChatGPT login" + } else { + "Disconnect ChatGPT login" + }, + ) + .size(ButtonSize::Sm) + .outline() + .disabled(disabled) + .on_click(cx.listener(|this, _, _, cx| { + this.enroll.toggle_removed("codex_login"); + if this.enroll.removed.contains(&"codex_login") { + this.enroll.codex_login = None; + this.enroll.login_status = None; + } + cx.notify(); + })), + ) + }) .children(removing) .children(replace), ) diff --git a/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/mod.rs b/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/mod.rs index 63b41dc43f0..b72a5b71c4e 100644 --- a/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/mod.rs +++ b/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/mod.rs @@ -4,6 +4,7 @@ //! talks to the community itself; this type only starts the session. //! Built only with the `voltage-sandbox` feature. +mod codex_login; pub mod config; mod enroll; mod enroll_form; diff --git a/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/settings.rs b/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/settings.rs index 0b3c0aff874..50e09006898 100644 --- a/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/settings.rs +++ b/apps/buzz-gpui/crates/app/src/agent_plugins/remote_workspace/settings.rs @@ -3,18 +3,22 @@ //! enrolls a developer profile with the Sandbox manager. Every agent asks the //! manager over its community's relay, with the build's manager key. +use super::codex_login::{LoginTask, PendingLogin, run as run_codex_login}; use super::config::{Config, EFFORTS, MANAGER_PUBKEY, RUNTIMES}; -use super::enroll::{OnFile, Outcome}; +use super::enroll::{ModelChanges, OnFile, Outcome}; use super::enroll_form::{self, EnrollForm}; use crate::agent_plugins::tabs::{ Picking, choice, field, instructions_field, instructions_state, options, saved_config, }; use crate::agents2::{AgentView, Agents2}; +use crate::runtime::Runtime; use buzz_ui::{ActiveTheme, Button, Surface, TextArea, TextField, TypeStyled, tokens::space}; use gpui_kit::component::input::{InputEvent, InputState, TextareaState}; use gpui_kit::prelude::FluentBuilder; use gpui_kit::*; use serde_json::Value; +use tokio::select; +use tokio::sync::mpsc::channel; #[derive(Clone, Copy, PartialEq)] pub enum Picker { @@ -224,7 +228,7 @@ impl SettingsTab { } }; self.enroll.show_on_file(OnFile::Loading, window, cx); - let asked = crate::runtime::Runtime::handle(cx).spawn(super::enroll::on_file( + let asked = Runtime::handle(cx).spawn(super::enroll::on_file( session, owner, MANAGER_PUBKEY.to_owned(), @@ -249,6 +253,9 @@ impl SettingsTab { /// community, as the signed-in owner. Its tokens are cleared at once, /// sent or not. With a profile on file, only what changes is sent. fn enroll(&mut self, window: &mut Window, cx: &mut Context) { + if self.enroll.sending || self.enroll.logging_in { + return; + } let relay = self.view.agent.relay.clone(); let owner = enroll_form::owner(cx); let values = self.enroll.submit(window, cx); @@ -266,17 +273,31 @@ impl SettingsTab { return; } }; + let codex_login = match self.enroll.codex_login.take() { + Some(login) => match login.into_cache(&owner.pubkey_hex(), &relay) { + Ok(cache) => Some(cache), + Err(error) => { + self.enroll.login_status = Some(error); + cx.notify(); + return; + } + }, + None => None, + }; self.enroll.sending = true; self.enroll.status = None; self.enroll.registration = None; - let sent = crate::runtime::Runtime::handle(cx).spawn(super::enroll::run( + let sent = Runtime::handle(cx).spawn(super::enroll::run( session, owner, MANAGER_PUBKEY.to_owned(), values, self.enroll.on_file.summary().cloned(), self.enroll.replace_key, - self.enroll.removed.clone(), + ModelChanges { + removed: self.enroll.removed.clone(), + codex_login, + }, )); cx.spawn_in(window, async move |this, cx| { let outcome = sent @@ -309,6 +330,81 @@ impl SettingsTab { } } +impl SettingsTab { + /// Device login results are fenced to this view, generation and owner. + pub(super) fn sign_in_chatgpt(&mut self, window: &mut Window, cx: &mut Context) { + if self.enroll.sending || self.enroll.logging_in { + return; + } + let Some(owner) = enroll_form::owner(cx) else { + self.enroll.login_status = Some("Sign in to Buzz first.".into()); + cx.notify(); + return; + }; + let owner = owner.pubkey_hex(); + let relay = self.view.agent.relay.clone(); + self.enroll.login_generation = self.enroll.login_generation.wrapping_add(1); + let generation = self.enroll.login_generation; + self.enroll.logging_in = true; + self.enroll.login_status = Some("Requesting a ChatGPT sign-in code…".into()); + self.enroll.codex_login = None; + let (progress, mut updates) = channel(1); + let task = Runtime::handle(cx).spawn(run_codex_login(progress)); + self.enroll.login_abort = Some(task.abort_handle()); + cx.spawn_in(window, async move |this, cx| { + let mut task = LoginTask(task); + let mut updates_open = true; + loop { + select! { + result = &mut task.0 => { + let _ = this.update_in(cx, |this, _, cx| { + if this.enroll.login_generation != generation { return; } + this.enroll.logging_in = false; + this.enroll.login_abort = None; + let same_owner = enroll_form::owner(cx).is_some_and(|identity| identity.pubkey_hex() == owner); + if !same_owner || this.view.agent.relay != relay { + this.enroll.login_status = Some("Buzz identity or community changed; start sign-in again.".into()); + } else { + match result { + Ok(Ok(cache)) => { + this.enroll.codex_login = Some(PendingLogin::new(cache, owner.clone(), relay.clone())); + this.enroll.removed.retain(|key| *key != "codex_login"); + this.enroll.login_status = Some("Signed in. Select Enroll profile to use this login for remote agents.".into()); + } + Ok(Err(error)) => this.enroll.login_status = Some(error), + Err(_) => this.enroll.login_status = Some("ChatGPT sign-in cancelled.".into()), + } + } + cx.notify(); + }); + break; + } + progress = updates.recv(), if updates_open => { + let Some(progress) = progress else { updates_open = false; continue; }; + let active = this.update_in(cx, |this, _, cx| { + let active = this.enroll.login_generation == generation + && this.view.agent.relay == relay + && enroll_form::owner(cx).is_some_and(|identity| identity.pubkey_hex() == owner); + if active { + this.enroll.login_status = Some(progress); + cx.notify(); + } else if this.enroll.login_generation == generation { + this.enroll.logging_in = false; + this.enroll.login_abort = None; + this.enroll.login_status = Some("Buzz identity or community changed; start sign-in again.".into()); + cx.notify(); + } + active + }).unwrap_or(false); + if !active { break; } + } + } + } + }).detach(); + cx.notify(); + } +} + impl Picking for SettingsTab { type Picker = Picker; @@ -443,7 +539,7 @@ impl Render for SettingsTab { if sending { "Enrolling…" } else { "Enroll profile" }, ) .loading(sending) - .disabled(sending) + .disabled(sending || self.enroll.logging_in) .on_click( cx.listener(|this, _, window, cx| this.enroll(window, cx)), ), diff --git a/apps/buzz-gpui/crates/identity/src/session_control.rs b/apps/buzz-gpui/crates/identity/src/session_control.rs index d82b475feb5..c6b990058a5 100644 --- a/apps/buzz-gpui/crates/identity/src/session_control.rs +++ b/apps/buzz-gpui/crates/identity/src/session_control.rs @@ -7,6 +7,7 @@ use super::Identity; use buzz_session_control::{self as control, Op, Reply, Request, WireError}; use nostr::prelude::{Event, EventId, PublicKey}; use nostr_wire::JsonUtil as _; +use serde_json::{Value, json}; impl Identity { /// The owner's signed `enroll` request carrying `profile` to `manager`. @@ -16,10 +17,26 @@ impl Identity { &self, manager: &PublicKey, request_id: &str, - profile: serde_json::Value, + profile: Value, created_at: u64, ) -> Result { - let payload = serde_json::json!({ "profile": profile }); + self.enroll_request_with_codex_login(manager, request_id, profile, None, created_at) + } + + /// Enroll a profile with an optional dedicated ChatGPT login. A null + /// login disconnects it. The entire payload is encrypted to the manager. + pub fn enroll_request_with_codex_login( + &self, + manager: &PublicKey, + request_id: &str, + profile: Value, + login: Option, + created_at: u64, + ) -> Result { + let mut payload = json!({ "profile": profile }); + if let Some(login) = login { + payload["codex_login"] = login; + } self.owner_request( manager, Request::new(Op::Enroll, request_id, payload), @@ -35,7 +52,7 @@ impl Identity { request_id: &str, created_at: u64, ) -> Result { - let request = Request::new(Op::Profile, request_id, serde_json::json!({})); + let request = Request::new(Op::Profile, request_id, json!({})); self.owner_request(manager, request, created_at) } @@ -105,7 +122,7 @@ mod tests { #[test] fn the_manager_opens_an_enroll_request_signed_by_the_owner() { let (owner, manager) = (Identity::ephemeral(), manager()); - let profile = serde_json::json!({"github": {"token": "ghp_fixture"}}); + let profile = json!({"github": {"token": "ghp_fixture"}}); let event = owner .enroll_request( &manager_key(&manager), @@ -128,20 +145,39 @@ mod tests { assert_eq!(opened.auth_tag, None); assert_eq!(opened.request.op, Op::Enroll); assert_eq!(opened.request.request_id, "enroll-1"); - assert_eq!( - opened.request.payload, - serde_json::json!({ "profile": profile }) - ); + assert_eq!(opened.request.payload, json!({ "profile": profile })); + } + + #[test] + fn a_dedicated_login_is_encrypted_to_the_manager() { + let (owner, manager_keys) = (Identity::ephemeral(), manager()); + let login = json!({"tokens": {"refresh_token": "private-upstream-fixture"}}); + let event = owner + .enroll_request_with_codex_login( + &manager_key(&manager_keys), + "chatgpt-login", + json!({}), + Some(login.clone()), + 1_700_000_000, + ) + .unwrap(); + let serialized = event.as_json(); + assert!(!serialized.contains("private-upstream-fixture")); + assert!(!serialized.contains("codex_login")); + let opened = control::open_request(&wire(&event), &manager_keys).unwrap(); + assert_eq!(opened.request.payload["codex_login"], login); + let stranger = Keys::parse(&"09".repeat(32)).unwrap(); + assert!(control::open_request(&wire(&event), &stranger).is_err()); } #[test] fn only_the_managers_reply_to_this_request_opens() { let (owner, manager) = (Identity::ephemeral(), manager()); let request = owner - .enroll_request(&manager_key(&manager), "enroll-1", serde_json::json!({}), 0) + .enroll_request(&manager_key(&manager), "enroll-1", json!({}), 0) .unwrap(); let reply = |keys: &Keys, to: &Event| { - let done = Reply::done("enroll-1", serde_json::json!({"ok": true})); + let done = Reply::done("enroll-1", json!({"ok": true})); let requester = nostr_wire::PublicKey::from_hex(&owner.pubkey_hex()).unwrap(); app(&control::build_reply(keys, &requester, &wire(to).id, &done, 0.into()).unwrap()) }; @@ -149,7 +185,7 @@ mod tests { let opened = owner .open_session_reply(&reply(&manager, &request), &manager_pubkey, &request.id) .unwrap(); - assert_eq!(opened.response, Some(serde_json::json!({"ok": true}))); + assert_eq!(opened.response, Some(json!({"ok": true}))); let stranger = Keys::parse(&"09".repeat(32)).unwrap(); let forged = reply(&stranger, &request); @@ -158,7 +194,7 @@ mod tests { Err(WireError::WrongSender) ); let other = owner - .enroll_request(&manager_pubkey, "enroll-2", serde_json::json!({}), 0) + .enroll_request(&manager_pubkey, "enroll-2", json!({}), 0) .unwrap(); assert_eq!( owner.open_session_reply(&reply(&manager, &other), &manager_pubkey, &request.id), diff --git a/apps/buzz-gpui/plugins/reminders/src/reminders.test.ts b/apps/buzz-gpui/plugins/reminders/src/reminders.test.ts index 8ddee6f62e5..1aa88c6e456 100644 --- a/apps/buzz-gpui/plugins/reminders/src/reminders.test.ts +++ b/apps/buzz-gpui/plugins/reminders/src/reminders.test.ts @@ -79,8 +79,9 @@ describe("reminders", () => { expect(host.sent).toContainEqual({ method: "message/markers", params: { action: "remind-me", messages: ["m-later", "m-old"] } }); // One that comes due after the baseline notifies, once. await new Promise((resolve) => setTimeout(resolve, 1100)); - await list([reminder("old", now() - 3600), reminder("now", now())]); - await list([reminder("old", now() - 3600), reminder("now", now())]); + const due = reminder("now", now()); + await list([reminder("old", now() - 3600), due]); + await list([reminder("old", now() - 3600), due]); expect(submitted).toHaveLength(1); expect(submitted[0]).toMatchObject({ category: "reminders", diff --git a/crates/buzz-sandbox-reaper/Cargo.toml b/crates/buzz-sandbox-reaper/Cargo.toml index 4d2b4ad5e9c..b0aa71eccc1 100644 --- a/crates/buzz-sandbox-reaper/Cargo.toml +++ b/crates/buzz-sandbox-reaper/Cargo.toml @@ -19,10 +19,11 @@ buzz-session-control = { workspace = true } buzz-ws-client = { workspace = true } nostr = { workspace = true } sha2 = { workspace = true } +subtle = { workspace = true } thiserror = { workspace = true } kube = { workspace = true, features = ["runtime"] } k8s-openapi = { workspace = true } -tokio = { workspace = true } +tokio = { workspace = true, features = ["fs"] } tokio-util = { workspace = true } futures-util = { workspace = true } serde = { workspace = true } @@ -31,6 +32,8 @@ chrono = { workspace = true } anyhow = { workspace = true } clap = { version = "4", features = ["derive", "env"] } axum = { workspace = true } +axum-server = { version = "0.8", features = ["tls-rustls"] } +tower-http = { workspace = true, features = ["timeout"] } http = "1" http-body-util = "0.1" metrics = { workspace = true } @@ -39,12 +42,15 @@ tracing = { workspace = true } tracing-subscriber = { workspace = true } aws-config = { version = "1", default-features = false, features = ["rt-tokio", "default-https-client"] } aws-sdk-s3 = { version = "1", default-features = false, features = ["rt-tokio", "default-https-client"] } +aws-sdk-dynamodb = { version = "1", default-features = false, features = ["rt-tokio", "default-https-client"] } aws-sdk-secretsmanager = { version = "1", default-features = false, features = ["rt-tokio", "default-https-client"] } # The enrollment GitHub login check (`enroll.rs`). reqwest = { workspace = true } toml = "1.0" zeroize = { workspace = true } url = { workspace = true } +base64 = { workspace = true } +uuid = { workspace = true } # Same reason as buzz-backend-kubernetes: the process-level CryptoProvider is # installed explicitly because the workspace unifies ring and aws-lc-rs. rustls = { version = "0.23", default-features = false, features = ["ring", "std"] } diff --git a/crates/buzz-sandbox-reaper/src/codex_login.rs b/crates/buzz-sandbox-reaper/src/codex_login.rs new file mode 100644 index 00000000000..a977d4b77ee --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/codex_login.rs @@ -0,0 +1,412 @@ +//! Manager-owned ChatGPT logins. Agents receive a broker credential, never +//! the upstream refresh token. A consistent journal selects an immutable +//! secret version. Enrollment activates a staged login after writing the +//! profile. The deployment runs one manager with Recreate. + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine}; +use serde::{Deserialize, Serialize}; +use serde_json::{from_slice, from_str, json, to_string, Value}; +use sha2::{Digest, Sha256}; +use subtle::ConstantTimeEq; +use thiserror::Error as ThisError; +use tokio::sync::{Mutex, OwnedMutexGuard}; +use tokio::time::timeout; +use url::Url; +use uuid::Uuid; +use zeroize::{Zeroize, Zeroizing}; + +use self::journal::Journal; +use crate::enroll::ProfileStore; +use crate::manager::Developer; + +mod journal; +mod oauth; +pub mod server; +pub use oauth::OAuth; + +const LOCK_TIMEOUT: Duration = Duration::from_secs(30); +const MAX_TOKEN_BYTES: usize = 8192; +const EXPIRY_WINDOW: i64 = 300; + +/// Refusals contain fixed labels only, including storage and OAuth errors. +#[derive(Debug, Clone, Copy, PartialEq, Eq, ThisError)] +pub enum Error { + #[error("invalid ChatGPT login")] + InvalidLogin, + #[error("ChatGPT login is not configured")] + NotConfigured, + #[error("ChatGPT login is disconnected or replaced")] + Unauthorized, + #[error("ChatGPT login storage is unavailable")] + Storage, + #[error("a refresh was interrupted; sign in to ChatGPT again")] + Interrupted, + #[error("ChatGPT login was rejected; sign in again")] + Rejected, + #[error("ChatGPT authentication service is unavailable")] + Upstream, + #[error("ChatGPT login is busy; retry")] + Busy, +} + +/// Secret-bearing types deliberately do not implement Debug. +#[derive(Clone, Serialize, Deserialize)] +pub struct Tokens { + pub id_token: String, + pub access_token: String, + pub refresh_token: String, + pub account_id: String, +} + +/// An owner may connect a dedicated login or disconnect remote agents. +/// The original cache is never written into the agent-readable profile. +pub enum Change { + Connect(Tokens), + Disconnect, +} + +impl Change { + /// Parse the optional encrypted enrollment change without exposing its contents. + pub fn parse(value: Option<&Value>) -> Result, Error> { + match value { + None => Ok(None), + Some(Value::Null) => Ok(Some(Self::Disconnect)), + Some(value) => Tokens::from_login(value).map(Self::Connect).map(Some), + } + } +} + +impl Drop for Tokens { + fn drop(&mut self) { + self.id_token.zeroize(); + self.access_token.zeroize(); + self.refresh_token.zeroize(); + } +} + +fn token(value: &str) -> bool { + !value.is_empty() && value.len() <= MAX_TOKEN_BYTES && !value.chars().any(char::is_whitespace) +} + +fn claims(bearer: &str) -> Result { + if !token(bearer) { + return Err(Error::InvalidLogin); + } + let mut parts = bearer.split('.'); + let (_, Some(payload), Some(_), None) = + (parts.next(), parts.next(), parts.next(), parts.next()) + else { + return Err(Error::InvalidLogin); + }; + let bytes = URL_SAFE_NO_PAD + .decode(payload) + .map_err(|_| Error::InvalidLogin)?; + from_slice(&bytes).map_err(|_| Error::InvalidLogin) +} + +impl Tokens { + /// Read a dedicated `codex login` cache, without accepting API-key auth. + pub fn from_login(value: &Value) -> Result { + if value["auth_mode"] != "chatgpt" || !value["OPENAI_API_KEY"].is_null() { + return Err(Error::InvalidLogin); + } + let string = |field: &str| { + value["tokens"][field] + .as_str() + .filter(|v| token(v)) + .map(str::to_owned) + .ok_or(Error::InvalidLogin) + }; + let tokens = Self { + id_token: string("id_token")?, + access_token: string("access_token")?, + refresh_token: string("refresh_token")?, + account_id: string("account_id")?, + }; + tokens.validate()?; + Ok(tokens) + } + + pub(super) fn validate(&self) -> Result<(), Error> { + if !token(&self.refresh_token) || !token(&self.account_id) { + return Err(Error::InvalidLogin); + } + let identity = claims(&self.id_token)?; + if identity["https://api.openai.com/auth"]["chatgpt_account_id"].as_str() + != Some(self.account_id.as_str()) + || claims(&self.access_token)?["exp"].as_i64().is_none() + { + return Err(Error::InvalidLogin); + } + Ok(()) + } + + fn expiring(&self, now: i64) -> Result { + claims(&self.access_token)?["exp"] + .as_i64() + .map(|expires| expires <= now.saturating_add(EXPIRY_WINDOW)) + .ok_or(Error::InvalidLogin) + } +} + +#[derive(Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +enum State { + Ready, + Refreshing, +} + +#[derive(Clone, Serialize, Deserialize)] +struct Login { + credential: String, + token_version: String, + state: State, + tokens: Tokens, +} + +impl Drop for Login { + fn drop(&mut self) { + self.credential.zeroize(); + } +} + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct Record { + #[serde(skip)] + revision: Option, + version: u32, + active: Option, + pending: Option, +} + +/// The sole owner of refreshes. Enrollment takes the same per-developer +/// lock as refresh, including the bounded storage and upstream operations. +pub struct Broker { + store: Arc, + journal: Arc, + oauth: OAuth, + developers: BTreeMap>)>, + url: String, + ca_certificate: String, +} + +impl Broker { + fn new( + store: Arc, + journal: Arc, + oauth: OAuth, + developers: impl IntoIterator, + url: String, + ca_certificate: String, + ) -> Result { + let parsed = Url::parse(&url).map_err(|_| Error::NotConfigured)?; + if parsed.scheme() != "https" + || parsed.host_str().is_none() + || !parsed.username().is_empty() + || parsed.password().is_some() + || parsed.query().is_some() + || parsed.fragment().is_some() + || !ca_certificate.starts_with("-----BEGIN CERTIFICATE-----\n") + { + return Err(Error::NotConfigured); + } + Ok(Self { + store, + journal, + oauth, + developers: developers + .into_iter() + .map(|d| (d.id.clone(), (d, Arc::new(Mutex::new(()))))) + .collect(), + url, + ca_certificate, + }) + } + + /// Serialize enrollment and refresh for one configured developer. + pub async fn lock(&self, developer: &Developer) -> Result, Error> { + let (_, lock) = self + .developers + .get(&developer.id) + .ok_or(Error::NotConfigured)?; + timeout(LOCK_TIMEOUT, Arc::clone(lock).lock_owned()) + .await + .map_err(|_| Error::Busy) + } + + fn secret(developer: &Developer) -> String { + format!("voltage/staging/agents/{}/codex-login", developer.id) + } + + async fn read(&self, developer: &Developer) -> Result { + let Some(revision) = self.journal.head(developer).await? else { + return Ok(Record { + revision: None, + version: 1, + active: None, + pending: None, + }); + }; + let stored = self + .store + .get_version(&Self::secret(developer), &developer.region, &revision) + .await + .map_err(|_| Error::Storage)? + .ok_or(Error::Storage)?; + if stored.version_id != revision { + return Err(Error::Storage); + } + let mut record: Record = from_str(&stored.text).map_err(|_| Error::Storage)?; + if record.version != 1 { + return Err(Error::Storage); + } + record.revision = Some(revision); + Ok(record) + } + + async fn write(&self, developer: &Developer, record: &mut Record) -> Result<(), Error> { + let text = Zeroizing::new(to_string(record).map_err(|_| Error::Storage)?); + let revision = self + .store + .put(&Self::secret(developer), &developer.region, &text) + .await + .map_err(|_| Error::Storage)?; + // Secret versions are immutable. Publish their exact IDs with a + // conditional write before any external effect; never read AWSCURRENT. + if let Err(error) = self + .journal + .advance(developer, record.revision.as_deref(), &revision) + .await + { + // A retry of an applied update fails its own condition; a lost + // success response must not leave a durable marker unowned. + if self.journal.head(developer).await?.as_deref() != Some(revision.as_str()) { + return Err(error); + } + } + record.revision = Some(revision); + Ok(()) + } + + /// Stage a login without revoking the active login. Caller holds the lock. + pub async fn stage(&self, developer: &Developer, tokens: Tokens) -> Result { + tokens.validate()?; + let mut record = self.read(developer).await?; + let credential = format!("{}.{}", developer.id, Uuid::new_v4().simple()); + record.pending = Some(Login { + credential: credential.clone(), + token_version: Uuid::new_v4().to_string(), + state: State::Ready, + tokens, + }); + self.write(developer, &mut record).await?; + Ok(json!({ + "url": self.url, + "ca_certificate": self.ca_certificate, + "refresh_token": format!("{credential}.seed"), + })) + } + + /// Activate only after committing the profile. Caller holds the lock. + pub async fn activate(&self, developer: &Developer, credential: &str) -> Result<(), Error> { + let mut record = self.read(developer).await?; + if select(&mut record, credential)? { + self.write(developer, &mut record).await?; + } + Ok(()) + } + + /// Revoke even credentials served by a stale profile replica. + /// Caller holds the lock. Already-issued access tokens expire upstream. + pub async fn disconnect(&self, developer: &Developer) -> Result<(), Error> { + let mut record = self.read(developer).await?; + record.active = None; + record.pending = None; + self.write(developer, &mut record).await + } + + /// Coalesce concurrent stale-token requests into one upstream refresh. + /// The client persists a broker token carrying the returned version; + /// an older version receives the already-refreshed bundle. + pub async fn refresh(&self, credential: &str, now: i64) -> Result { + if credential.len() > 256 { + return Err(Error::Unauthorized); + } + let mut parts = credential.split('.'); + let (Some(id), Some(_), Some(requested_version), None) = + (parts.next(), parts.next(), parts.next(), parts.next()) + else { + return Err(Error::Unauthorized); + }; + let (developer, _) = self.developers.get(id).ok_or(Error::Unauthorized)?; + let _lock = self.lock(developer).await?; + let mut record = self.read(developer).await?; + let login = record.active.as_mut().ok_or(Error::Unauthorized)?; + if !same_credential(&format!("{}.seed", login.credential), credential) { + return Err(Error::Unauthorized); + } + if login.state != State::Ready { + return Err(Error::Interrupted); + } + if requested_version == login.token_version || login.tokens.expiring(now)? { + // Persist intent before contacting the rotating-token issuer. + // Cancellation or a lost response leaves this marker durable: + // the next process refuses to reuse an uncertain refresh token. + login.state = State::Refreshing; + self.write(developer, &mut record).await?; + let active = record.active.as_ref().ok_or(Error::NotConfigured)?; + let tokens = self.oauth.refresh(&active.tokens).await?; + let active = record.active.as_mut().ok_or(Error::NotConfigured)?; + active.tokens = tokens; + active.token_version = Uuid::new_v4().to_string(); + active.state = State::Ready; + self.write(developer, &mut record).await?; + } + let active = record.active.as_ref().ok_or(Error::NotConfigured)?; + Ok(json!({ + "access_token": active.tokens.access_token, + "id_token": active.tokens.id_token, + "refresh_token": format!("{}.{}", active.credential, active.token_version), + "account_id": active.tokens.account_id, + })) + } +} + +// Compare fixed-sized digests so mismatches do not expose a token prefix. +fn same_credential(left: &str, right: &str) -> bool { + fn base(value: &str) -> Option<&str> { + value.rsplit_once('.').map(|(base, _)| base) + } + match (base(left), base(right)) { + (Some(left), Some(right)) => bool::from(Sha256::digest(left).ct_eq(&Sha256::digest(right))), + _ => false, + } +} + +fn select(record: &mut Record, credential: &str) -> Result { + if record + .pending + .as_ref() + .is_some_and(|login| same_credential(&format!("{}.seed", login.credential), credential)) + { + record.active = record.pending.take(); + return Ok(true); + } + if record + .active + .as_ref() + .is_some_and(|login| same_credential(&format!("{}.seed", login.credential), credential)) + { + return Ok(false); + } + Err(Error::Unauthorized) +} + +#[cfg(test)] +#[path = "codex_login_tests.rs"] +mod tests; diff --git a/crates/buzz-sandbox-reaper/src/codex_login/journal.rs b/crates/buzz-sandbox-reaper/src/codex_login/journal.rs new file mode 100644 index 00000000000..a751c216efa --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/codex_login/journal.rs @@ -0,0 +1,106 @@ +//! Strongly consistent, metadata-only pointer to the committed secret version. + +use std::future::Future; +use std::pin::Pin; +use std::time::Duration; + +use aws_config::timeout::TimeoutConfig; +use aws_config::{defaults, BehaviorVersion}; +use aws_sdk_dynamodb::{types::AttributeValue, Client}; + +use super::Error; +use crate::manager::Developer; + +pub(super) type JournalResult<'a, T> = Pin> + Send + 'a>>; + +pub(super) trait Journal: Send + Sync { + fn head<'a>(&'a self, developer: &'a Developer) -> JournalResult<'a, Option>; + fn advance<'a>( + &'a self, + developer: &'a Developer, + expected: Option<&'a str>, + revision: &'a str, + ) -> JournalResult<'a, ()>; +} + +pub(super) struct DynamoJournal { + client: Client, + table: String, +} + +impl DynamoJournal { + pub async fn new(table: String) -> Result { + if table.is_empty() { + return Err(Error::NotConfigured); + } + let config = defaults(BehaviorVersion::latest()) + .timeout_config( + TimeoutConfig::builder() + .connect_timeout(Duration::from_secs(5)) + .operation_attempt_timeout(Duration::from_secs(10)) + .operation_timeout(Duration::from_secs(25)) + .build(), + ) + .load() + .await; + Ok(Self { + client: Client::new(&config), + table, + }) + } +} + +impl Journal for DynamoJournal { + fn head<'a>(&'a self, developer: &'a Developer) -> JournalResult<'a, Option> { + Box::pin(async move { + let output = self + .client + .get_item() + .table_name(&self.table) + .key("developer", AttributeValue::S(developer.id.clone())) + .consistent_read(true) + .send() + .await + .map_err(|_| Error::Storage)?; + let Some(item) = output.item else { + return Ok(None); + }; + item.get("revision") + .and_then(|v| v.as_s().ok()) + .cloned() + .map(Some) + .ok_or(Error::Storage) + }) + } + + fn advance<'a>( + &'a self, + developer: &'a Developer, + expected: Option<&'a str>, + revision: &'a str, + ) -> JournalResult<'a, ()> { + Box::pin(async move { + let mut request = self + .client + .update_item() + .table_name(&self.table) + .key("developer", AttributeValue::S(developer.id.clone())) + .update_expression("SET revision = :next") + .expression_attribute_values(":next", AttributeValue::S(revision.to_owned())); + request = if let Some(expected) = expected { + request + .condition_expression("revision = :previous") + .expression_attribute_values( + ":previous", + AttributeValue::S(expected.to_owned()), + ) + } else { + request.condition_expression("attribute_not_exists(revision)") + }; + // A lost success response is safe: the caller never contacts OAuth. + // A fresh consistent read recovers the committed version. + request.send().await.map_err(|_| Error::Storage)?; + Ok(()) + }) + } +} diff --git a/crates/buzz-sandbox-reaper/src/codex_login/oauth.rs b/crates/buzz-sandbox-reaper/src/codex_login/oauth.rs new file mode 100644 index 00000000000..e8fe478d4f2 --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/codex_login/oauth.rs @@ -0,0 +1,103 @@ +//! The pinned Codex 0.159.3 refresh contract. Never follow redirects and +//! never retain or report an upstream response body on failure. + +use std::time::Duration; + +use futures_util::StreamExt; +use reqwest::{redirect::Policy, Client}; +use serde::Deserialize; +use serde_json::{from_slice, json}; +#[cfg(test)] +use url::Url; +use zeroize::Zeroizing; + +use super::{Error, Tokens}; + +pub const CLIENT_ID: &str = "app_EMoamEEZ73f0CkXaXp7hrann"; +const ENDPOINT: &str = "https://auth.openai.com/oauth/token"; +const MAX_RESPONSE: usize = 65_536; + +/// Fixed upstream transport for the pinned native refresh grant. +pub struct OAuth { + client: Client, + endpoint: String, +} + +#[derive(Deserialize)] +struct Refreshed { + access_token: String, + id_token: Option, + refresh_token: Option, +} + +impl OAuth { + /// Build the pinned endpoint client with bounded responses and no redirects. + pub fn new() -> Result { + Self::endpoint(ENDPOINT) + } + + fn endpoint(endpoint: &str) -> Result { + let client = Client::builder() + .connect_timeout(Duration::from_secs(5)) + .timeout(Duration::from_secs(15)) + .redirect(Policy::none()) + .build() + .map_err(|_| Error::Upstream)?; + Ok(Self { + client, + endpoint: endpoint.to_string(), + }) + } + + #[cfg(test)] + pub(super) fn local(endpoint: &str) -> Result { + let url = Url::parse(endpoint).map_err(|_| Error::Upstream)?; + if url.scheme() != "http" || url.host_str() != Some("127.0.0.1") { + return Err(Error::Upstream); + } + Self::endpoint(endpoint) + } + + pub(super) async fn refresh(&self, previous: &Tokens) -> Result { + let response = self + .client + .post(&self.endpoint) + .json(&json!({ + "client_id": CLIENT_ID, + "grant_type": "refresh_token", + "refresh_token": previous.refresh_token, + })) + .send() + .await + .map_err(|_| Error::Upstream)?; + if !response.status().is_success() { + return Err(if response.status().is_client_error() { + Error::Rejected + } else { + Error::Upstream + }); + } + let mut body = Zeroizing::new(Vec::new()); + let mut stream = response.bytes_stream(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.map_err(|_| Error::Upstream)?; + if body.len().saturating_add(chunk.len()) > MAX_RESPONSE { + return Err(Error::Upstream); + } + body.extend_from_slice(&chunk); + } + let refreshed: Refreshed = from_slice(&body).map_err(|_| Error::Upstream)?; + let tokens = Tokens { + access_token: refreshed.access_token, + id_token: refreshed + .id_token + .unwrap_or_else(|| previous.id_token.clone()), + refresh_token: refreshed + .refresh_token + .unwrap_or_else(|| previous.refresh_token.clone()), + account_id: previous.account_id.clone(), + }; + tokens.validate().map_err(|_| Error::Upstream)?; + Ok(tokens) + } +} diff --git a/crates/buzz-sandbox-reaper/src/codex_login/server.rs b/crates/buzz-sandbox-reaper/src/codex_login/server.rs new file mode 100644 index 00000000000..8faefd58f04 --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/codex_login/server.rs @@ -0,0 +1,255 @@ +//! The private TLS refresh endpoint used by the pinned Codex client. + +use std::net::SocketAddr; +use std::path::PathBuf; +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{anyhow, Result}; +use axum::extract::rejection::JsonRejection; +use axum::extract::{DefaultBodyLimit, State}; +use axum::http::{header::CACHE_CONTROL, StatusCode}; +use axum::response::{IntoResponse, Response}; +use axum::routing::{get, post}; +use axum::{Json, Router}; +use axum_server::tls_rustls::RustlsConfig; +use axum_server::{bind_rustls, Handle}; +use chrono::Utc; +use clap::Args; +use serde::Deserialize; +use serde_json::json; +use tokio::fs::read_to_string; +use tokio::sync::Semaphore; +use tokio::time::{interval_at, timeout, Instant, MissedTickBehavior}; +use tokio::{join, select}; +use tokio_util::sync::CancellationToken; +use tower_http::timeout::TimeoutLayer; +use tracing::warn; +use zeroize::Zeroize; + +use super::journal::DynamoJournal; +use super::oauth::CLIENT_ID; +use super::{Broker, Error, OAuth}; +use crate::enroll::ProfileStore; +use crate::manager::Developer; + +/// No listener or certificate access occurs unless `public_url` is set. +#[derive(Args)] +pub struct Options { + #[arg(long, env = "BUZZ_CODEX_REFRESH_URL")] + public_url: Option, + #[arg(long, env = "BUZZ_CODEX_JOURNAL_TABLE")] + journal_table: Option, + #[arg( + long, + env = "BUZZ_CODEX_REFRESH_LISTEN", + default_value = "0.0.0.0:8443" + )] + listen: SocketAddr, + #[arg( + long, + env = "BUZZ_CODEX_CA_FILE", + default_value = "/etc/buzz-codex-ca/ca.crt" + )] + ca_file: PathBuf, + #[arg( + long, + env = "BUZZ_CODEX_TLS_CERT", + default_value = "/etc/buzz-codex-tls/tls.crt" + )] + certificate: PathBuf, + #[arg( + long, + env = "BUZZ_CODEX_TLS_KEY", + default_value = "/etc/buzz-codex-tls/tls.key" + )] + private_key: PathBuf, +} + +/// Concurrent refreshes across all developers; a bootstrap burst queues +/// here (within the 90 s request timeout) rather than failing a Sandbox. +const PERMITS: usize = 16; +const PERMIT_WAIT: Duration = Duration::from_secs(30); + +/// The optional broker listener and its manager-owned login service. +pub struct Server { + options: Options, + tls: RustlsConfig, + pub broker: Arc, +} + +impl Server { + /// Load mounted certificates before accepting enrollment or refreshes. + pub async fn prepare( + options: Options, + store: Arc, + developers: impl IntoIterator, + ) -> Result> { + let Some(url) = options.public_url.clone() else { + return Ok(None); + }; + let certificate = read_to_string(&options.ca_file) + .await + .map_err(|_| anyhow!("could not read the ChatGPT broker CA certificate"))?; + let broker = Arc::new(Broker::new( + store, + Arc::new( + DynamoJournal::new( + options + .journal_table + .clone() + .ok_or_else(|| anyhow!("ChatGPT broker journal table is required"))?, + ) + .await?, + ), + OAuth::new()?, + developers, + url, + certificate, + )?); + let tls = RustlsConfig::from_pem_file(&options.certificate, &options.private_key) + .await + .map_err(|_| anyhow!("could not load the ChatGPT broker TLS certificate and key"))?; + Ok(Some(Self { + options, + tls, + broker, + })) + } + + /// Serve with bounded shutdown and reload renewed leaf certificates. + pub async fn run(self, stop: CancellationToken) -> Result<()> { + let handle = Handle::new(); + let server = bind_rustls(self.options.listen, self.tls.clone()) + .handle(handle.clone()) + .serve(router(self.broker).into_make_service()); + let serving = async { + let result = server.await; + stop.cancel(); + result.map_err(|_| anyhow!("ChatGPT broker listener failed")) + }; + let stopping = async { + stop.cancelled().await; + handle.graceful_shutdown(Some(Duration::from_secs(20))); + }; + let reloading = async { + let mut interval = interval_at( + Instant::now() + Duration::from_secs(60), + Duration::from_secs(60), + ); + interval.set_missed_tick_behavior(MissedTickBehavior::Skip); + loop { + select! { + () = stop.cancelled() => break, + _ = interval.tick() => { + if self.tls.reload_from_pem_file(&self.options.certificate, &self.options.private_key).await.is_err() { + warn!("could not reload the ChatGPT broker TLS certificate"); + } + } + } + } + }; + let (result, (), ()) = join!(serving, stopping, reloading); + result + } +} + +#[derive(Clone)] +struct Endpoint { + broker: Arc, + permits: Arc, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct Refresh { + client_id: String, + grant_type: String, + refresh_token: String, +} + +impl Drop for Refresh { + fn drop(&mut self) { + self.refresh_token.zeroize(); + } +} + +pub(super) fn router(broker: Arc) -> Router { + router_with(broker, Arc::new(Semaphore::new(PERMITS))) +} + +pub(super) fn router_with(broker: Arc, permits: Arc) -> Router { + Router::new() + .route("/healthz", get(|| async { StatusCode::OK })) + .route("/oauth/token", post(refresh)) + .layer(DefaultBodyLimit::max(1024)) + .layer(TimeoutLayer::with_status_code( + StatusCode::REQUEST_TIMEOUT, + Duration::from_secs(90), + )) + .with_state(Endpoint { broker, permits }) +} + +async fn refresh( + State(endpoint): State, + payload: Result, JsonRejection>, +) -> Response { + let request = match payload { + Ok(Json(request)) => request, + Err(rejected) => { + let status = if rejected.status() == StatusCode::PAYLOAD_TOO_LARGE { + StatusCode::PAYLOAD_TOO_LARGE + } else { + StatusCode::BAD_REQUEST + }; + return invalid_request(status); + } + }; + if request.client_id != CLIENT_ID || request.grant_type != "refresh_token" { + return invalid_request(StatusCode::BAD_REQUEST); + } + let Ok(Ok(_permit)) = timeout(PERMIT_WAIT, endpoint.permits.acquire()).await else { + return failure(Error::Busy); + }; + match endpoint + .broker + .refresh(&request.refresh_token, Utc::now().timestamp()) + .await + { + Ok(response) => ( + StatusCode::OK, + [(CACHE_CONTROL, "no-store")], + Json(response), + ) + .into_response(), + Err(error) => failure(error), + } +} + +fn invalid_request(status: StatusCode) -> Response { + ( + status, + [(CACHE_CONTROL, "no-store")], + Json(json!({"error": "invalid_request"})), + ) + .into_response() +} + +fn failure(error: Error) -> Response { + let (status, code) = match error { + Error::Unauthorized + | Error::NotConfigured + | Error::Interrupted + | Error::Rejected + | Error::InvalidLogin => (StatusCode::UNAUTHORIZED, "invalid_grant"), + Error::Storage | Error::Upstream | Error::Busy => { + (StatusCode::SERVICE_UNAVAILABLE, "temporarily_unavailable") + } + }; + ( + status, + [(CACHE_CONTROL, "no-store")], + Json(json!({"error": code, "error_description": error.to_string()})), + ) + .into_response() +} diff --git a/crates/buzz-sandbox-reaper/src/codex_login_tests.rs b/crates/buzz-sandbox-reaper/src/codex_login_tests.rs new file mode 100644 index 00000000000..78c922e6347 --- /dev/null +++ b/crates/buzz-sandbox-reaper/src/codex_login_tests.rs @@ -0,0 +1,686 @@ +use super::oauth::CLIENT_ID; +use crate::enroll::tests::fake_github; +use crate::enroll::{Enrollment, GitHub}; +use crate::profile::tests::valid; +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use axum::serve; +use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use http_body_util::BodyExt; +use std::collections::HashMap; +use std::future::Future; +use std::pin::Pin; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Mutex as StdMutex; +use tokio::net::TcpListener; +use tokio::sync::{Notify, Semaphore}; +use tokio::task::{JoinHandle, JoinSet}; +use tokio::{spawn, test as async_test}; +use tower::ServiceExt; + +use axum::extract::State as RequestState; +use axum::routing::post; +use axum::{Json, Router}; + +use super::journal::{Journal, JournalResult}; +use super::*; +use crate::enroll::{StoreFailed, Stored}; + +const NOW: i64 = 1_800_000_000; +const CA: &str = "-----BEGIN CERTIFICATE-----\nfixture\n-----END CERTIFICATE-----\n"; + +fn jwt(claims: Value) -> String { + format!( + "e30.{}.signature", + URL_SAFE_NO_PAD.encode(claims.to_string()) + ) +} + +fn tokens(account: &str, expires: i64, refresh: &str) -> Tokens { + Tokens { + id_token: jwt(json!({"https://api.openai.com/auth": {"chatgpt_account_id": account}})), + access_token: jwt(json!({"exp": expires})), + refresh_token: refresh.into(), + account_id: account.into(), + } +} + +fn developer() -> Developer { + Developer { + id: "alice".into(), + namespace: "buzz-agent-staging-alice".into(), + profile_secret_arn: "profile-alice".into(), + region: "us-west-2".into(), + } +} + +#[derive(Default)] +struct Store { + values: StdMutex>, + versions: StdMutex>, + heads: StdMutex>, + puts: AtomicUsize, + fail_at: AtomicUsize, + advances: AtomicUsize, + /// The numbered advance is applied, then its response is lost. + lose_advance_at: AtomicUsize, +} + +impl ProfileStore for Store { + fn get<'a>( + &'a self, + arn: &'a str, + _region: &'a str, + ) -> Pin, StoreFailed>> + Send + 'a>> { + Box::pin(async move { + Ok(self.values.lock().unwrap().get(arn).map(|value| Stored { + text: Zeroizing::new(value.clone()), + version_id: self.puts.load(Ordering::SeqCst).to_string(), + created: None, + })) + }) + } + + fn get_version<'a>( + &'a self, + arn: &'a str, + _region: &'a str, + version: &'a str, + ) -> Pin, StoreFailed>> + Send + 'a>> { + Box::pin(async move { + Ok(self + .versions + .lock() + .unwrap() + .get(&(arn.to_owned(), version.to_owned())) + .map(|text| Stored { + text: Zeroizing::new(text.clone()), + version_id: version.to_owned(), + created: None, + })) + }) + } + + fn put<'a>( + &'a self, + arn: &'a str, + _region: &'a str, + value: &'a str, + ) -> Pin> + Send + 'a>> { + Box::pin(async move { + let number = self.puts.fetch_add(1, Ordering::SeqCst) + 1; + if self.fail_at.load(Ordering::SeqCst) == number { + return Err(StoreFailed("unavailable")); + } + self.values.lock().unwrap().insert(arn.into(), value.into()); + self.versions + .lock() + .unwrap() + .insert((arn.into(), number.to_string()), value.into()); + Ok(number.to_string()) + }) + } +} + +impl Journal for Store { + fn head<'a>(&'a self, developer: &'a Developer) -> JournalResult<'a, Option> { + Box::pin(async move { Ok(self.heads.lock().unwrap().get(&developer.id).cloned()) }) + } + fn advance<'a>( + &'a self, + developer: &'a Developer, + expected: Option<&'a str>, + revision: &'a str, + ) -> JournalResult<'a, ()> { + Box::pin(async move { + let number = self.advances.fetch_add(1, Ordering::SeqCst) + 1; + let mut heads = self.heads.lock().unwrap(); + if heads.get(&developer.id).map(String::as_str) != expected { + return Err(Error::Storage); + } + heads.insert(developer.id.clone(), revision.to_owned()); + if self.lose_advance_at.load(Ordering::SeqCst) == number { + return Err(Error::Storage); + } + Ok(()) + }) + } +} + +struct Issuer { + calls: AtomicUsize, + response: StdMutex, + entered: Notify, + release: Semaphore, +} + +struct Upstream { + endpoint: String, + issuer: Arc, + task: JoinHandle<()>, +} + +impl Drop for Upstream { + fn drop(&mut self) { + self.task.abort(); + } +} + +async fn upstream() -> Upstream { + let issuer = Arc::new(Issuer { + calls: AtomicUsize::new(0), + response: StdMutex::new(json!({ + "access_token": jwt(json!({"exp": NOW + 7200})), + "id_token": tokens("workspace", NOW + 7200, "unused").id_token, + "refresh_token": "rotated-upstream-secret", + })), + entered: Notify::new(), + release: Semaphore::new(100), + }); + async fn refresh( + RequestState(issuer): RequestState>, + Json(request): Json, + ) -> Json { + assert_eq!(request["client_id"], CLIENT_ID); + assert_eq!(request["grant_type"], "refresh_token"); + assert_eq!(request["refresh_token"], "upstream-secret"); + issuer.calls.fetch_add(1, Ordering::SeqCst); + issuer.entered.notify_one(); + let permit = issuer.release.acquire().await.unwrap(); + permit.forget(); + Json(issuer.response.lock().unwrap().clone()) + } + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}/oauth/token", listener.local_addr().unwrap()); + let app = Router::new() + .route("/oauth/token", post(refresh)) + .with_state(Arc::clone(&issuer)); + let task = spawn(async move { serve(listener, app).await.unwrap() }); + Upstream { + endpoint, + issuer, + task, + } +} + +fn broker(store: Arc, upstream: &Upstream) -> Arc { + Arc::new( + Broker::new( + store.clone(), + store, + OAuth::local(&upstream.endpoint).unwrap(), + [developer()], + "https://codex-refresh.buzz-sandbox-manager.svc.cluster.local:8443/oauth/token".into(), + CA.into(), + ) + .unwrap(), + ) +} + +async fn enroll(broker: &Broker, store: &Store, _previous: Option<&Value>) -> Value { + let developer = developer(); + let _lock = broker.lock(&developer).await.unwrap(); + let metadata = broker + .stage( + &developer, + tokens("workspace", NOW + 3600, "upstream-secret"), + ) + .await + .unwrap(); + let profile = json!({"codex_login": metadata}); + store + .put( + &developer.profile_secret_arn, + &developer.region, + &profile.to_string(), + ) + .await + .unwrap(); + broker.activate(&developer, cookie(&profile)).await.unwrap(); + profile +} + +fn cookie(profile: &Value) -> &str { + profile["codex_login"]["refresh_token"].as_str().unwrap() +} + +#[test] +fn dedicated_login_validation_refuses_api_keys_and_mismatched_workspaces() { + let bundle = tokens("workspace", NOW + 3600, "secret"); + let mut login = json!({"auth_mode": "chatgpt", "OPENAI_API_KEY": null, "tokens": bundle}); + assert!(Tokens::from_login(&login).is_ok()); + login["auth_mode"] = json!("apikey"); + assert!(matches!( + Tokens::from_login(&login), + Err(Error::InvalidLogin) + )); + login["auth_mode"] = json!("chatgpt"); + login["tokens"]["account_id"] = json!("other-workspace"); + assert!(matches!( + Tokens::from_login(&login), + Err(Error::InvalidLogin) + )); + login["tokens"]["account_id"] = json!("workspace"); + login["tokens"]["access_token"] = json!("not-a-jwt"); + assert!(matches!( + Tokens::from_login(&login), + Err(Error::InvalidLogin) + )); +} + +#[async_test] +async fn bootstrap_returns_only_short_lived_tokens_without_rotating_a_fresh_login() { + let upstream = upstream().await; + let store = Arc::new(Store::default()); + let broker = broker(Arc::clone(&store), &upstream); + let profile = enroll(&broker, &store, None).await; + let response = broker.refresh(cookie(&profile), NOW).await.unwrap(); + assert_eq!(response["account_id"], "workspace"); + assert_eq!(upstream.issuer.calls.load(Ordering::SeqCst), 0); + assert!(!response.to_string().contains("upstream-secret")); + assert!(!profile.to_string().contains("upstream-secret")); + assert!(response["refresh_token"] + .as_str() + .unwrap() + .starts_with("alice.")); +} + +#[async_test] +async fn concurrent_sandboxes_and_workers_perform_one_upstream_rotation() { + let upstream = upstream().await; + let store = Arc::new(Store::default()); + let broker = broker(Arc::clone(&store), &upstream); + let profile = enroll(&broker, &store, None).await; + let seeded = broker.refresh(cookie(&profile), NOW).await.unwrap(); + let credential = seeded["refresh_token"].as_str().unwrap().to_string(); + let mut jobs = JoinSet::new(); + for _ in 0..20 { + let (broker, credential) = (Arc::clone(&broker), credential.clone()); + jobs.spawn(async move { broker.refresh(&credential, NOW).await }); + } + let mut responses = Vec::new(); + while let Some(response) = jobs.join_next().await { + responses.push(response.unwrap().unwrap()); + } + assert_eq!(upstream.issuer.calls.load(Ordering::SeqCst), 1); + assert!(responses.iter().all(|r| r == &responses[0])); + assert!(!responses[0].to_string().contains("rotated-upstream-secret")); + assert_ne!(responses[0]["refresh_token"], credential); +} + +#[async_test] +async fn a_fresh_bootstrap_refreshes_an_expired_shared_access_token() { + let upstream = upstream().await; + let store = Arc::new(Store::default()); + let broker = broker(Arc::clone(&store), &upstream); + let profile = enroll(&broker, &store, None).await; + broker.refresh(cookie(&profile), NOW + 3500).await.unwrap(); + assert_eq!(upstream.issuer.calls.load(Ordering::SeqCst), 1); +} + +#[async_test] +async fn unknown_developers_and_wrong_credentials_never_reach_the_issuer() { + let upstream = upstream().await; + let store = Arc::new(Store::default()); + let broker = broker(Arc::clone(&store), &upstream); + let profile = enroll(&broker, &store, None).await; + for credential in [ + "alice.wrong.seed", + "bob.wrong.seed", + "", + "alice.wrong.seed.extra", + ] { + assert_eq!( + broker.refresh(credential, NOW).await.unwrap_err(), + Error::Unauthorized + ); + } + assert!(broker.refresh(cookie(&profile), NOW).await.is_ok()); + assert_eq!(upstream.issuer.calls.load(Ordering::SeqCst), 0); +} + +#[async_test] +async fn staging_a_new_login_preserves_the_active_login_until_profile_commit() { + let upstream = upstream().await; + let store = Arc::new(Store::default()); + let broker = broker(Arc::clone(&store), &upstream); + let old = enroll(&broker, &store, None).await; + broker.refresh(cookie(&old), NOW).await.unwrap(); + let developer = developer(); + let pending = { + let _lock = broker.lock(&developer).await.unwrap(); + broker + .stage( + &developer, + tokens("workspace", NOW + 3600, "new-login-secret"), + ) + .await + .unwrap() + }; + assert!(broker.refresh(cookie(&old), NOW).await.is_ok()); + let new = json!({"codex_login": pending}); + assert_eq!( + broker.refresh(cookie(&new), NOW).await.unwrap_err(), + Error::Unauthorized + ); + store + .put( + &developer.profile_secret_arn, + &developer.region, + &new.to_string(), + ) + .await + .unwrap(); + broker.activate(&developer, cookie(&new)).await.unwrap(); + assert_eq!( + broker.refresh(cookie(&old), NOW).await.unwrap_err(), + Error::Unauthorized + ); + let response = broker.refresh(cookie(&new), NOW).await.unwrap(); + assert!(!response.to_string().contains("new-login-secret")); +} + +#[async_test] +async fn disconnecting_the_profile_denies_new_runs_even_with_a_valid_cached_bundle() { + let upstream = upstream().await; + let store = Arc::new(Store::default()); + let broker = broker(Arc::clone(&store), &upstream); + let profile = enroll(&broker, &store, None).await; + let developer = developer(); + store + .put(&developer.profile_secret_arn, &developer.region, "{}") + .await + .unwrap(); + broker.disconnect(&developer).await.unwrap(); + assert_eq!( + broker.refresh(cookie(&profile), NOW).await.unwrap_err(), + Error::Unauthorized + ); + assert_eq!(upstream.issuer.calls.load(Ordering::SeqCst), 0); +} + +#[async_test] +async fn persistence_failure_before_the_refresh_marker_never_rotates_upstream() { + let upstream = upstream().await; + let store = Arc::new(Store::default()); + let broker = broker(Arc::clone(&store), &upstream); + let profile = enroll(&broker, &store, None).await; + let seeded = broker.refresh(cookie(&profile), NOW).await.unwrap(); + store + .fail_at + .store(store.puts.load(Ordering::SeqCst) + 1, Ordering::SeqCst); + let credential = seeded["refresh_token"].as_str().unwrap(); + assert_eq!( + broker.refresh(credential, NOW).await.unwrap_err(), + Error::Storage + ); + assert_eq!(upstream.issuer.calls.load(Ordering::SeqCst), 0); + assert!(broker.refresh(credential, NOW).await.is_ok()); +} + +#[async_test] +async fn a_lost_write_after_rotation_leaves_a_durable_marker_across_manager_restart() { + let upstream = upstream().await; + let store = Arc::new(Store::default()); + let first = broker(Arc::clone(&store), &upstream); + let profile = enroll(&first, &store, None).await; + let seeded = first.refresh(cookie(&profile), NOW).await.unwrap(); + store + .fail_at + .store(store.puts.load(Ordering::SeqCst) + 2, Ordering::SeqCst); + let credential = seeded["refresh_token"].as_str().unwrap(); + assert_eq!( + first.refresh(credential, NOW).await.unwrap_err(), + Error::Storage + ); + let restarted = broker(Arc::clone(&store), &upstream); + assert_eq!( + restarted.refresh(credential, NOW).await.unwrap_err(), + Error::Interrupted + ); + assert_eq!(upstream.issuer.calls.load(Ordering::SeqCst), 1); + let new = enroll(&restarted, &store, Some(&profile)).await; + assert!(restarted.refresh(cookie(&new), NOW).await.is_ok()); +} + +/// An SDK retry of an applied conditional update fails its own condition. +/// The refresh marker it published is then owned, not stranded. +#[async_test] +async fn a_lost_advance_response_for_an_applied_update_completes_the_refresh() { + let upstream = upstream().await; + let store = Arc::new(Store::default()); + let first = broker(Arc::clone(&store), &upstream); + let profile = enroll(&first, &store, None).await; + let seeded = first.refresh(cookie(&profile), NOW).await.unwrap(); + store + .lose_advance_at + .store(store.advances.load(Ordering::SeqCst) + 1, Ordering::SeqCst); + let credential = seeded["refresh_token"].as_str().unwrap(); + let rotated = first.refresh(credential, NOW).await.unwrap(); + assert_eq!(upstream.issuer.calls.load(Ordering::SeqCst), 1); + // The marker is owned: after a restart the older credential receives the + // refreshed bundle instead of an interrupted-login refusal. + let restarted = broker(Arc::clone(&store), &upstream); + let again = restarted.refresh(credential, NOW).await.unwrap(); + assert_eq!(again["refresh_token"], rotated["refresh_token"]); + assert_eq!(upstream.issuer.calls.load(Ordering::SeqCst), 1); +} + +#[async_test] +async fn cancellation_during_upstream_io_cannot_reuse_a_possibly_consumed_token() { + let upstream = upstream().await; + upstream.issuer.release.forget_permits(100); + let store = Arc::new(Store::default()); + let first = broker(Arc::clone(&store), &upstream); + let profile = enroll(&first, &store, None).await; + let seeded = first.refresh(cookie(&profile), NOW).await.unwrap(); + let credential = seeded["refresh_token"].as_str().unwrap().to_string(); + let job = { + let first = Arc::clone(&first); + let credential = credential.clone(); + spawn(async move { first.refresh(&credential, NOW).await }) + }; + upstream.issuer.entered.notified().await; + job.abort(); + assert!(job.await.unwrap_err().is_cancelled()); + let restarted = broker(Arc::clone(&store), &upstream); + assert_eq!( + restarted.refresh(&credential, NOW).await.unwrap_err(), + Error::Interrupted + ); + assert_eq!(upstream.issuer.calls.load(Ordering::SeqCst), 1); +} + +#[async_test] +async fn an_invalid_upstream_bundle_is_not_persisted_or_echoed() { + let upstream = upstream().await; + *upstream.issuer.response.lock().unwrap() = json!({ + "access_token": "synthetic-private-upstream-response", + "refresh_token": "synthetic-private-rotated-token", + }); + let store = Arc::new(Store::default()); + let broker = broker(Arc::clone(&store), &upstream); + let profile = enroll(&broker, &store, None).await; + let error = broker + .refresh(cookie(&profile), NOW + 3500) + .await + .unwrap_err(); + assert_eq!(error, Error::Upstream); + assert!(!error.to_string().contains("synthetic-private")); + assert_eq!( + broker.refresh(cookie(&profile), NOW).await.unwrap_err(), + Error::Interrupted + ); +} + +#[async_test] +async fn an_uncommitted_secret_version_and_stale_current_value_cannot_bypass_the_journal() { + let upstream = upstream().await; + let store = Arc::new(Store::default()); + let first = broker(Arc::clone(&store), &upstream); + let profile = enroll(&first, &store, None).await; + let seeded = first.refresh(cookie(&profile), NOW).await.unwrap(); + let developer = developer(); + let mut left = first.read(&developer).await.unwrap(); + let mut right = first.read(&developer).await.unwrap(); + left.active.as_mut().unwrap().state = State::Refreshing; + first.write(&developer, &mut left).await.unwrap(); + // A second manager's stale Ready record writes a secret version but loses CAS. + assert_eq!( + first.write(&developer, &mut right).await.unwrap_err(), + Error::Storage + ); + // AWSCURRENT is now Ready, but a restart follows the committed marker version. + let restarted = broker(Arc::clone(&store), &upstream); + assert_eq!( + restarted + .refresh(seeded["refresh_token"].as_str().unwrap(), NOW) + .await + .unwrap_err(), + Error::Interrupted + ); + assert_eq!(upstream.issuer.calls.load(Ordering::SeqCst), 0); +} + +#[async_test] +async fn enrollment_activates_only_after_profile_commit_and_revokes_stale_profile_grants() { + let upstream = upstream().await; + let store = Arc::new(Store::default()); + let broker = broker(Arc::clone(&store), &upstream); + let developer = developer(); + let github = fake_github(&[("ghp_fixtureGitHubTokenDoNotLeak", "ada-l")]).await; + let enrollment = + Enrollment::new(GitHub::new(&github).unwrap(), store.clone()).with_broker(broker.clone()); + let profile = valid("alice", &"a".repeat(64)); + let response = enrollment + .run_with_login( + &developer, + profile.clone(), + Some(Change::Connect(tokens( + "workspace", + NOW + 3600, + "upstream-secret", + ))), + ) + .await; + assert_eq!(response["ok"], true); + assert_eq!(response["profile"]["codex_login"], true); + assert!(!response.to_string().contains("upstream-secret")); + let stored = store + .get(&developer.profile_secret_arn, &developer.region) + .await + .unwrap() + .unwrap(); + assert!(!stored.text.contains("upstream-secret")); + let old: Value = from_str(&stored.text).unwrap(); + assert!(broker.refresh(cookie(&old), NOW).await.is_ok()); + store + .fail_at + .store(store.puts.load(Ordering::SeqCst) + 2, Ordering::SeqCst); + let failed = enrollment + .run_with_login( + &developer, + json!({}), + Some(Change::Connect(tokens( + "workspace", + NOW + 3600, + "replacement-secret", + ))), + ) + .await; + assert_eq!(failed["ok"], false); + assert!(broker.refresh(cookie(&old), NOW).await.is_ok()); + let disconnected = enrollment + .run_with_login(&developer, json!({}), Some(Change::Disconnect)) + .await; + assert_eq!(disconnected["ok"], true); + // A stale profile replica must not resurrect the disconnected grant. + store + .values + .lock() + .unwrap() + .insert(developer.profile_secret_arn.clone(), old.to_string()); + assert_eq!( + broker.refresh(cookie(&old), NOW).await.unwrap_err(), + Error::Unauthorized + ); +} + +#[async_test] +async fn the_http_boundary_limits_requests_and_never_echoes_credentials() { + let upstream = upstream().await; + let store = Arc::new(Store::default()); + let broker = broker(Arc::clone(&store), &upstream); + let profile = enroll(&broker, &store, None).await; + let app = server::router(broker); + for (payload, status) in [ + ( + json!({"client_id": CLIENT_ID, "grant_type": "refresh_token", "refresh_token": cookie(&profile)}), + StatusCode::OK, + ), + ( + json!({"client_id": "other-client", "grant_type": "refresh_token", "refresh_token": "private-invalid-cookie"}), + StatusCode::BAD_REQUEST, + ), + ( + json!({"client_id": CLIENT_ID, "grant_type": "refresh_token", "refresh_token": "private-invalid-cookie"}), + StatusCode::UNAUTHORIZED, + ), + ( + json!({"client_id": CLIENT_ID, "grant_type": "refresh_token", "refresh_token": "x".repeat(2000)}), + StatusCode::PAYLOAD_TOO_LARGE, + ), + ] { + let request = Request::post("/oauth/token") + .header("content-type", "application/json") + .body(Body::from(payload.to_string())) + .unwrap(); + let response = app.clone().oneshot(request).await.unwrap(); + assert_eq!(response.status(), status); + assert_eq!(response.headers()["cache-control"], "no-store"); + let body = response.into_body().collect().await.unwrap().to_bytes(); + let text = String::from_utf8(body.to_vec()).unwrap(); + assert!(!text.contains("upstream-secret")); + assert!(!text.contains("private-invalid-cookie")); + } + let request = Request::post("/oauth/token") + .header("content-type", "application/json") + .body(Body::from( + json!("private-malformed-credential").to_string(), + )) + .unwrap(); + let response = app.oneshot(request).await.unwrap(); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + assert_eq!(response.headers()["cache-control"], "no-store"); + let body = response.into_body().collect().await.unwrap().to_bytes(); + assert!(!String::from_utf8(body.to_vec()) + .unwrap() + .contains("private-malformed")); + assert_eq!(upstream.issuer.calls.load(Ordering::SeqCst), 0); +} + +/// A request arriving while every permit is held waits for one instead of +/// failing the Sandbox that is bootstrapping. +#[async_test] +async fn a_request_during_a_burst_waits_for_a_permit() { + let upstream = upstream().await; + let store = Arc::new(Store::default()); + let broker = broker(Arc::clone(&store), &upstream); + let profile = enroll(&broker, &store, None).await; + let permits = Arc::new(Semaphore::new(1)); + let app = server::router_with(broker, Arc::clone(&permits)); + let held = permits.acquire_owned().await.unwrap(); + spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + drop(held); + }); + let request = Request::post("/oauth/token") + .header("content-type", "application/json") + .body(Body::from( + json!({"client_id": CLIENT_ID, "grant_type": "refresh_token", "refresh_token": cookie(&profile)}) + .to_string(), + )) + .unwrap(); + let response = app.oneshot(request).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); +} diff --git a/crates/buzz-sandbox-reaper/src/enroll.rs b/crates/buzz-sandbox-reaper/src/enroll.rs index 3890854b017..e439c949016 100644 --- a/crates/buzz-sandbox-reaper/src/enroll.rs +++ b/crates/buzz-sandbox-reaper/src/enroll.rs @@ -12,12 +12,21 @@ use std::pin::Pin; use std::sync::Arc; use std::time::Duration; +use aws_config::timeout::TimeoutConfig; +use aws_config::{BehaviorVersion, SdkConfig}; use aws_sdk_secretsmanager::error::{ProvideErrorMetadata, SdkError}; +use aws_sdk_secretsmanager::operation::get_secret_value::GetSecretValueOutput; +use aws_sdk_secretsmanager::{ + config::{Builder, Region}, + primitives::DateTimeFormat, + Client, +}; use buzz_backend_kubernetes::wire::ErrorCode; use nostr::PublicKey; use serde_json::Value; use zeroize::Zeroizing; +use crate::codex_login::{Broker, Change}; use crate::manager::{refusal, Developer}; use crate::profile; @@ -39,7 +48,12 @@ pub fn admit_profile( ) -> Result { let mut profile = payload .as_object() - .filter(|payload| payload.len() == 1) + .filter(|payload| { + payload.contains_key("profile") + && payload + .keys() + .all(|key| key == "profile" || key == "codex_login") + }) .and_then(|payload| payload.get("profile")) .filter(|profile| profile.is_object()) .cloned() @@ -92,6 +106,16 @@ pub trait ProfileStore: Send + Sync { region: &'a str, ) -> Pin, StoreFailed>> + Send + 'a>>; + /// Read an immutable version. Failure must never fall back to AWSCURRENT. + fn get_version<'a>( + &'a self, + _arn: &'a str, + _region: &'a str, + _version: &'a str, + ) -> Pin, StoreFailed>> + Send + 'a>> { + Box::pin(async { Err(StoreFailed("version_reads_not_supported")) }) + } + /// Write `profile` as the new current version of the secret `arn` in /// `region`, and return the version id. fn put<'a>( @@ -105,17 +129,17 @@ pub trait ProfileStore: Send + Sync { /// Secrets Manager `GetSecretValue` and `PutSecretValue` with the ambient /// credentials (IRSA in-cluster). It writes no files. pub struct SecretsManagerStore { - shared: aws_config::SdkConfig, + shared: SdkConfig, } impl SecretsManagerStore { pub async fn new() -> Self { - let timeouts = aws_config::timeout::TimeoutConfig::builder() + let timeouts = TimeoutConfig::builder() .connect_timeout(Duration::from_secs(5)) .operation_attempt_timeout(Duration::from_secs(10)) .operation_timeout(Duration::from_secs(25)) .build(); - let shared = aws_config::defaults(aws_config::BehaviorVersion::latest()) + let shared = aws_config::defaults(BehaviorVersion::latest()) .timeout_config(timeouts) .load() .await; @@ -124,13 +148,11 @@ impl SecretsManagerStore { } impl SecretsManagerStore { - fn client(&self, region: &str) -> aws_sdk_secretsmanager::Client { - let config = aws_sdk_secretsmanager::config::Builder::from(&self.shared) - .region(aws_sdk_secretsmanager::config::Region::new( - region.to_string(), - )) + fn client(&self, region: &str) -> Client { + let config = Builder::from(&self.shared) + .region(Region::new(region.to_string())) .build(); - aws_sdk_secretsmanager::Client::from_conf(config) + Client::from_conf(config) } } @@ -157,18 +179,26 @@ impl ProfileStore for SecretsManagerStore { } Err(error) => return Err(store_error(error)), }; - let created = output.created_date().and_then(|at| { - at.fmt(aws_sdk_secretsmanager::primitives::DateTimeFormat::DateTime) - .ok() - }); - match (output.secret_string(), output.version_id()) { - (Some(text), Some(version_id)) => Ok(Some(Stored { - text: Zeroizing::new(text.to_string()), - version_id: version_id.to_string(), - created, - })), - _ => Ok(None), - } + Ok(stored_value(output)) + }) + } + + fn get_version<'a>( + &'a self, + arn: &'a str, + region: &'a str, + version: &'a str, + ) -> Pin, StoreFailed>> + Send + 'a>> { + Box::pin(async move { + let output = self + .client(region) + .get_secret_value() + .secret_id(arn) + .version_id(version) + .send() + .await + .map_err(store_error)?; + Ok(stored_value(output)) }) } @@ -195,6 +225,17 @@ impl ProfileStore for SecretsManagerStore { } } +fn stored_value(output: GetSecretValueOutput) -> Option { + let created = output + .created_date() + .and_then(|at| at.fmt(DateTimeFormat::DateTime).ok()); + Some(Stored { + text: Zeroizing::new(output.secret_string()?.to_owned()), + version_id: output.version_id()?.to_owned(), + created, + }) +} + /// A label for an SDK error. The error itself is never logged or returned. fn store_error(error: SdkError) -> StoreFailed { StoreFailed(match &error { @@ -272,11 +313,21 @@ impl GitHub { pub struct Enrollment { github: GitHub, store: Arc, + broker: Option>, } impl Enrollment { pub fn new(github: GitHub, store: Arc) -> Self { - Self { github, store } + Self { + github, + store, + broker: None, + } + } + + pub fn with_broker(mut self, broker: Arc) -> Self { + self.broker = Some(broker); + self } /// The developer's stored profile, parsed. An unreadable stored value @@ -303,6 +354,29 @@ impl Enrollment { /// version id and its [`profile::summary`]. Every profile copy is /// dropped, and the serialized one zeroized, before this returns. pub async fn run(&self, developer: &Developer, partial: Value) -> Value { + self.run_with_login(developer, partial, None).await + } + + /// Enroll or revoke a dedicated login under the same lock as token refresh. + pub async fn run_with_login( + &self, + developer: &Developer, + partial: Value, + login: Option, + ) -> Value { + let _guard = match &self.broker { + Some(broker) => match broker.lock(developer).await { + Ok(guard) => Some(guard), + Err(error) => return refusal(ErrorCode::ProfileStoreFailed, &error.to_string()), + }, + None if matches!(login, Some(Change::Connect(_))) => { + return refusal( + ErrorCode::InvalidProfile, + "ChatGPT login is not configured on this manager", + ); + } + None => None, + }; let stored = match self.stored(developer).await { Ok(stored) => stored.map(|(profile, _)| profile), Err(refused) => return refused, @@ -318,7 +392,38 @@ impl Enrollment { ); } let new_token = stored.is_none() || partial.get("github").is_some(); - let profile = profile::merge(stored, partial); + let connects_codex = matches!(login, Some(Change::Connect(_))); + let disconnects_codex = matches!(login, Some(Change::Disconnect)); + let replaces_codex = partial + .pointer("/models/openai_api_key") + .is_some_and(Value::is_string) + || partial + .pointer("/models/codex_access_token") + .is_some_and(Value::is_string) + || partial.get("codex_login").is_some_and(Value::is_null); + if self.broker.is_none() + && (replaces_codex || disconnects_codex) + && stored + .as_ref() + .is_some_and(|p| p.get("codex_login").is_some()) + { + return refusal( + ErrorCode::ProfileStoreFailed, + "ChatGPT broker is unavailable; enable it before disconnecting this login", + ); + } + let mut profile = profile::merge(stored, partial); + if replaces_codex || login.is_some() { + if let Some(profile) = profile.as_object_mut() { + profile.remove("codex_login"); + } + } + if matches!(login, Some(Change::Connect(_))) { + if let Some(models) = profile["models"].as_object_mut() { + models.remove("openai_api_key"); + models.remove("codex_access_token"); + } + } if let Err(profile::Invalid(message)) = profile::validate(&profile) { return refusal(ErrorCode::InvalidProfile, message); } @@ -338,6 +443,30 @@ impl Enrollment { Err(message) => return refusal(ErrorCode::GithubLoginMismatch, message), } } + if let Some(Change::Connect(tokens)) = login { + let Some(broker) = &self.broker else { + return refusal( + ErrorCode::InvalidProfile, + "ChatGPT login is not configured on this manager", + ); + }; + match broker.stage(developer, tokens).await { + Ok(metadata) => profile["codex_login"] = metadata, + Err(error) => return refusal(ErrorCode::ProfileStoreFailed, &error.to_string()), + } + } + if let Err(profile::Invalid(message)) = profile::validate(&profile) { + return refusal(ErrorCode::InvalidProfile, message); + } + // Revoke before changing the profile: stale Secrets Manager replicas + // may still serve its old broker credential, which must no longer work. + if !connects_codex && (disconnects_codex || replaces_codex) { + if let Some(broker) = &self.broker { + if let Err(error) = broker.disconnect(developer).await { + return refusal(ErrorCode::ProfileStoreFailed, &error.to_string()); + } + } + } let text = Zeroizing::new(profile.to_string()); match self .store @@ -345,6 +474,16 @@ impl Enrollment { .await { Ok(version_id) => { + if let Some(broker) = &self.broker { + if let Some(credential) = profile + .pointer("/codex_login/refresh_token") + .and_then(Value::as_str) + { + if let Err(error) = broker.activate(developer, credential).await { + return refusal(ErrorCode::ProfileStoreFailed, &error.to_string()); + } + } + } let now = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); let summary = profile::summary(&profile, &version_id, Some(now)); serde_json::json!({"ok": true, "version_id": version_id, "profile": summary}) diff --git a/crates/buzz-sandbox-reaper/src/main.rs b/crates/buzz-sandbox-reaper/src/main.rs index 6d7823e24cc..788f1ed8f47 100644 --- a/crates/buzz-sandbox-reaper/src/main.rs +++ b/crates/buzz-sandbox-reaper/src/main.rs @@ -9,6 +9,7 @@ //! `serve` is the session manager (§Relay transport): it runs developers' //! provider requests, received over the Buzz relay, in their namespaces. +mod codex_login; mod config; mod enroll; mod evidence; @@ -56,7 +57,7 @@ enum Command { /// Reconcile Agent Sandbox session lifecycles. Reconcile(ReconcileArgs), /// Serve Sandbox session requests received over the Buzz relay. - Serve(serve::ServeArgs), + Serve(Box), } #[derive(Args)] @@ -110,7 +111,7 @@ async fn main() -> Result<()> { .init(); match Cli::parse().command { Command::Reconcile(args) => reconcile(args).await, - Command::Serve(args) => serve::serve(args).await, + Command::Serve(args) => serve::serve(*args).await, } } diff --git a/crates/buzz-sandbox-reaper/src/manager.rs b/crates/buzz-sandbox-reaper/src/manager.rs index 112f8be9408..96dac11553a 100644 --- a/crates/buzz-sandbox-reaper/src/manager.rs +++ b/crates/buzz-sandbox-reaper/src/manager.rs @@ -7,6 +7,7 @@ //! Nothing here logs: the caller logs an [`Admission`]'s event id, op and //! code, never content. +use crate::codex_login::Change; use std::collections::{BTreeMap, HashMap}; use std::sync::Mutex; @@ -229,6 +230,11 @@ impl ManagerConfig { pub fn len(&self) -> usize { self.developers.len() } + + /// Bind the broker to the same configured developers as the relay. + pub fn developers(&self) -> impl Iterator + '_ { + self.developers.values().cloned() + } } /// Who a reply goes to: the request's signer, `e`-tagging its event. @@ -266,6 +272,7 @@ pub enum Work { Enroll { developer: Developer, profile: serde_json::Value, + login: Option, }, /// The `profile` operation: describe the developer's stored profile. Profile { developer: Developer }, @@ -311,9 +318,19 @@ impl Work { Err(error) => error, } } - Work::Enroll { developer, profile } => { - manager.enrollment.run(&developer, profile).await - } + Work::Enroll { + developer, + profile, + login, + } => match login { + Some(login) => { + manager + .enrollment + .run_with_login(&developer, profile, Some(login)) + .await + } + None => manager.enrollment.run(&developer, profile).await, + }, Work::Profile { developer } => manager.enrollment.describe(&developer).await, } } @@ -470,10 +487,15 @@ impl Manager { // 3. The payload acts for the signer: its agent key, or for `enroll` // a valid profile of the signing owner's developer. let work = if op == Op::Enroll { + let login = match Change::parse(opened.request.payload.get("codex_login")) { + Ok(login) => login, + Err(error) => return refused(ErrorCode::InvalidProfile, &error.to_string()), + }; match enroll::admit_profile(&opened.request.payload, &owner, developer) { Ok(profile) => Work::Enroll { developer: developer.clone(), profile, + login, }, Err((code, message)) => return refused(code, message), } diff --git a/crates/buzz-sandbox-reaper/src/manager_tests.rs b/crates/buzz-sandbox-reaper/src/manager_tests.rs index de5d2bc883d..f0d02a616ae 100644 --- a/crates/buzz-sandbox-reaper/src/manager_tests.rs +++ b/crates/buzz-sandbox-reaper/src/manager_tests.rs @@ -389,7 +389,10 @@ async fn enroll_runs_for_the_signing_owners_developer() { let Admission::Start { job, .. } = manager.admit(&sent.event(), now()) else { panic!("not started"); }; - let Work::Enroll { developer, profile } = job.work else { + let Work::Enroll { + developer, profile, .. + } = job.work + else { panic!("not an enrollment"); }; assert_eq!(developer.id, "another-dev"); diff --git a/crates/buzz-sandbox-reaper/src/profile.rs b/crates/buzz-sandbox-reaper/src/profile.rs index d3c4d17ce24..d23490cdfc8 100644 --- a/crates/buzz-sandbox-reaper/src/profile.rs +++ b/crates/buzz-sandbox-reaper/src/profile.rs @@ -5,6 +5,7 @@ //! rule. Refusals are the image's messages, which never include a value. use serde_json::{Map, Value}; +use url::Url; /// Largest profile, measured as Python's `json.dumps` writes it: the /// Secrets Manager `SecretString` limit. @@ -111,7 +112,15 @@ fn truthy(value: Option<&toml::Value>) -> bool { /// `profile.py`'s `validate(profile)`, in its order. pub fn validate(profile: &Value) -> Result<(), Invalid> { - let top = fields(profile, FIELDS)?; + let top = profile + .as_object() + .filter(|map| { + FIELDS.iter().all(|field| map.contains_key(*field)) + && map + .keys() + .all(|field| FIELDS.contains(&field.as_str()) || field == "codex_login") + }) + .ok_or(Invalid("Profile fields are incomplete or unsupported"))?; let version = &top["version"]; require( (version.is_u64() || version.is_i64()) && version.as_i64() == Some(1), @@ -166,6 +175,13 @@ pub fn validate(profile: &Value) -> Result<(), Invalid> { .as_object() .filter(|models| models.keys().all(|k| MODEL_FIELDS.contains(&k.as_str()))) .ok_or(Invalid("Unsupported model credential type"))?; + if let Some(login) = top.get("codex_login") { + validate_codex_login(login, top["developer"].as_str().unwrap_or_default())?; + require( + !models.contains_key("openai_api_key") && !models.contains_key("codex_access_token"), + "Choose a ChatGPT login, an OpenAI API key, or a Codex access token", + )?; + } // Codex reads the access token before its saved API-key login, so allow only one. require( !(models.contains_key("openai_api_key") && models.contains_key("codex_access_token")), @@ -191,6 +207,51 @@ pub fn validate(profile: &Value) -> Result<(), Invalid> { ) } +/// Broker metadata only. The upstream refresh token belongs in the +/// manager-only login secret, never this agent-readable profile. +fn validate_codex_login(value: &Value, developer: &str) -> Result<(), Invalid> { + let login = fields(value, &["url", "ca_certificate", "refresh_token"])?; + let url = login["url"] + .as_str() + .and_then(|value| Url::parse(value).ok()) + .ok_or(Invalid("Invalid ChatGPT broker URL"))?; + require( + url.scheme() == "https" + && url.host_str() == Some("codex-refresh.buzz-sandbox-manager.svc.cluster.local") + && url.port() == Some(8443) + && url.path() == "/oauth/token" + && url.username().is_empty() + && url.password().is_none() + && url.query().is_none() + && url.fragment().is_none(), + "Invalid ChatGPT broker URL", + )?; + require( + login["ca_certificate"].as_str().is_some_and(|certificate| { + certificate.len() <= 8192 + && certificate.starts_with("-----BEGIN CERTIFICATE-----\n") + && certificate + .trim_end() + .ends_with("-----END CERTIFICATE-----") + }), + "Invalid ChatGPT broker certificate", + )?; + token(&login["refresh_token"])?; + let credential = login["refresh_token"].as_str().unwrap_or_default(); + let prefix = format!("{developer}."); + require( + credential.strip_prefix(&prefix).is_some_and(|body| { + body.strip_suffix(".seed").is_some_and(|random| { + random.len() == 32 + && random + .bytes() + .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase()) + }) + }), + "Invalid ChatGPT broker credential", + ) +} + /// The byte length of Python's `json.dumps(value)`: `", "` and `": "` /// separators, and every character outside `' '..='~'` escaped, as `\uXXXX` /// (a surrogate pair above U+FFFF) unless it has a short escape. @@ -245,7 +306,16 @@ const EXCLUSIVE_MODELS: &[(&str, &str)] = &[ pub fn check_partial(profile: &Value) -> Result<(), Invalid> { const INCOMPLETE: &str = "Profile fields are incomplete or unsupported"; let top = profile.as_object().ok_or(Invalid(INCOMPLETE))?; - require(top.keys().all(|k| FIELDS.contains(&k.as_str())), INCOMPLETE)?; + require( + top.keys() + .all(|k| FIELDS.contains(&k.as_str()) || k == "codex_login"), + INCOMPLETE, + )?; + // Clients may disconnect, but only the manager creates broker metadata. + require( + top.get("codex_login").is_none_or(Value::is_null), + "ChatGPT broker metadata is managed by the Sandbox manager", + )?; let version = top.get("version").unwrap_or(&Value::Null); require( (version.is_u64() || version.is_i64()) && version.as_i64() == Some(1), @@ -336,7 +406,7 @@ pub fn summary(profile: &Value, version_id: &str, updated_at: Option) -> .map(|models| models.keys().collect()) .unwrap_or_default(); models.sort(); - serde_json::json!({ + let mut summary = serde_json::json!({ "version_id": version_id, "updated_at": updated_at, "git": { @@ -347,7 +417,11 @@ pub fn summary(profile: &Value, version_id: &str, updated_at: Option) -> "github": {"login": text("/github/login"), "token": present("/github/token")}, "via": {"profile": present("/via/profile"), "op_token": present("/via/op_token")}, "models": models, - }) + }); + if profile.get("codex_login").is_some_and(Value::is_object) { + summary["codex_login"] = Value::Bool(true); + } + summary } #[cfg(test)] diff --git a/crates/buzz-sandbox-reaper/src/serve.rs b/crates/buzz-sandbox-reaper/src/serve.rs index 26345f3e755..e77a9180cbe 100644 --- a/crates/buzz-sandbox-reaper/src/serve.rs +++ b/crates/buzz-sandbox-reaper/src/serve.rs @@ -6,6 +6,7 @@ //! Logs carry only event ids, operations and codes. `buzz_ws_client` logs //! whole frames at debug level, so `main` caps that target at info. +use crate::codex_login::server::{Options as CodexLoginOptions, Server as CodexLoginServer}; use std::future::Future; use std::panic::AssertUnwindSafe; use std::path::PathBuf; @@ -22,9 +23,12 @@ use futures_util::FutureExt; use nostr::{Event, Keys, Tag, Timestamp}; use tokio::sync::mpsc; use tokio::task::JoinSet; +use tokio::{join, select}; +use tokio_util::sync::CancellationToken; use crate::enroll::{self, Enrollment, SecretsManagerStore}; use crate::manager::{response_value, Admission, Job, Manager, ManagerConfig, Waiter}; +use crate::shutdown; /// Bound on connecting, authenticating and subscribing. const CONNECT_TIMEOUT: Duration = Duration::from_secs(30); @@ -39,6 +43,8 @@ const SUBSCRIPTION: &str = "session-control"; #[derive(Args)] pub struct ServeArgs { + #[command(flatten)] + codex: CodexLoginOptions, /// Path to the JSON config (ConfigMap `sandbox-manager-config`, key `config.json`). #[arg( long, @@ -96,7 +102,12 @@ pub async fn serve(args: ServeArgs) -> Result<()> { .map_err(|_| anyhow::anyhow!("BUZZ_AUTH_TAG is not a valid owner attestation"))?; let github = enroll::GitHub::new(&args.github_api_url) .map_err(|e| anyhow::anyhow!("BUZZ_GITHUB_API_URL: {e}"))?; - let enrollment = Enrollment::new(github, Arc::new(SecretsManagerStore::new().await)); + let store = Arc::new(SecretsManagerStore::new().await); + let broker = CodexLoginServer::prepare(args.codex, store.clone(), config.developers()).await?; + let mut enrollment = Enrollment::new(github, store); + if let Some(server) = &broker { + enrollment = enrollment.with_broker(Arc::clone(&server.broker)); + } let client = kube::Client::try_default() .await .context("build in-cluster Kubernetes client")?; @@ -105,7 +116,23 @@ pub async fn serve(args: ServeArgs) -> Result<()> { Arc::new(Manager::new(keys, config, client, enrollment).with_sandbox_defaults(defaults)); tracing::info!(manager = %manager.keys().public_key(), developers, "buzz-sandbox-reaper serve started"); - run(manager, &args.relay_url, auth_tag, crate::shutdown()).await; + if let Some(broker) = broker { + let stop = CancellationToken::new(); + let relay = async { + run(manager, &args.relay_url, auth_tag, stop.cancelled()).await; + stop.cancel(); + }; + let signal = async { + select! { + () = shutdown() => stop.cancel(), + () = stop.cancelled() => {}, + } + }; + let (result, (), ()) = join!(broker.run(stop.clone()), relay, signal); + result?; + } else { + run(manager, &args.relay_url, auth_tag, shutdown()).await; + } Ok(()) } @@ -134,7 +161,7 @@ pub async fn run( tokio::pin!(shutdown); let mut backoff = MIN_BACKOFF; loop { - let connected = tokio::select! { + let connected = select! { () = &mut shutdown => break, connected = tokio::time::timeout( CONNECT_TIMEOUT, @@ -158,7 +185,7 @@ pub async fn run( } Err(error) => tracing::warn!(error, "could not connect to the relay"), } - tokio::select! { + select! { () = &mut shutdown => break, () = tokio::time::sleep(backoff) => {} } @@ -218,7 +245,7 @@ impl Session<'_> { shutdown: &mut std::pin::Pin<&mut impl Future>, ) -> Option<&'static str> { loop { - tokio::select! { + select! { () = shutdown.as_mut() => return None, Some(reply) = queued.recv() => { if let Err(error) = self.publish(reply).await { diff --git a/voltage/images/buzz-agent/README.md b/voltage/images/buzz-agent/README.md index 8168d9975f2..8e84aa50a8e 100644 --- a/voltage/images/buzz-agent/README.md +++ b/voltage/images/buzz-agent/README.md @@ -184,3 +184,24 @@ The test prints only counts and elapsed time. It does not persist credentials ou Each merge to `main` that changes the image inputs publishes this image to `voltage-buzz-agent-staging`, tagged with the full commit SHA. See [../README.md](../README.md) for the workflow, the digest to pin, and where to pin it. The voltage-terraform admission policy admits only digest-pinned images from `voltage-buzz-agent-staging` in developer namespaces. + +## Shared ChatGPT login + +A profile may select `codex_login` instead of `openai_api_key` or `codex_access_token`. +Its fields are the fixed internal broker URL, its public CA certificate, and a broker credential. +The real upstream refresh token stays with the manager. +Both interactive and worker bootstrap call the broker before starting their harness. +Revoked or unavailable credentials stop startup with a fixed refusal that contains no token values. + +Bootstrap writes a private `.codex/auth.json` with short-lived tokens and a versioned broker credential, +plus `config.toml` selecting file storage. It sets `CODEX_REFRESH_TOKEN_URL_OVERRIDE` and +`CODEX_CA_CERTIFICATE` (system trust roots plus the broker CA). These files are excluded from checkpoints. +Codex 0.159.3 refreshes through that override and persists the returned broker credential. +Do not upgrade Codex without checking this contract. See the +[broker rollout and recovery guide](../buzz-sandbox-reaper/codex-login.md). + +Run the local TLS bootstrap test with synthetic credentials: + +```sh +python3 voltage/images/buzz-agent/test-codex-login.py buzz-agent:local +``` diff --git a/voltage/images/buzz-agent/profile.py b/voltage/images/buzz-agent/profile.py index aada8dd2ee7..cd1b433ff60 100644 --- a/voltage/images/buzz-agent/profile.py +++ b/voltage/images/buzz-agent/profile.py @@ -1,14 +1,19 @@ #!/usr/bin/env python3 """Load the developer profile the Sandbox manager enrolled into that developer's Sandboxes.""" import argparse +import base64 +from datetime import datetime, timezone import json import os from pathlib import Path import re import stat +import ssl import subprocess import sys import tomllib +from urllib import request as http +from urllib.parse import urlsplit MAX_BYTES = 65536 DIRECTORY = Path('/home/agent/.config/buzz-agent') @@ -17,7 +22,8 @@ 'openrouter_api_key'} # Inherited values would override or mix with the profile's selected accounts. INHERITED_MODEL_VARIABLES = ('OPENAI_API_KEY', 'CODEX_API_KEY', 'CODEX_ACCESS_TOKEN', 'ANTHROPIC_API_KEY', - 'ANTHROPIC_AUTH_TOKEN', 'CLAUDE_CODE_OAUTH_TOKEN', 'OPENROUTER_API_KEY') + 'ANTHROPIC_AUTH_TOKEN', 'CLAUDE_CODE_OAUTH_TOKEN', 'OPENROUTER_API_KEY', + 'CODEX_REFRESH_TOKEN_URL_OVERRIDE', 'CODEX_CA_CERTIFICATE') # The shared Depot cache credential: the via profile's `depot` service, read from the vault. DEPOT_SERVICE = 'depot' # The developer's agent-only commit signing key, registered on their GitHub account at enrollment. @@ -51,7 +57,8 @@ def token(value): def validate(profile): - fields(profile, FIELDS) + require(isinstance(profile, dict) and FIELDS <= set(profile) + and set(profile) <= FIELDS | {'codex_login'}, 'Profile fields are incomplete or unsupported') require(type(profile['version']) is int and profile['version'] == 1, 'Unsupported profile version') require(isinstance(profile['developer'], str) and re.fullmatch(r'[a-z0-9](?:[a-z0-9-]{0,30}[a-z0-9])?', profile['developer']), 'Invalid developer ID') @@ -74,6 +81,10 @@ def validate(profile): require(config.get('services') and config.get('providers'), 'via profile requires providers and services') require(isinstance(profile['models'], dict) and set(profile['models']) <= MODEL_FIELDS, 'Unsupported model credential type') + if 'codex_login' in profile: + validate_codex_login(profile['codex_login'], profile['developer']) + require(not {'openai_api_key', 'codex_access_token'} & set(profile['models']), + 'Choose a ChatGPT login, an OpenAI API key, or a Codex access token') # Codex reads the access token before its saved API-key login, so allow only one. require(not {'openai_api_key', 'codex_access_token'} <= set(profile['models']), 'Choose either an OpenAI API key or a Codex access token') @@ -87,6 +98,70 @@ def validate(profile): require(len(json.dumps(profile).encode()) <= MAX_BYTES, 'Profile exceeds Secrets Manager size limit') +def validate_codex_login(login, developer): + fields(login, {'url', 'ca_certificate', 'refresh_token'}) + require(isinstance(login['url'], str), 'Invalid ChatGPT broker URL') + url = urlsplit(login['url']) + require(url.scheme == 'https' + and url.hostname == 'codex-refresh.buzz-sandbox-manager.svc.cluster.local' + and url.port == 8443 and url.path == '/oauth/token' + and url.username is None and url.password is None + and not url.query and not url.fragment, 'Invalid ChatGPT broker URL') + certificate = login['ca_certificate'] + require(isinstance(certificate, str) and len(certificate) <= 8192 + and certificate.startswith('-----BEGIN CERTIFICATE-----\n') + and certificate.rstrip().endswith('-----END CERTIFICATE-----'), 'Invalid ChatGPT broker certificate') + token(login['refresh_token']) + require(re.fullmatch(re.escape(developer) + r'\.[a-f0-9]{32}\.seed', login['refresh_token']), + 'Invalid ChatGPT broker credential') + + +class NoRedirect(http.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + +def configure_codex_login(login, env): + """Validate the broker grant before every new Sandbox or worker starts.""" + step('loading the shared ChatGPT login') + try: + context = ssl.create_default_context(cadata=login['ca_certificate']) + # Ignore ambient HTTP proxies for this private, authenticated endpoint. + opener = http.build_opener(http.ProxyHandler({}), http.HTTPSHandler(context=context), NoRedirect()) + body = json.dumps({'client_id': 'app_EMoamEEZ73f0CkXaXp7hrann', + 'grant_type': 'refresh_token', + 'refresh_token': login['refresh_token']}).encode() + req = http.Request(login['url'], data=body, headers={'Content-Type': 'application/json'}) + with opener.open(req, timeout=60) as response: + raw = response.read(MAX_BYTES + 1) + require(len(raw) <= MAX_BYTES, 'Invalid ChatGPT broker response') + refreshed = json.loads(raw) + for key in ('access_token', 'id_token', 'refresh_token', 'account_id'): + token(refreshed[key]) + require(refreshed['refresh_token'].rsplit('.', 1)[0] + == login['refresh_token'].rsplit('.', 1)[0], 'Invalid ChatGPT broker response') + claims = json.loads(base64.urlsafe_b64decode(refreshed['id_token'].split('.')[1] + '===')) + require(claims['https://api.openai.com/auth']['chatgpt_account_id'] == refreshed['account_id'], + 'Invalid ChatGPT broker response') + except (KeyError, ValueError, TypeError, OSError, IndexError, ssl.SSLError): + raise Refused('Shared ChatGPT login unavailable; enroll again if it was disconnected or interrupted') from None + directory = Path('/home/agent/.codex') + write_file(directory / 'config.toml', 'cli_auth_credentials_store = "file"\n') + write_file(directory / 'auth.json', json.dumps({ + 'auth_mode': 'chatgpt', 'OPENAI_API_KEY': None, + 'tokens': {key: refreshed[key] for key in ('id_token', 'access_token', 'refresh_token', 'account_id')}, + 'last_refresh': datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z'), + })) + # Keep public roots as well as the private broker CA: Codex uses this + # bundle for its upstream HTTPS connections too. + system_ca = ssl.get_default_verify_paths().cafile + roots = Path(system_ca).read_text() if system_ca else '' + certificate_file = DIRECTORY / 'codex-ca.pem' + write_file(certificate_file, roots + '\n' + login['ca_certificate']) + env.update(CODEX_HOME=str(directory), CODEX_CONFIG='{"cli_auth_credentials_store":"file"}', + CODEX_REFRESH_TOKEN_URL_OVERRIDE=login['url'], CODEX_CA_CERTIFICATE=str(certificate_file)) + + def secret_region(arn, developer): match = re.fullmatch(r'arn:aws:secretsmanager:([a-z0-9-]+):[0-9]{12}:secret:voltage/staging/agents/' + re.escape(developer) + r'/profile-[a-zA-Z0-9]{6}', arn) @@ -191,6 +266,8 @@ def bootstrap(): if 'codex_access_token' in profile['models']: # A workspace access token: no refresh token, so concurrent Sandboxes share it safely. env['CODEX_ACCESS_TOKEN'] = profile['models']['codex_access_token'] + if 'codex_login' in profile: + configure_codex_login(profile['codex_login'], env) if 'anthropic_api_key' in profile['models']: env['ANTHROPIC_API_KEY'] = profile['models']['anthropic_api_key'] if 'openrouter_api_key' in profile['models']: diff --git a/voltage/images/buzz-agent/test-codex-login.py b/voltage/images/buzz-agent/test-codex-login.py new file mode 100644 index 00000000000..b6a54f97c59 --- /dev/null +++ b/voltage/images/buzz-agent/test-codex-login.py @@ -0,0 +1,204 @@ +"""Exercise agent bootstrap against a local TLS broker with synthetic tokens only.""" +import base64 +from copy import deepcopy +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import importlib.util +import json +import os +from queue import Queue, Empty +from pathlib import Path +import ssl +import shutil +import signal +import sys +import subprocess +import tempfile +from threading import Thread +import unittest +from unittest.mock import patch + +spec = importlib.util.spec_from_file_location('agent_profile', Path(__file__).with_name('profile.py')) +profile = importlib.util.module_from_spec(spec) +spec.loader.exec_module(profile) + + +def jwt(claims): + return 'e30.' + base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=') + '.fixture' + + +class BootstrapTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.directory = tempfile.TemporaryDirectory() + directory = Path(cls.directory.name) + cert, key = directory / 'cert.pem', directory / 'key.pem' + ca, ca_key, csr = directory / 'ca.pem', directory / 'ca-key.pem', directory / 'request.pem' + extensions = directory / 'extensions' + extensions.write_text('subjectAltName=DNS:localhost\nbasicConstraints=critical,CA:FALSE\n' + 'keyUsage=critical,digitalSignature,keyEncipherment\nextendedKeyUsage=serverAuth\n' + 'authorityKeyIdentifier=keyid,issuer\nsubjectKeyIdentifier=hash\n') + for args in [ + ['req', '-x509', '-newkey', 'rsa:2048', '-nodes', '-keyout', str(ca_key), + '-out', str(ca), '-days', '1', '-subj', '/CN=Fixture CA', '-addext', 'basicConstraints=critical,CA:TRUE', + '-addext', 'keyUsage=critical,keyCertSign,cRLSign'], + ['req', '-newkey', 'rsa:2048', '-nodes', '-keyout', str(key), '-out', str(csr), '-subj', '/CN=localhost'], + ['x509', '-req', '-in', str(csr), '-CA', str(ca), '-CAkey', str(ca_key), '-CAcreateserial', + '-out', str(cert), '-days', '1', '-extfile', str(extensions)], + ]: + subprocess.run(['openssl', *args], check=True, capture_output=True) + cls.ca = ca.read_text() + cls.cookie = 'alice.' + 'a' * 32 + '.seed' + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *_): + pass + + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers['content-length']))) + cls.requests.append(body) + self.send_response(cls.status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(cls.response))) + if cls.status == 302: + self.send_header('Location', 'https://attacker.invalid') + self.end_headers() + self.wfile.write(cls.response) + + cls.server = ThreadingHTTPServer(('localhost', 0), Handler) + context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.load_cert_chain(cert, key) + cls.server.socket = context.wrap_socket(cls.server.socket, server_side=True) + cls.thread = Thread(target=cls.server.serve_forever, daemon=True) + cls.thread.start() + + @classmethod + def tearDownClass(cls): + cls.server.shutdown() + cls.server.server_close() + cls.thread.join() + cls.directory.cleanup() + + def setUp(self): + cls = type(self) + cls.requests = [] + cls.status = 200 + self.bundle = { + 'access_token': jwt({'exp': 2_000_000_000}), + 'id_token': jwt({'https://api.openai.com/auth': {'chatgpt_account_id': 'workspace'}}), + 'account_id': 'workspace', 'refresh_token': self.cookie.rsplit('.', 1)[0] + '.version-1', + } + cls.response = json.dumps(self.bundle).encode() + self.login = {'url': f'https://localhost:{self.server.server_port}/oauth/token', + 'ca_certificate': self.ca, 'refresh_token': self.cookie} + self.files = {} + self.writer = patch.object(profile, 'write_file', side_effect=lambda path, text: self.files.__setitem__(str(path), text)) + self.writer.start() + self.addCleanup(self.writer.stop) + + def test_bootstrap_persists_only_broker_credentials_and_checks_tls_even_with_a_proxy(self): + env = {} + with patch.dict('os.environ', {'HTTPS_PROXY': 'http://127.0.0.1:1'}): + profile.configure_codex_login(self.login, env) + auth = json.loads(self.files['/home/agent/.codex/auth.json']) + self.assertEqual(auth['tokens'], self.bundle) + self.assertEqual(self.requests[0]['refresh_token'], self.cookie) + self.assertEqual(env['CODEX_REFRESH_TOKEN_URL_OVERRIDE'], self.login['url']) + self.assertIn(self.ca, self.files[env['CODEX_CA_CERTIFICATE']]) + self.assertNotIn('upstream-refresh-secret', json.dumps(self.files)) + + def test_a_revoked_grant_stops_boot_before_any_credential_file_is_written(self): + type(self).status = 401 + type(self).response = b'private-error-must-not-display' + with self.assertRaises(profile.Refused) as refusal: + profile.configure_codex_login(self.login, {}) + self.assertNotIn('private-error', str(refusal.exception)) + self.assertEqual(self.files, {}) + + def test_redirects_untrusted_tls_bad_workspaces_and_oversized_replies_are_refused(self): + for scenario in ('redirect', 'untrusted-ca', 'workspace', 'oversized'): + with self.subTest(scenario=scenario): + login = deepcopy(self.login) + type(self).status = 302 if scenario == 'redirect' else 200 + bundle = deepcopy(self.bundle) + if scenario == 'workspace': + bundle['account_id'] = 'other-workspace' + type(self).response = b'x' * 65537 if scenario == 'oversized' else json.dumps(bundle).encode() + if scenario == 'untrusted-ca': + login['ca_certificate'] = 'invalid-ca-private-fixture' + with self.assertRaises(profile.Refused): + profile.configure_codex_login(login, {}) + self.assertEqual(self.files, {}) + + @unittest.skipUnless(os.environ.get('BUZZ_TEST_PACKAGED_CODEX') == '1', 'requires the packaged pinned Codex binary') + def test_the_pinned_codex_client_refreshes_and_persists_the_broker_credential(self): + with tempfile.TemporaryDirectory() as directory: + home = Path(directory) + (home / 'config.toml').write_text('cli_auth_credentials_store = "file"\n') + initial = {**self.bundle, 'refresh_token': self.cookie} + (home / 'auth.json').write_text(json.dumps({ + 'auth_mode': 'chatgpt', 'OPENAI_API_KEY': None, 'tokens': initial, + 'last_refresh': '2026-10-01T00:00:00Z', + })) + (home / 'ca.pem').write_text(self.ca) + env = {**os.environ, 'CODEX_HOME': directory, 'CODEX_CA_CERTIFICATE': str(home / 'ca.pem'), + 'CODEX_REFRESH_TOKEN_URL_OVERRIDE': self.login['url']} + for key in ('OPENAI_API_KEY', 'CODEX_API_KEY', 'CODEX_ACCESS_TOKEN'): + env.pop(key, None) + process = subprocess.Popen([shutil.which('codex'), 'app-server'], env=env, + stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, + text=True, start_new_session=True) + frames = Queue() + reader = Thread(target=lambda: [frames.put(line) for line in process.stdout], daemon=True) + reader.start() + try: + def request(identifier, method, params): + process.stdin.write(json.dumps({'id': identifier, 'method': method, 'params': params}) + '\n') + process.stdin.flush() + for _ in range(100): + try: + reply = json.loads(frames.get(timeout=10)) + except Empty: + self.fail(f'pinned Codex client did not answer; broker requests: {len(self.requests)}') + if reply.get('id') == identifier: + self.assertNotIn('error', reply, 'pinned Codex rejected the synthetic refresh workflow') + return reply + self.fail('too many unrelated Codex frames') + request(1, 'initialize', {'clientInfo': {'name': 'buzz-broker-test', 'version': '1.0'}}) + process.stdin.write('{"method":"initialized"}\n') + process.stdin.flush() + reply = request(2, 'getAuthStatus', {'refreshToken': True, 'includeToken': False}) + self.assertEqual(reply['result']['authMethod'], 'chatgpt') + request(3, 'getAuthStatus', {'refreshToken': True, 'includeToken': False}) + self.assertGreaterEqual(len(self.requests), 1) + self.assertEqual(self.requests[0]['refresh_token'], self.cookie) + self.assertEqual(self.requests[-1]['refresh_token'], self.bundle['refresh_token']) + saved = json.loads((home / 'auth.json').read_text()) + self.assertEqual(saved['tokens']['refresh_token'], self.bundle['refresh_token']) + finally: + os.killpg(process.pid, signal.SIGKILL) + process.wait(timeout=10) + process.stdin.close() + process.stdout.close() + reader.join(timeout=5) + + def test_profile_metadata_rejects_external_urls_and_credentials_for_another_developer(self): + login = {**self.login, 'url': 'https://codex-refresh.buzz-sandbox-manager.svc.cluster.local:8443/oauth/token'} + profile.validate_codex_login(login, 'alice') + for key, bad in [('url', 'https://attacker.invalid/oauth/token'), + ('refresh_token', 'bob.' + 'a' * 32 + '.seed')]: + with self.assertRaises(profile.Refused): + profile.validate_codex_login({**login, key: bad}, 'alice') + + +if __name__ == '__main__': + if len(sys.argv) == 2 and not sys.argv[1].startswith('-'): + directory = Path(__file__).resolve().parent + result = subprocess.run([ + 'docker', 'run', '--rm', '--network', 'none', '--cap-drop', 'ALL', + '--security-opt', 'no-new-privileges', '--env', 'BUZZ_TEST_PACKAGED_CODEX=1', + '--mount', f'type=bind,src={directory / "profile.py"},dst=/tmp/profile.py,readonly', + '--mount', f'type=bind,src={Path(__file__).resolve()},dst=/tmp/test-codex-login.py,readonly', + '--entrypoint', 'python3', sys.argv[1], '/tmp/test-codex-login.py', + ], timeout=120) + sys.exit(result.returncode) + unittest.main() diff --git a/voltage/images/buzz-sandbox-reaper/codex-login.md b/voltage/images/buzz-sandbox-reaper/codex-login.md new file mode 100644 index 00000000000..cc890345cd6 --- /dev/null +++ b/voltage/images/buzz-sandbox-reaper/codex-login.md @@ -0,0 +1,98 @@ +# Shared ChatGPT login for remote agents + +DEV-2948 adds a private refresh broker to the Sandbox manager. It prevents concurrent +Sandboxes and workers from rotating one shared OpenAI refresh token independently. +Enrollment remains an owner-signed, NIP-44 encrypted `enroll` request over the relay. +The HTTPS endpoint exists because the pinned native Codex refresh client requires it. + +```mermaid +sequenceDiagram + participant App as Buzz app + participant Manager as Singleton manager + participant SM as Secrets Manager + participant Journal as DynamoDB journal + participant Agent as Sandbox or worker + participant OAuth as OpenAI OAuth + App->>Manager: Encrypted dedicated login + Manager->>SM: Stage pending login, retain active login + Manager->>Journal: CAS exact secret version + Manager->>SM: Commit profile with broker credential and CA + Manager->>Journal: Activate staged login via new secret version + Agent->>Manager: HTTPS bootstrap or versioned refresh + Manager->>Journal: Consistent read of exact secret version + Manager->>SM: Read that immutable version + opt Token expired or current version needs refresh + Manager->>Journal: Commit Refreshing marker via new secret version + Manager->>OAuth: Refresh once + Manager->>Journal: Commit rotated bundle via new secret version + end + Manager-->>Agent: Short-lived tokens and versioned broker credential +``` + +## Ownership and failure handling + +| Store | Contents | Readers | +| --- | --- | --- | +| Developer profile secret | Existing accounts, broker URL, public CA, revocable broker credential | That developer's agents and manager | +| `voltage/staging/agents//codex-login` secret | Active and pending login, real tokens, refresh state | Manager only | +| `voltage-staging-agent-codex-logins` DynamoDB table | Developer ID and committed secret version ID | Manager only | + +Every journal read is strongly consistent. A conditional update publishes an exact immutable +secret version; the broker never follows `AWSCURRENT`. An unpublished or temporarily unreadable +version cannot cause the broker to fall back to an older refresh token. +A refresh marker is committed before contacting OpenAI. Cancellation, an uncertain OAuth result, +or a failed result write leaves the login interrupted. The next attempt requires re-enrollment. +This favors an explicit sign-in over risking reuse of a rotated token. + +Enrollment and refresh share a per-developer mutex. A failed profile write preserves the prior active +login. Activation follows the profile write; if activation fails, enrollment reports failure and +new sessions refuse the staged credential until enrollment succeeds. Retrying enrollment can activate +the persisted pending login. Disconnect revokes in the journal before changing the profile, so a stale +profile cannot restore access. Already-issued access tokens remain valid until upstream expiry or revocation. +The manager must still run as one replica with `Recreate`: relay replay and idempotency caches are local. + +Bodies, upstream replies, wait times, concurrency, certificate reload, and shutdown are bounded. +Error responses and logs never include credentials or provider bodies. Runtime credential files +are private and excluded from checkpoints. Broker grants authorize one developer only. + +## Rollout + +The chart defaults to `codexLogin.enabled: false`. This change alone enables no live broker. + +1. Review and apply the companion voltage-terraform staging-agent change: empty login secrets, + manager IAM, metadata journal, and agent egress to the manager on TCP 8443. Review the entire plan; + exclude unrelated existing staging-agent drift from this rollout. +2. Merge the Buzz implementation and let the Voltage images workflow publish both manager and agent + images for that commit. Use immutable image digests. +3. Pin the agent digest in both the manager's interactive provider configuration (CD values) and the + Terraform worker template. Apply/sync those changes before enrolling the new profile shape. + An old bootstrap rejects profiles with `codex_login`. +4. Pin the new manager digest, set `codexLogin.enabled: true`, and retain + `codexLogin.journalTable: voltage-staging-agent-codex-logins` in staging CD values. + Review the render, then manually sync `buzz-sandbox-manager`. + cert-manager issues a private CA and a leaf certificate for + `codex-refresh.buzz-sandbox-manager.svc.cluster.local`. +5. Check Certificate readiness, Deployment readiness, and the HTTPS health endpoint. + The manager mounts only the public CA and the leaf key, never the CA private key. + It reloads the leaf certificate every 60 seconds. CA replacement requires re-enrollment. +6. Build the Voltage GPUI app with `voltage-sandbox`. In the form, sign in with ChatGPT using + device auth, choose the Voltage workspace, and enroll. Test a fresh interactive session and worker. + +Disabling the broker causes enrolled shared-login sessions to fail closed. Before rolling back to +an old manager or agent image, disconnect shared logins and enroll a supported API key or workspace +access token. Do not copy the upstream refresh token into a profile or an agent auth cache. + +## Validation before marking DEV-2948 complete + +Automated tests use synthetic tokens and a local OAuth issuer; they do not prove a real OpenAI subscription. +Run the native form, cancel and replace a sign-in, confirm the dedicated cache is removed, and enroll. +Start five concurrent Sandboxes plus workers using one login, maintain activity beyond 24 hours, +and retain redacted evidence of successful refreshes without 401/re-login loops. Exercise manager restart, +re-enrollment, and disconnect. A fresh session using a revoked or stale profile must fail before its harness starts. +The issue remains open until this live acceptance test and human app testing succeed. + +The native client contract is pinned to Codex 0.159.3: JSON refresh grant, the fixed public client ID, +`CODEX_REFRESH_TOKEN_URL_OVERRIDE`, and file-backed persistence of the returned refresh credential. +OpenAI's [managed-auth guidance](https://developers.openai.com/codex/auth/ci-cd-auth) advises against +sharing an ordinary `auth.json` across concurrent runners. This is a custom broker, not a supported +managed-login sharing mode. Verify the pinned contract before any Codex upgrade. diff --git a/voltage/remote-agents.md b/voltage/remote-agents.md index 60214f87a8a..fdca77b97ab 100644 --- a/voltage/remote-agents.md +++ b/voltage/remote-agents.md @@ -108,10 +108,11 @@ sequenceDiagram | GitHub token | Enroll form | Profile (Secrets Manager) | The Sandbox: clone, push, PRs | | Agent signing key | Created by the app | Profile; public half on GitHub | The Sandbox: sign commits | | 1Password service account token | Enroll form | Profile | The Sandbox: read the Buzz Agents Dev vault | -| Codex, Claude, OpenAI, Anthropic, OpenRouter credentials | Enroll form | Profile | The harness in the Sandbox | +| Claude, OpenAI, OpenRouter credentials and Codex access tokens | Enroll form | Profile | The harness in the Sandbox | +| ChatGPT refresh token | Dedicated sign-in button | Manager-only Secrets Manager secret | Manager refresh broker | | Depot, Grafana, Linear, Argo CD tokens | Nobody: admins keep them in the vault | 1Password Buzz Agents Dev | The Sandbox, through `via` | -The app never saves your tokens. The profile is encrypted in transit and at rest. Checkpoints never include credentials. +The app removes the dedicated ChatGPT login cache after enrollment or cancellation. Other enrollment credentials are held in memory only. The profile is encrypted in transit and at rest. Checkpoints never include credentials. ## Onboard a developer @@ -219,6 +220,31 @@ To check the token, run this command. The output must show **Buzz Agents Dev** a OP_SERVICE_ACCOUNT_TOKEN='ops_…' op vault list ``` +#### Shared ChatGPT login for remote Codex agents + +Requires the DEV-2948 app, agent image, and manager rollout described in +[the deployment guide](images/buzz-sandbox-reaper/codex-login.md). +Install the Codex CLI locally and enable device-code sign-in in your ChatGPT workspace. + +1. In the Remote workspace enrollment form, select **Sign in with ChatGPT**. +2. Open the displayed OpenAI device page and enter the one-time code. Choose the Voltage workspace. +3. Return to Buzz and select **Enroll profile**. A completed browser sign-in is staged until you enroll. + +Buzz uses a dedicated temporary Codex cache. Your normal local Codex login remains separate. +The encrypted owner-signed enrollment sends the login to the manager. +The manager alone stores and rotates the upstream refresh token. Each Sandbox and worker receives +short-lived access tokens and a revocable broker credential through private HTTPS. +Starting a session checks that credential with the manager before launching the harness. + +To replace the login, sign in again and enroll. To revoke it, select **Disconnect ChatGPT login** +and enroll. New sessions and future refreshes are then refused, including sessions with a stale profile. +Already-issued access tokens remain usable until OpenAI expires or revokes them; stop existing sessions +when immediate termination is required. An OpenAI API key or Codex access token also replaces this login. + +If a manager restart interrupts a refresh whose outcome is uncertain, sign in and enroll again. +The manager refuses to retry a potentially consumed upstream refresh token. +This broker uses refresh behavior specific to the pinned Codex version; verify it before upgrading. + #### Codex access token (ChatGPT Business or Enterprise) 1. On chatgpt.com, switch to the **Voltage** workspace. @@ -227,7 +253,7 @@ OP_SERVICE_ACCOUNT_TOKEN='ops_…' op vault list 4. Copy the token (`at-…`). It shows only once. The token does not refresh, so all of your Sandboxes can use it at the same time. Before it expires, create a new one and enroll again. -A personal ChatGPT Plus or Pro account cannot create these tokens. Use an OpenAI API key instead. +A personal ChatGPT Plus or Pro account cannot create these workspace access tokens. Use the shared-login flow above or an OpenAI API key instead. #### Claude subscription token (Claude Pro, Max, Team, or Enterprise)