diff --git a/.github/workflows/monad-images.yml b/.github/workflows/monad-images.yml index 902ec75102..80c142de74 100644 --- a/.github/workflows/monad-images.yml +++ b/.github/workflows/monad-images.yml @@ -1,8 +1,8 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -# Builds Monad's OpenShell images (gateway, supervisor, sandbox) and pushes -# them to ghcr.io/monad-inc/openshell. See MONAD.md. +# Builds Monad's OpenShell images (gateway, supervisor, sandbox, CLI and +# provider sync) and pushes them to ghcr.io/monad-inc/openshell. See MONAD.md. name: Monad Images @@ -82,6 +82,8 @@ jobs: - { package: openshell-sandbox, binary: openshell-sandbox, triple: aarch64-unknown-linux-musl } - { package: openshell-cli, binary: openshell, triple: x86_64-unknown-linux-musl } - { package: openshell-cli, binary: openshell, triple: aarch64-unknown-linux-musl } + - { package: openshell-provider-sync, binary: openshell-provider-sync, triple: x86_64-unknown-linux-gnu } + - { package: openshell-provider-sync, binary: openshell-provider-sync, triple: aarch64-unknown-linux-gnu } # Repository variables MONAD_RUNNER_X64 / MONAD_RUNNER_ARM64 select larger runners # (e.g. ubuntu-24.04-8core / ubuntu-24.04-arm-8core); unset uses the standard ones. runs-on: ${{ startsWith(matrix.triple, 'aarch64') && (vars.MONAD_RUNNER_ARM64 || 'ubuntu-24.04-arm') || (vars.MONAD_RUNNER_X64 || 'ubuntu-24.04') }} @@ -117,6 +119,7 @@ jobs: - { component: supervisor, binary: openshell-supervisor, target-suffix: unknown-linux-gnu } - { component: sandbox, binary: openshell-sandbox, target-suffix: unknown-linux-musl } - { component: cli, binary: openshell, target-suffix: unknown-linux-musl } + - { component: provider-sync, binary: openshell-provider-sync, target-suffix: unknown-linux-gnu } runs-on: ubuntu-24.04 timeout-minutes: 30 permissions: diff --git a/AGENTS.md b/AGENTS.md index 8c70f2a3c7..c778fe173d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -41,6 +41,7 @@ Do not rely on this file for a full inventory. The detailed public and contribut | `crates/openshell-gateway/` | Gateway binary composition | Links selected first-party compute drivers into the backend-agnostic server registry | | `crates/openshell-sdk/` | Shared client SDK | Async Rust gateway client (gRPC transport, TLS, OIDC refresh, edge tunnel); consumed by CLI, TUI, and `@openshell/sdk` | | `crates/openshell-providers/` | Provider management | Credential provider backends | +| `crates/openshell-provider-sync/` | Provider sync controller | Keeps provider credentials current with labelled Kubernetes Secrets | | `crates/openshell-tui/` | Terminal UI | Ratatui-based dashboard for monitoring | | `crates/openshell-driver-kubernetes-secrets/` | Kubernetes Secrets credential driver | In-process `CredentialDriver` backend for OpenShell-managed K8s Secret storage | | `crates/openshell-driver-vault/` | Vault credential driver | In-process `CredentialDriver` backend for Vault-compatible KV storage | diff --git a/Cargo.lock b/Cargo.lock index 7c5c10331e..4633cdf62e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4886,6 +4886,27 @@ dependencies = [ "signal-hook", ] +[[package]] +name = "openshell-provider-sync" +version = "0.0.0" +dependencies = [ + "async-trait", + "clap", + "futures", + "k8s-openapi", + "kube", + "miette", + "openshell-core", + "openshell-sdk", + "rand 0.9.4", + "sha2 0.10.9", + "thiserror 2.0.20", + "tokio", + "tonic", + "tracing", + "tracing-subscriber", +] + [[package]] name = "openshell-providers" version = "0.0.0" diff --git a/MONAD.md b/MONAD.md index f360845d31..222019e6d7 100644 --- a/MONAD.md +++ b/MONAD.md @@ -17,7 +17,7 @@ This fork carries Monad's changes on top of upstream OpenShell releases: OCSF Au ## Images -`.github/workflows/monad-images.yml` builds the gateway, supervisor, sandbox runtime, and CLI for amd64 and arm64 and pushes them to `ghcr.io/monad-inc/openshell/{gateway,supervisor,sandbox,cli}`. The packages are private. +`.github/workflows/monad-images.yml` builds the gateway, supervisor, sandbox runtime, CLI, and provider sync controller for amd64 and arm64 and pushes them to `ghcr.io/monad-inc/openshell/{gateway,supervisor,sandbox,cli,provider-sync}`. The packages are private. The `cli` image only carries the static `openshell` binary. Copy it into images that need a client matching the gateway build: @@ -34,6 +34,8 @@ COPY --from=ghcr.io/monad-inc/openshell/cli: /usr/local/bin/openshell /usr/ The gateway binary is compiled to launch Monad's supervisor and sandbox runtime images at the same `sha-` tag, so the gateway and supervisor always match. +The `provider-sync` image runs `openshell-provider-sync`, which pushes credential values from labelled Kubernetes Secrets to gateway providers. It is a separate Deployment with the same tag as the gateway; see `crates/openshell-provider-sync/README.md`. + Binaries build on the standard `ubuntu-24.04` and `ubuntu-24.04-arm` runners. To use larger org runners, set the repository variables `MONAD_RUNNER_X64` and `MONAD_RUNNER_ARM64` (for example `ubuntu-24.04-8core` and `ubuntu-24.04-arm-8core`); the runner group must allow this repository. To cut a release, tag `monad/main` with `monad-v`, for example `git tag monad-v0.1.3-monad.1 && git push origin monad-v0.1.3-monad.1`. diff --git a/crates/openshell-provider-sync/Cargo.toml b/crates/openshell-provider-sync/Cargo.toml new file mode 100644 index 0000000000..6cb67f75d7 --- /dev/null +++ b/crates/openshell-provider-sync/Cargo.toml @@ -0,0 +1,38 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +[package] +name = "openshell-provider-sync" +description = "Keeps OpenShell provider credentials in sync with Kubernetes Secrets" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true + +[[bin]] +name = "openshell-provider-sync" +path = "src/main.rs" + +[dependencies] +openshell-core = { path = "../openshell-core", default-features = false } +openshell-sdk = { path = "../openshell-sdk" } +async-trait = "0.1" +clap = { workspace = true } +futures = { workspace = true } +k8s-openapi = { workspace = true } +kube = { workspace = true } +miette = { workspace = true } +rand = { workspace = true } +sha2 = { workspace = true } +thiserror = { workspace = true } +tokio = { workspace = true } +tonic = { workspace = true } +tracing = { workspace = true } +tracing-subscriber = { workspace = true } + +[dev-dependencies] +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } + +[lints] +workspace = true diff --git a/crates/openshell-provider-sync/README.md b/crates/openshell-provider-sync/README.md new file mode 100644 index 0000000000..68443d688d --- /dev/null +++ b/crates/openshell-provider-sync/README.md @@ -0,0 +1,124 @@ +# openshell-provider-sync + +Keeps provider credentials current with Kubernetes Secrets that something else +manages, such as External Secrets Operator syncing from Vault or 1Password. + +The controller watches Secrets labelled `openshell.nvidia.com/provider-sync=true`. +Each Secret names the providers it feeds and the credential keys it backs for +each one: + +```yaml +apiVersion: v1 +kind: Secret +metadata: + name: pm + labels: + openshell.nvidia.com/provider-sync: "true" + annotations: + openshell.nvidia.com/provider-workspace: pm # optional + openshell.nvidia.com/provider-credentials: >- + slack-pm:SLACK_BOT_TOKEN, slack-pm:SLACK_APP_TOKEN, + linear:LINEAR_API_KEY, + claude-code-apikey:ANTHROPIC_API_KEY, + claude-code-oauth:CLAUDE_CODE_OAUTH_TOKEN +``` + +An entry is `[provider:]CREDENTIAL_KEY[=secret-key]`, separated by commas or +newlines: + +- `slack-pm:SLACK_BOT_TOKEN` reads the Secret key `SLACK_BOT_TOKEN` into the + credential `SLACK_BOT_TOKEN` of provider `slack-pm`. +- `linear:LINEAR_API_KEY=linear-key` reads the Secret key `linear-key` instead. +- An entry without a `provider:` prefix goes to the provider named by the + `openshell.nvidia.com/provider` annotation, which is otherwise optional. + +All providers of one Secret are in one workspace, from +`openshell.nvidia.com/provider-workspace` or the controller's default. Only listed +keys are read, so a key added to the Secret later never becomes a credential by +accident, and one Secret key may be listed only once. + +## Behavior + +- **What it sends.** When a mapped value changes, the controller sends only the + changed keys of that provider as a value-only `UpdateProvider`. The gateway + merges them into the provider, and attached sandboxes refetch on their next + provider poll, every 10s by default. +- **Running processes.** A process that is already running gets the new value + only when the credential's profile sets `stable_placeholder: true`. Without it, + only processes started after the update see the new value. +- **What it leaves alone.** It never creates or deletes providers, never clears + a credential when its Secret is deleted or unlabelled, and never touches + credential keys that no Secret lists. Providers are still created by whatever + creates them today. +- **Each provider on its own.** A provider that fails does not hold back the + other providers of the same Secret. +- **Empty means absent.** A Secret key that is missing and one whose value is + empty are treated the same, so a Secret that writes every known key and leaves + unused ones empty works. An empty value is never sent. +- **All or nothing per provider.** If some but not all of a provider's keys are + set, or any is not UTF-8, nothing is sent to that provider, so a half-written + Secret cannot produce a partial rotation. +- **Absent providers.** A provider with none of its keys set in the Secret is + skipped. This lets a Secret list alternative credentials, such as an API key + and an OAuth token, and feed whichever one it holds. When it switches, the + newly present provider is sent in full. +- **One Secret per credential.** A second live Secret that maps the same + workspace, provider and credential key is refused with a + `ProviderSyncConflict` event. +- **Startup and restarts.** The watch starts with a full list, so every labelled + Secret is synced at startup. State is in memory: a salted fingerprint of each + value sent, never the value. After a restart every Secret is sent once more. + That is a value-only update and does not disturb running sandboxes. +- **Retries.** + - A missing provider is retried every 30s. + - Transient gateway failures are retried every 10s. + - Invalid annotations, conflicts, partly set keys and rejected updates + are retried every 5m, or as soon as the Secret changes. + - An absent provider is not retried; the next change to the Secret is seen. +- **Audit.** Each outcome is published as a Kubernetes Event on the Secret + (`ProviderSynced`, `ProviderSyncSkipped`, `ProviderNotFound`, + `ProviderSyncConflict`, `InvalidProviderSync`, `ProviderSyncFailed`). The + gateway separately audits each `UpdateProvider` with the controller's + principal. Neither carries values. + +## Running + +The Monad fork publishes the controller as `ghcr.io/monad-inc/openshell/provider-sync`, +with the same tags as the gateway. Run it as its own Deployment. + +```shell +openshell-provider-sync \ + --gateway http://openshell.openshell.svc:8080 \ + --namespace agents \ + --default-workspace default +``` + +| Flag | Environment | Purpose | +|---|---|---| +| `--gateway` | `OPENSHELL_GATEWAY_ENDPOINT` | Gateway URL | +| `--gateway-ca-file` | `OPENSHELL_GATEWAY_CA_FILE` | CA for a private-CA gateway | +| `--gateway-token-file` | `OPENSHELL_GATEWAY_TOKEN_FILE` | Bearer token, re-read when the gateway rejects it. Omit for a gateway that allows unauthenticated users | +| `--namespace` | `OPENSHELL_PROVIDER_SYNC_NAMESPACE` | Namespace to watch; all namespaces when omitted | +| `--default-workspace` | `OPENSHELL_PROVIDER_SYNC_DEFAULT_WORKSPACE` | Workspace when a Secret has no workspace annotation | +| `--reporter`, `--reporter-instance` | `OPENSHELL_PROVIDER_SYNC_REPORTER`, `POD_NAME` | Event reporter | +| `--no-events` | | Disable Kubernetes Events | +| `--log-json` | `OPENSHELL_LOG_JSON` | Log as JSON lines | + +On a gateway that requires authentication, the principal needs `provider:write` +with the workspace `admin` role in every workspace it updates. The token file +suits a projected ServiceAccount token or a Secret-mounted token that something +else rotates on disk. + +The ServiceAccount needs `get`, `list` and `watch` on Secrets in the watched +namespace, and `create` and `patch` on `events.k8s.io` Events. + +The gateway client is the Rust SDK, which speaks bearer-token auth only. A +gateway that requires mTLS client certificates is not supported yet. + +## Limits + +- **One replica.** There is no leader election. Two replicas would send the same + idempotent updates twice; that is wasteful but not wrong. +- **No activation wait.** The controller does not wait for supervisors to report + that they activated the update. The gateway's mutation id is logged for + correlation. diff --git a/crates/openshell-provider-sync/src/controller.rs b/crates/openshell-provider-sync/src/controller.rs new file mode 100644 index 0000000000..b0c2fb026e --- /dev/null +++ b/crates/openshell-provider-sync/src/controller.rs @@ -0,0 +1,692 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Watches labelled Secrets and pushes changed values to their providers. + +use std::sync::Arc; +use std::time::Duration; + +use futures::StreamExt; +use k8s_openapi::api::core::v1::Secret; +use kube::runtime::controller::{Action, Controller}; +use kube::runtime::events::{Event, EventType, Recorder, Reporter}; +use kube::runtime::reflector::{ObjectRef, Store}; +use kube::runtime::watcher; +use kube::{Api, Client, Resource, ResourceExt}; +use tokio::sync::Mutex; +use tracing::{debug, info, warn}; + +use crate::gateway::{ProviderUpdater, UpdateError}; +use crate::mapping::{self, Extracted, MappingError}; +use crate::state::{Conflict, SecretRef, SyncState}; + +/// How long to wait before retrying a provider that does not exist yet. +const PROVIDER_MISSING_RETRY: Duration = Duration::from_secs(30); +/// How long to wait before retrying a transient gateway failure. +const TRANSIENT_RETRY: Duration = Duration::from_secs(10); +/// How long to wait before retrying a Secret that can only succeed once it, +/// the provider or the gateway's configuration changes. +const INVALID_RETRY: Duration = Duration::from_mins(5); + +/// The result of syncing one provider. +#[derive(Debug, PartialEq, Eq)] +pub enum Outcome { + /// Values pushed; `keys` lists the credential keys that changed. + Updated { + keys: Vec, + mutation_id: String, + }, + /// Nothing changed since the last push. + Unchanged, + /// The Secret holds none of this provider's keys, so nothing was sent. + Absent { keys: Vec }, +} + +#[derive(Debug, thiserror::Error)] +pub enum SyncError { + #[error(transparent)] + Mapping(#[from] MappingError), + #[error(transparent)] + Conflict(#[from] Conflict), + #[error(transparent)] + Update(#[from] UpdateError), +} + +impl SyncError { + fn retry_after(&self) -> Duration { + match self { + Self::Update(UpdateError::ProviderNotFound { .. }) => PROVIDER_MISSING_RETRY, + Self::Update(UpdateError::Transient(_)) => TRANSIENT_RETRY, + Self::Mapping(_) | Self::Conflict(_) | Self::Update(UpdateError::Rejected(_)) => { + INVALID_RETRY + } + } + } + + fn reason(&self) -> &'static str { + match self { + Self::Mapping(_) => "InvalidProviderSync", + Self::Conflict(_) => "ProviderSyncConflict", + Self::Update(UpdateError::ProviderNotFound { .. }) => "ProviderNotFound", + Self::Update(_) => "ProviderSyncFailed", + } + } +} + +/// What happened to each provider a Secret feeds. +#[derive(Debug)] +pub struct SyncReport { + pub workspace: String, + /// In provider-name order. Each provider succeeds or fails on its own. + pub providers: Vec<(String, Result)>, +} + +impl SyncReport { + /// When to look at the Secret again: the soonest retry of any failed + /// provider, or `None` when every provider succeeded. + fn retry_after(&self) -> Option { + self.providers + .iter() + .filter_map(|(_, result)| result.as_ref().err().map(SyncError::retry_after)) + .min() + } +} + +/// Shared by every reconcile. +pub struct Context { + updater: Arc, + state: Mutex, + default_workspace: String, +} + +impl Context { + pub fn new(updater: Arc, default_workspace: impl Into) -> Self { + Self { + updater, + state: Mutex::new(SyncState::new()), + default_workspace: default_workspace.into(), + } + } + + /// Sync one Secret: parse its annotations, then for each provider read + /// the mapped keys and push the ones that changed. + /// + /// Fails only when the annotations are invalid. A provider that fails is + /// reported in the [`SyncReport`] and does not hold back the others. + /// + /// `owner_exists` reports whether another Secret holding a claim still + /// exists; see [`SyncState::claim`]. + pub async fn sync( + &self, + secret_ref: &SecretRef, + secret: &Secret, + owner_exists: impl Fn(&SecretRef) -> bool + Sync, + ) -> Result { + let target = match mapping::parse_target(secret.annotations(), &self.default_workspace) { + Ok(target) => target, + Err(err) => { + // An invalid Secret holds no claims, so fixing a sibling is + // not blocked by it. + self.state.lock().await.forget(secret_ref); + return Err(err); + } + }; + + // The lock is held across the gateway calls. Updates are rare and + // small, and holding it keeps two Secrets from racing on a claim. + let mut state = self.state.lock().await; + state.retain(secret_ref, &target); + let mut providers = Vec::with_capacity(target.providers.len()); + for (provider, keys) in &target.providers { + let result: Result = async { + state.claim(secret_ref, &target.workspace, provider, keys, &owner_exists)?; + let values = match mapping::extract_values(keys, secret.data.as_ref())? { + Extracted::Present(values) => values, + Extracted::Absent => { + state.clear_applied(secret_ref, provider); + return Ok(Outcome::Absent { + keys: keys.keys().cloned().collect(), + }); + } + }; + let changed = state.changed(secret_ref, &target.workspace, provider, &values); + if changed.is_empty() { + return Ok(Outcome::Unchanged); + } + let mutation_id = self + .updater + .update_credentials(&target.workspace, provider, &changed) + .await?; + state.commit(secret_ref, &target.workspace, provider, &values); + Ok(Outcome::Updated { + keys: changed.keys().cloned().collect(), + mutation_id, + }) + } + .await; + providers.push((provider.clone(), result)); + } + Ok(SyncReport { + workspace: target.workspace, + providers, + }) + } + + async fn forget(&self, secret_ref: &SecretRef) { + self.state.lock().await.forget(secret_ref); + } +} + +struct Runtime { + ctx: Context, + store: Store, + recorder: Option, +} + +fn secret_ref(secret: &Secret) -> SecretRef { + SecretRef { + namespace: secret.namespace().unwrap_or_default(), + name: secret.name_any(), + } +} + +fn is_live(store: &Store, secret: &SecretRef) -> bool { + store + .get(&ObjectRef::new(&secret.name).within(&secret.namespace)) + .is_some_and(|secret| secret.meta().deletion_timestamp.is_none()) +} + +async fn reconcile(secret: Arc, runtime: Arc) -> Result { + let secret_ref = secret_ref(&secret); + if secret.meta().deletion_timestamp.is_some() { + runtime.ctx.forget(&secret_ref).await; + return Ok(Action::await_change()); + } + let report = match runtime + .ctx + .sync(&secret_ref, &secret, |owner| is_live(&runtime.store, owner)) + .await + { + Ok(report) => report, + Err(err) => { + let err = SyncError::from(err); + warn!(secret = %secret_ref, error = %err, "provider sync failed"); + publish( + &runtime, + &secret, + EventType::Warning, + err.reason(), + err.to_string(), + ) + .await; + return Err(err); + } + }; + let workspace = &report.workspace; + for (provider, result) in &report.providers { + match result { + Ok(Outcome::Updated { keys, mutation_id }) => { + info!( + secret = %secret_ref, + workspace, + provider, + keys = ?keys, + mutation_id, + "provider credentials updated from Secret" + ); + let note = format!( + "Updated {} on provider {workspace}/{provider}", + keys.join(", ") + ); + publish(&runtime, &secret, EventType::Normal, "ProviderSynced", note).await; + } + Ok(Outcome::Unchanged) => { + debug!(secret = %secret_ref, workspace, provider, "provider credentials unchanged"); + } + Ok(Outcome::Absent { keys }) => { + debug!( + secret = %secret_ref, + workspace, + provider, + keys = ?keys, + "Secret holds none of the provider's keys" + ); + let note = format!( + "Skipped provider {workspace}/{provider}: Secret has no value for {}", + keys.join(", ") + ); + publish( + &runtime, + &secret, + EventType::Normal, + "ProviderSyncSkipped", + note, + ) + .await; + } + Err(err) => { + warn!( + secret = %secret_ref, + workspace, + provider, + error = %err, + "provider sync failed" + ); + let note = format!("Provider {workspace}/{provider}: {err}"); + publish(&runtime, &secret, EventType::Warning, err.reason(), note).await; + } + } + } + Ok(report + .retry_after() + .map_or_else(Action::await_change, Action::requeue)) +} + +fn error_policy(_secret: Arc, err: &SyncError, _runtime: Arc) -> Action { + Action::requeue(err.retry_after()) +} + +async fn publish(runtime: &Runtime, secret: &Secret, type_: EventType, reason: &str, note: String) { + let Some(recorder) = &runtime.recorder else { + return; + }; + let event = Event { + type_, + reason: reason.to_string(), + note: Some(note), + action: "UpdateProviderCredentials".to_string(), + secondary: None, + }; + if let Err(err) = recorder.publish(&event, &secret.object_ref(&())).await { + debug!(error = %err, "failed to publish Kubernetes event"); + } +} + +/// Options for [`run`]. +pub struct Options { + /// Watch one namespace, or every namespace when `None`. + pub namespace: Option, + /// Reporter for Kubernetes Events on the Secret; `None` disables them. + pub reporter: Option, +} + +/// Run until the watch stream ends or the process receives a shutdown signal. +/// +/// The watch starts with a full list, so every labelled Secret is synced once +/// at startup, not only when it next changes. +pub async fn run(client: Client, ctx: Context, options: Options) { + let api: Api = options.namespace.as_ref().map_or_else( + || Api::all(client.clone()), + |namespace| Api::namespaced(client.clone(), namespace), + ); + let controller = Controller::new( + api, + watcher::Config::default().labels(&mapping::label_selector()), + ); + let runtime = Arc::new(Runtime { + ctx, + store: controller.store(), + recorder: options + .reporter + .map(|reporter| Recorder::new(client, reporter)), + }); + controller + .shutdown_on_signal() + .run(reconcile, error_policy, runtime) + .for_each(|result| async move { + if let Err(err) = result { + debug!(error = %err, "reconcile did not complete"); + } + }) + .await; +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::mapping::{ + CREDENTIALS_ANNOTATION, CredentialValues, PROVIDER_ANNOTATION, WORKSPACE_ANNOTATION, + }; + use k8s_openapi::ByteString; + use kube::api::ObjectMeta; + use std::collections::BTreeMap; + use std::sync::Mutex as StdMutex; + + /// Workspace, provider and values of one update. + type UpdateCall = (String, String, BTreeMap); + + /// Fails updates for the named provider, or for every provider when the + /// name is `None`. + type FailRule = (Option<&'static str>, fn() -> UpdateError); + + #[derive(Default)] + struct FakeUpdater { + calls: StdMutex>, + fail_with: StdMutex>, + } + + #[async_trait::async_trait] + impl ProviderUpdater for FakeUpdater { + async fn update_credentials( + &self, + workspace: &str, + provider: &str, + values: &CredentialValues, + ) -> Result { + let fail_with = *self.fail_with.lock().unwrap(); + if let Some((only, fail)) = fail_with + && only.is_none_or(|only| only == provider) + { + return Err(fail()); + } + self.calls.lock().unwrap().push(( + workspace.to_string(), + provider.to_string(), + values.0.clone(), + )); + Ok("mutation-1".to_string()) + } + } + + fn secret( + name: &str, + annotations: &[(&str, &str)], + data: &[(&str, &str)], + ) -> (SecretRef, Secret) { + let secret = Secret { + metadata: ObjectMeta { + name: Some(name.to_string()), + namespace: Some("agents".to_string()), + annotations: Some( + annotations + .iter() + .map(|(k, v)| ((*k).to_string(), (*v).to_string())) + .collect(), + ), + ..Default::default() + }, + data: Some( + data.iter() + .map(|(k, v)| ((*k).to_string(), ByteString(v.as_bytes().to_vec()))) + .collect(), + ), + ..Default::default() + }; + (secret_ref(&secret), secret) + } + + fn github(token: &str) -> (SecretRef, Secret) { + secret( + "github-creds", + &[ + (PROVIDER_ANNOTATION, "github"), + (WORKSPACE_ANNOTATION, "oncall"), + (CREDENTIALS_ANNOTATION, "GITHUB_TOKEN=token"), + ], + &[("token", token), ("unrelated", "x")], + ) + } + + /// An agent Secret feeding three providers, with two alternative Claude + /// credentials of which only one is set. + fn agent(data: &[(&str, &str)]) -> (SecretRef, Secret) { + secret( + "pm", + &[ + (WORKSPACE_ANNOTATION, "pm"), + ( + CREDENTIALS_ANNOTATION, + "slack-pm:SLACK_BOT_TOKEN, slack-pm:SLACK_APP_TOKEN, linear:LINEAR_API_KEY,\ + claude-code-apikey:ANTHROPIC_API_KEY, claude-code-oauth:CLAUDE_CODE_OAUTH_TOKEN", + ), + ], + data, + ) + } + + fn outcome<'a>(report: &'a SyncReport, provider: &str) -> &'a Result { + &report + .providers + .iter() + .find(|(name, _)| name == provider) + .unwrap() + .1 + } + + #[tokio::test] + async fn pushes_on_first_sight_and_on_change_only() { + let updater = Arc::new(FakeUpdater::default()); + let ctx = Context::new(updater.clone(), "default"); + + let (r, s) = github("v1"); + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert!( + matches!(outcome(&report, "github"), Ok(Outcome::Updated { keys, .. }) if keys == &["GITHUB_TOKEN"]) + ); + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert!(matches!(outcome(&report, "github"), Ok(Outcome::Unchanged))); + assert_eq!(report.retry_after(), None); + + let (r, s) = github("v2"); + ctx.sync(&r, &s, |_| true).await.unwrap(); + + let calls = updater.calls.lock().unwrap(); + assert_eq!(calls.len(), 2); + assert_eq!(calls[1].0, "oncall"); + assert_eq!(calls[1].1, "github"); + assert_eq!( + calls[1].2, + BTreeMap::from([("GITHUB_TOKEN".to_string(), "v2".to_string())]) + ); + } + + #[tokio::test] + async fn one_secret_updates_each_provider_with_its_own_keys() { + let updater = Arc::new(FakeUpdater::default()); + let ctx = Context::new(updater.clone(), "default"); + let (r, s) = agent(&[ + ("SLACK_BOT_TOKEN", "bot"), + ("SLACK_APP_TOKEN", "app"), + ("LINEAR_API_KEY", "lin"), + ("ANTHROPIC_API_KEY", "ant"), + // Unused keys are written empty, which counts as absent. + ("CLAUDE_CODE_OAUTH_TOKEN", ""), + ]); + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert_eq!(report.workspace, "pm"); + assert!(matches!( + outcome(&report, "claude-code-oauth"), + Ok(Outcome::Absent { .. }) + )); + assert_eq!(report.retry_after(), None); + + let mut calls = updater.calls.lock().unwrap().clone(); + calls.sort(); + let providers: Vec<(&str, Vec<&str>)> = calls + .iter() + .map(|(_, provider, values)| { + ( + provider.as_str(), + values.keys().map(String::as_str).collect(), + ) + }) + .collect(); + assert_eq!( + providers, + [ + ("claude-code-apikey", vec!["ANTHROPIC_API_KEY"]), + ("linear", vec!["LINEAR_API_KEY"]), + ("slack-pm", vec!["SLACK_APP_TOKEN", "SLACK_BOT_TOKEN"]), + ] + ); + assert!(calls.iter().all(|(workspace, _, _)| workspace == "pm")); + } + + #[tokio::test] + async fn rotating_one_key_touches_only_its_provider() { + let updater = Arc::new(FakeUpdater::default()); + let ctx = Context::new(updater.clone(), "default"); + let base = [ + ("SLACK_BOT_TOKEN", "bot"), + ("SLACK_APP_TOKEN", "app"), + ("LINEAR_API_KEY", "lin"), + ("ANTHROPIC_API_KEY", "ant"), + ]; + let (r, s) = agent(&base); + ctx.sync(&r, &s, |_| true).await.unwrap(); + updater.calls.lock().unwrap().clear(); + + let mut rotated = base; + rotated[0] = ("SLACK_BOT_TOKEN", "bot-2"); + let (r, s) = agent(&rotated); + ctx.sync(&r, &s, |_| true).await.unwrap(); + let calls = updater.calls.lock().unwrap(); + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].1, "slack-pm"); + assert_eq!( + calls[0].2, + BTreeMap::from([("SLACK_BOT_TOKEN".to_string(), "bot-2".to_string())]) + ); + } + + #[tokio::test] + async fn a_failing_provider_does_not_hold_back_the_others() { + let updater = Arc::new(FakeUpdater::default()); + let ctx = Context::new(updater.clone(), "default"); + *updater.fail_with.lock().unwrap() = + Some((Some("linear"), || UpdateError::ProviderNotFound { + workspace: "pm".to_string(), + provider: "linear".to_string(), + })); + // Slack is half-written; Linear does not exist yet; Claude is fine. + let (r, s) = agent(&[ + ("SLACK_BOT_TOKEN", "bot"), + ("LINEAR_API_KEY", "lin"), + ("ANTHROPIC_API_KEY", "ant"), + ]); + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert!(matches!( + outcome(&report, "slack-pm"), + Err(SyncError::Mapping(MappingError::MissingSecretKey { .. })) + )); + assert!(matches!( + outcome(&report, "linear"), + Err(SyncError::Update(UpdateError::ProviderNotFound { .. })) + )); + assert!(matches!( + outcome(&report, "claude-code-apikey"), + Ok(Outcome::Updated { .. }) + )); + assert_eq!(report.retry_after(), Some(PROVIDER_MISSING_RETRY)); + + let calls = updater.calls.lock().unwrap(); + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].1, "claude-code-apikey"); + } + + #[tokio::test] + async fn switching_alternative_credentials_pushes_the_new_one() { + let updater = Arc::new(FakeUpdater::default()); + let ctx = Context::new(updater.clone(), "default"); + let (r, s) = agent(&[("ANTHROPIC_API_KEY", "ant")]); + ctx.sync(&r, &s, |_| true).await.unwrap(); + let (r, s) = agent(&[("CLAUDE_CODE_OAUTH_TOKEN", "oauth")]); + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert!(matches!( + outcome(&report, "claude-code-apikey"), + Ok(Outcome::Absent { .. }) + )); + assert!(matches!( + outcome(&report, "claude-code-oauth"), + Ok(Outcome::Updated { .. }) + )); + // Switching back resends the API key, even with the same value. + let (r, s) = agent(&[("ANTHROPIC_API_KEY", "ant")]); + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert!(matches!( + outcome(&report, "claude-code-apikey"), + Ok(Outcome::Updated { .. }) + )); + } + + #[tokio::test] + async fn failed_update_is_retried_with_the_same_values() { + let updater = Arc::new(FakeUpdater::default()); + let ctx = Context::new(updater.clone(), "default"); + let (r, s) = github("v1"); + + *updater.fail_with.lock().unwrap() = + Some((None, || UpdateError::Transient("unavailable".to_string()))); + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert_eq!(report.retry_after(), Some(TRANSIENT_RETRY)); + + *updater.fail_with.lock().unwrap() = None; + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert!(matches!( + outcome(&report, "github"), + Ok(Outcome::Updated { .. }) + )); + } + + #[tokio::test] + async fn missing_provider_retries_sooner_than_invalid_secret() { + let updater = Arc::new(FakeUpdater::default()); + let ctx = Context::new(updater.clone(), "default"); + *updater.fail_with.lock().unwrap() = Some((None, || UpdateError::ProviderNotFound { + workspace: "w".to_string(), + provider: "p".to_string(), + })); + let (r, s) = github("v1"); + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + let err = outcome(&report, "github").as_ref().unwrap_err(); + assert_eq!(err.retry_after(), PROVIDER_MISSING_RETRY); + assert_eq!(err.reason(), "ProviderNotFound"); + + let (r, s) = secret("bad", &[(PROVIDER_ANNOTATION, "p")], &[]); + let err = SyncError::from(ctx.sync(&r, &s, |_| true).await.unwrap_err()); + assert_eq!(err.retry_after(), INVALID_RETRY); + assert_eq!(err.reason(), "InvalidProviderSync"); + } + + #[tokio::test] + async fn half_written_secret_pushes_nothing() { + let updater = Arc::new(FakeUpdater::default()); + let ctx = Context::new(updater.clone(), "default"); + let (r, s) = secret( + "two-keys", + &[ + (PROVIDER_ANNOTATION, "p"), + (CREDENTIALS_ANNOTATION, "A=a,B=b"), + ], + &[("a", "1")], + ); + let report = ctx.sync(&r, &s, |_| true).await.unwrap(); + assert!(matches!( + outcome(&report, "p"), + Err(SyncError::Mapping(MappingError::MissingSecretKey { .. })) + )); + assert_eq!(report.retry_after(), Some(INVALID_RETRY)); + assert!(updater.calls.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn second_secret_for_the_same_credential_is_refused() { + let updater = Arc::new(FakeUpdater::default()); + let ctx = Context::new(updater.clone(), "default"); + let (r1, s1) = github("v1"); + ctx.sync(&r1, &s1, |_| true).await.unwrap(); + + let (r2, mut s2) = github("v2"); + s2.metadata.name = Some("github-creds-copy".to_string()); + let r2 = SecretRef { + name: "github-creds-copy".to_string(), + ..r2 + }; + let report = ctx.sync(&r2, &s2, |_| true).await.unwrap(); + assert!(matches!( + outcome(&report, "github"), + Err(SyncError::Conflict(_)) + )); + assert_eq!(updater.calls.lock().unwrap().len(), 1); + } +} diff --git a/crates/openshell-provider-sync/src/gateway.rs b/crates/openshell-provider-sync/src/gateway.rs new file mode 100644 index 0000000000..106bbaf243 --- /dev/null +++ b/crates/openshell-provider-sync/src/gateway.rs @@ -0,0 +1,201 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Writing credential values to the gateway. + +use std::collections::HashMap; +use std::path::PathBuf; + +use openshell_core::proto::datamodel::v1::{ObjectMeta, Provider}; +use openshell_core::proto::{UpdateProviderRequest, workspace_selector}; +use openshell_sdk::{OpenShellClient, Refresh, RefreshError, RefreshedToken}; +use tonic::Code; + +use crate::mapping::CredentialValues; + +#[derive(Debug, thiserror::Error)] +pub enum UpdateError { + /// The provider does not exist yet. Whatever creates providers (a deploy + /// job, a human) has not run; retry later. + #[error("provider {workspace}/{provider} does not exist")] + ProviderNotFound { workspace: String, provider: String }, + /// The gateway refused the values. Retrying the same values will fail + /// again; the Secret has to change. + #[error("gateway rejected the update: {0}")] + Rejected(String), + /// A transport, auth or server failure that may clear on retry. + #[error("gateway update failed: {0}")] + Transient(String), +} + +impl UpdateError { + fn from_status(status: &tonic::Status, workspace: &str, provider: &str) -> Self { + match status.code() { + Code::NotFound => Self::ProviderNotFound { + workspace: workspace.to_string(), + provider: provider.to_string(), + }, + Code::InvalidArgument | Code::FailedPrecondition | Code::PermissionDenied => { + Self::Rejected(format!("{:?}: {}", status.code(), status.message())) + } + _ => Self::Transient(format!("{:?}: {}", status.code(), status.message())), + } + } +} + +/// Writes credential values to a provider. +#[async_trait::async_trait] +pub trait ProviderUpdater: Send + Sync + 'static { + /// Merge `values` into the provider's credentials. Keys not in `values` + /// are left unchanged. Returns the gateway's mutation receipt id. + async fn update_credentials( + &self, + workspace: &str, + provider: &str, + values: &CredentialValues, + ) -> Result; +} + +/// [`ProviderUpdater`] over the gateway's `UpdateProvider` RPC. +pub struct GatewayUpdater { + client: OpenShellClient, +} + +impl GatewayUpdater { + pub fn new(client: OpenShellClient) -> Self { + Self { client } + } +} + +fn update_request( + workspace: &str, + provider: &str, + values: &CredentialValues, +) -> UpdateProviderRequest { + UpdateProviderRequest { + workspace_scope: Some(workspace_selector(workspace)), + provider: Some(Provider { + metadata: Some(ObjectMeta { + name: provider.to_string(), + workspace: workspace.to_string(), + // Zero: last write wins. The Secret is the source of truth + // for these keys, so there is no concurrent edit to protect. + resource_version: 0, + ..Default::default() + }), + credentials: values + .0 + .iter() + .map(|(key, value)| (key.clone(), value.clone())) + .collect::>(), + ..Default::default() + }), + ..Default::default() + } +} + +#[async_trait::async_trait] +impl ProviderUpdater for GatewayUpdater { + async fn update_credentials( + &self, + workspace: &str, + provider: &str, + values: &CredentialValues, + ) -> Result { + let mut retried_auth = false; + loop { + let mut grpc = self + .client + .raw_grpc_fresh() + .await + .map_err(|err| UpdateError::Transient(err.to_string()))?; + match grpc + .update_provider(update_request(workspace, provider, values)) + .await + { + Ok(response) => return Ok(response.into_inner().mutation_id), + Err(status) if status.code() == Code::Unauthenticated && !retried_auth => { + retried_auth = true; + let refreshed = self + .client + .force_refresh() + .await + .map_err(|err| UpdateError::Transient(err.to_string()))?; + if !refreshed { + return Err(UpdateError::from_status(&status, workspace, provider)); + } + } + Err(status) => return Err(UpdateError::from_status(&status, workspace, provider)), + } + } + } +} + +/// Bearer token read from a file, re-read whenever the client refreshes. +/// +/// Suits a projected `ServiceAccount` token or a Secret-mounted token that +/// something else rotates on disk. +pub struct TokenFile { + path: PathBuf, +} + +impl TokenFile { + pub fn new(path: PathBuf) -> Self { + Self { path } + } + + pub fn read(&self) -> Result { + let token = std::fs::read_to_string(&self.path).map_err(|err| { + RefreshError::Transient(format!("read {}: {err}", self.path.display())) + })?; + let token = token.trim(); + if token.is_empty() { + return Err(RefreshError::Transient(format!( + "{} is empty", + self.path.display() + ))); + } + Ok(token.to_string()) + } +} + +#[async_trait::async_trait] +impl Refresh for TokenFile { + async fn refresh(&self) -> Result { + self.read().map(RefreshedToken::new) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::BTreeMap; + + #[test] + fn request_carries_only_the_given_keys() { + let values = CredentialValues(BTreeMap::from([("A".to_string(), "1".to_string())])); + let request = update_request("oncall", "github", &values); + let provider = request.provider.unwrap(); + let metadata = provider.metadata.unwrap(); + assert_eq!(metadata.name, "github"); + assert_eq!(metadata.workspace, "oncall"); + assert_eq!(metadata.resource_version, 0); + assert_eq!(provider.r#type, ""); + assert_eq!(provider.profile_workspace, ""); + assert!(provider.credential_handles.is_empty()); + assert!(provider.config.is_empty()); + assert_eq!(provider.credentials.len(), 1); + assert_eq!(provider.credentials["A"], "1"); + assert_eq!(request.workspace_scope, Some(workspace_selector("oncall"))); + } + + #[test] + fn classifies_gateway_errors() { + let not_found = UpdateError::from_status(&tonic::Status::not_found("x"), "w", "p"); + assert!(matches!(not_found, UpdateError::ProviderNotFound { .. })); + let rejected = UpdateError::from_status(&tonic::Status::invalid_argument("x"), "w", "p"); + assert!(matches!(rejected, UpdateError::Rejected(_))); + let transient = UpdateError::from_status(&tonic::Status::unavailable("x"), "w", "p"); + assert!(matches!(transient, UpdateError::Transient(_))); + } +} diff --git a/crates/openshell-provider-sync/src/lib.rs b/crates/openshell-provider-sync/src/lib.rs new file mode 100644 index 0000000000..5eeeac09f0 --- /dev/null +++ b/crates/openshell-provider-sync/src/lib.rs @@ -0,0 +1,20 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Keeps `OpenShell` provider credentials in sync with Kubernetes Secrets. +//! +//! Secrets managed outside `OpenShell` (for example by External Secrets +//! Operator) opt in with a label and name the providers and credential keys +//! they back. When a Secret's mapped values change, the controller sends the +//! changed keys to the gateway as a value-only `UpdateProvider`. Running +//! sandboxes pick the new values up on their next provider poll, and +//! credentials with `stable_placeholder: true` reach already-running +//! processes too. +//! +//! The controller never creates or deletes providers, and never clears a +//! credential when its Secret goes away. + +pub mod controller; +pub mod gateway; +pub mod mapping; +pub mod state; diff --git a/crates/openshell-provider-sync/src/main.rs b/crates/openshell-provider-sync/src/main.rs new file mode 100644 index 0000000000..ddb8a0eea3 --- /dev/null +++ b/crates/openshell-provider-sync/src/main.rs @@ -0,0 +1,135 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! `openshell-provider-sync`: sync provider credentials from Kubernetes Secrets. + +use std::path::PathBuf; +use std::sync::Arc; + +use clap::Parser; +use kube::runtime::events::Reporter; +use miette::{IntoDiagnostic, Result, WrapErr}; +use openshell_provider_sync::controller::{self, Context, Options}; +use openshell_provider_sync::gateway::{GatewayUpdater, TokenFile}; +use openshell_sdk::{AuthConfig, ClientConfig, OpenShellClient}; +use tracing_subscriber::EnvFilter; + +#[derive(Debug, Parser)] +#[command( + name = "openshell-provider-sync", + about = "Sync OpenShell provider credentials from labelled Kubernetes Secrets", + version +)] +struct Cli { + /// Gateway URL, e.g. `http://openshell.openshell.svc:8080`. + #[arg(long, env = "OPENSHELL_GATEWAY_ENDPOINT")] + gateway: String, + + /// PEM CA certificate for a gateway with a private CA. + #[arg(long, env = "OPENSHELL_GATEWAY_CA_FILE")] + gateway_ca_file: Option, + + /// File holding a bearer token for the gateway. Re-read when the gateway + /// rejects the current token. Omit for a gateway that allows + /// unauthenticated users. + #[arg(long, env = "OPENSHELL_GATEWAY_TOKEN_FILE")] + gateway_token_file: Option, + + /// Namespace to watch. Watches every namespace when omitted. + #[arg(long, env = "OPENSHELL_PROVIDER_SYNC_NAMESPACE")] + namespace: Option, + + /// Workspace for Secrets without a provider-workspace annotation. + #[arg( + long, + env = "OPENSHELL_PROVIDER_SYNC_DEFAULT_WORKSPACE", + default_value = "default" + )] + default_workspace: String, + + /// Name reported on Kubernetes Events; usually the Deployment name. + #[arg( + long, + env = "OPENSHELL_PROVIDER_SYNC_REPORTER", + default_value = "openshell-provider-sync" + )] + reporter: String, + + /// Instance reported on Kubernetes Events; usually the pod name. + #[arg(long, env = "POD_NAME")] + reporter_instance: Option, + + /// Do not publish Kubernetes Events on synced Secrets. + #[arg(long)] + no_events: bool, + + /// Log as JSON lines. + #[arg(long, env = "OPENSHELL_LOG_JSON")] + log_json: bool, +} + +#[tokio::main] +async fn main() -> Result<()> { + let cli = Cli::parse(); + init_tracing(cli.log_json); + + let mut config = ClientConfig::new(cli.gateway.clone()); + if let Some(path) = &cli.gateway_ca_file { + config.ca_cert = Some( + std::fs::read(path) + .into_diagnostic() + .wrap_err_with(|| format!("read gateway CA {}", path.display()))?, + ); + } + if let Some(path) = cli.gateway_token_file { + let token_file = TokenFile::new(path); + let token = token_file.read().map_err(|err| miette::miette!("{err}"))?; + config.auth = Some(AuthConfig::Oidc { + token, + expires_at: None, + refresh: Some(Arc::new(token_file)), + }); + } + let gateway = OpenShellClient::connect(config) + .await + .map_err(|err| miette::miette!("connect to gateway {}: {err}", cli.gateway))?; + + let kube_client = kube::Client::try_default() + .await + .into_diagnostic() + .wrap_err("build Kubernetes client")?; + + let reporter = (!cli.no_events).then_some(Reporter { + controller: cli.reporter, + instance: cli.reporter_instance, + }); + tracing::info!( + gateway = %cli.gateway, + namespace = cli.namespace.as_deref().unwrap_or(""), + default_workspace = %cli.default_workspace, + "starting provider sync" + ); + controller::run( + kube_client, + Context::new( + Arc::new(GatewayUpdater::new(gateway)), + cli.default_workspace, + ), + Options { + namespace: cli.namespace, + reporter, + }, + ) + .await; + Ok(()) +} + +fn init_tracing(json: bool) { + let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); + let builder = tracing_subscriber::fmt().with_env_filter(filter); + if json { + builder.json().init(); + } else { + builder.init(); + } +} diff --git a/crates/openshell-provider-sync/src/mapping.rs b/crates/openshell-provider-sync/src/mapping.rs new file mode 100644 index 0000000000..44ab2681bb --- /dev/null +++ b/crates/openshell-provider-sync/src/mapping.rs @@ -0,0 +1,515 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! How a Kubernetes Secret declares the provider credentials it backs. +//! +//! A Secret opts in with the [`SYNC_LABEL`] label and names its targets with +//! annotations. The credential list is explicit: only the listed keys are +//! read, so an unrelated key added to the Secret never becomes a provider +//! credential. +//! +//! ```yaml +//! metadata: +//! labels: +//! openshell.nvidia.com/provider-sync: "true" +//! annotations: +//! openshell.nvidia.com/provider-workspace: pm # optional +//! openshell.nvidia.com/provider-credentials: >- +//! slack-pm:SLACK_BOT_TOKEN, slack-pm:SLACK_APP_TOKEN, +//! linear:LINEAR_API_KEY=linear-key +//! ``` +//! +//! Each credential entry is `[provider:]CREDENTIAL_KEY[=secret-key]`. The +//! Secret key defaults to the credential key. An entry without a provider +//! goes to the provider named by [`PROVIDER_ANNOTATION`], which is required +//! only when such an entry exists. One Secret can feed several providers, all +//! in one workspace. + +use std::collections::{BTreeMap, BTreeSet}; +use std::fmt; + +use k8s_openapi::ByteString; + +/// Label selecting the Secrets this controller watches. +pub const SYNC_LABEL: &str = "openshell.nvidia.com/provider-sync"; +/// Label value that enables sync. +pub const SYNC_LABEL_VALUE: &str = "true"; +/// Annotation naming the provider for entries without a `provider:` prefix. +pub const PROVIDER_ANNOTATION: &str = "openshell.nvidia.com/provider"; +/// Annotation naming the providers' workspace. Optional. +pub const WORKSPACE_ANNOTATION: &str = "openshell.nvidia.com/provider-workspace"; +/// Annotation listing the credential keys to sync. +pub const CREDENTIALS_ANNOTATION: &str = "openshell.nvidia.com/provider-credentials"; + +/// Label selector for [`SYNC_LABEL`]. +pub fn label_selector() -> String { + format!("{SYNC_LABEL}={SYNC_LABEL_VALUE}") +} + +/// Credential key -> Secret data key for one provider. Ordered so updates +/// are deterministic. +pub type KeyMap = BTreeMap; + +/// The providers a Secret feeds, and which of its keys feed which credential. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SyncTarget { + pub workspace: String, + /// Provider name -> its keys. Never empty, and every provider has a key. + pub providers: BTreeMap, +} + +/// Credential values read from a Secret, keyed by credential key. +/// +/// `Debug` prints only the keys. +#[derive(Clone, Default, PartialEq, Eq)] +pub struct CredentialValues(pub BTreeMap); + +impl fmt::Debug for CredentialValues { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_set().entries(self.0.keys()).finish() + } +} + +impl CredentialValues { + pub fn keys(&self) -> impl Iterator { + self.0.keys() + } + + pub fn is_empty(&self) -> bool { + self.0.is_empty() + } + + pub fn len(&self) -> usize { + self.0.len() + } +} + +#[derive(Debug, thiserror::Error, PartialEq, Eq)] +pub enum MappingError { + #[error("annotation {0} is missing or empty")] + MissingAnnotation(&'static str), + #[error("annotation {annotation} has an invalid value: {reason}")] + InvalidAnnotation { + annotation: &'static str, + reason: String, + }, + #[error( + "secret key '{secret_key}' for credential {credential_key} is missing or empty, but other keys of the same provider are set" + )] + MissingSecretKey { + credential_key: String, + secret_key: String, + }, + #[error("secret key '{secret_key}' for credential {credential_key} is not UTF-8")] + NotUtf8 { + credential_key: String, + secret_key: String, + }, +} + +/// Parse a Secret's annotations into a [`SyncTarget`]. +pub fn parse_target( + annotations: &BTreeMap, + default_workspace: &str, +) -> Result { + let default_provider = optional(annotations, PROVIDER_ANNOTATION); + if let Some(provider) = default_provider + && !is_provider_name(provider) + { + return Err(MappingError::InvalidAnnotation { + annotation: PROVIDER_ANNOTATION, + reason: format!("'{provider}' is not a valid provider name"), + }); + } + let workspace = optional(annotations, WORKSPACE_ANNOTATION) + .unwrap_or(default_workspace) + .to_string(); + let raw = optional(annotations, CREDENTIALS_ANNOTATION) + .ok_or(MappingError::MissingAnnotation(CREDENTIALS_ANNOTATION))?; + let providers = parse_credentials(raw, default_provider)?; + Ok(SyncTarget { + workspace, + providers, + }) +} + +fn optional<'a>(annotations: &'a BTreeMap, annotation: &str) -> Option<&'a str> { + annotations + .get(annotation) + .map(|value| value.trim()) + .filter(|value| !value.is_empty()) +} + +fn parse_credentials( + raw: &str, + default_provider: Option<&str>, +) -> Result, MappingError> { + let invalid = |reason: String| MappingError::InvalidAnnotation { + annotation: CREDENTIALS_ANNOTATION, + reason, + }; + let mut providers: BTreeMap = BTreeMap::new(); + let mut secret_keys = BTreeSet::new(); + for entry in raw + .split([',', '\n']) + .map(str::trim) + .filter(|entry| !entry.is_empty()) + { + let (provider, mapping) = match entry.split_once(':') { + Some((provider, mapping)) => (provider.trim(), mapping.trim()), + None => ( + default_provider.ok_or(MappingError::MissingAnnotation(PROVIDER_ANNOTATION))?, + entry, + ), + }; + if !is_provider_name(provider) { + return Err(invalid(format!( + "'{provider}' is not a valid provider name (letters, digits, '-', '_' and '.')" + ))); + } + let (credential_key, secret_key) = match mapping.split_once('=') { + Some((credential, secret)) => (credential.trim(), secret.trim()), + None => (mapping, mapping), + }; + if !is_credential_key(credential_key) { + return Err(invalid(format!( + "'{credential_key}' is not a valid credential key (letters, digits and '_', not starting with a digit)" + ))); + } + if !is_secret_data_key(secret_key) { + return Err(invalid(format!( + "'{secret_key}' is not a valid Secret key (letters, digits, '-', '_' and '.')" + ))); + } + if !secret_keys.insert(secret_key.to_string()) { + // One Secret key feeding two credentials is almost certainly a + // typo, and would let one value leak into a second binding. + return Err(invalid(format!( + "Secret key '{secret_key}' is mapped more than once" + ))); + } + if providers + .entry(provider.to_string()) + .or_default() + .insert(credential_key.to_string(), secret_key.to_string()) + .is_some() + { + return Err(invalid(format!( + "credential '{provider}:{credential_key}' is listed more than once" + ))); + } + } + if providers.is_empty() { + return Err(MappingError::MissingAnnotation(CREDENTIALS_ANNOTATION)); + } + Ok(providers) +} + +fn is_provider_name(name: &str) -> bool { + !name.is_empty() + && name.len() <= 253 + && name + .chars() + .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')) +} + +fn is_credential_key(key: &str) -> bool { + let mut chars = key.chars(); + chars + .next() + .is_some_and(|first| first.is_ascii_alphabetic() || first == '_') + && chars.all(|c| c.is_ascii_alphanumeric() || c == '_') +} + +fn is_secret_data_key(key: &str) -> bool { + !key.is_empty() + && key.len() <= 253 + && key + .chars() + .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')) +} + +/// What a Secret holds for one provider. +#[derive(Debug, PartialEq, Eq)] +pub enum Extracted { + /// Every mapped key is set. + Present(CredentialValues), + /// None of the mapped keys is set: each is missing or empty. The Secret + /// does not back this provider right now, as when only one of two + /// alternative credentials is set. + Absent, +} + +/// Read one provider's credential values from a Secret's data. +/// +/// A key counts as set when it is present and not empty, so a Secret that +/// writes every known key and leaves unused ones empty behaves like one that +/// omits them. Every mapped key must be set and UTF-8, or none may be set. +/// Anything in between is an error and nothing is returned, so a +/// half-written Secret never produces a partial update. +pub fn extract_values( + keys: &KeyMap, + data: Option<&BTreeMap>, +) -> Result { + let lookup = |secret_key: &String| { + data.and_then(|data| data.get(secret_key)) + .filter(|bytes| !bytes.0.is_empty()) + }; + if keys.values().all(|secret_key| lookup(secret_key).is_none()) { + return Ok(Extracted::Absent); + } + let mut values = BTreeMap::new(); + for (credential_key, secret_key) in keys { + let bytes = lookup(secret_key).ok_or_else(|| MappingError::MissingSecretKey { + credential_key: credential_key.clone(), + secret_key: secret_key.clone(), + })?; + let value = String::from_utf8(bytes.0.clone()).map_err(|_| MappingError::NotUtf8 { + credential_key: credential_key.clone(), + secret_key: secret_key.clone(), + })?; + values.insert(credential_key.clone(), value); + } + Ok(Extracted::Present(CredentialValues(values))) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn annotations(pairs: &[(&str, &str)]) -> BTreeMap { + pairs + .iter() + .map(|(k, v)| ((*k).to_string(), (*v).to_string())) + .collect() + } + + fn data(pairs: &[(&str, &[u8])]) -> BTreeMap { + pairs + .iter() + .map(|(k, v)| ((*k).to_string(), ByteString(v.to_vec()))) + .collect() + } + + fn keys(pairs: &[(&str, &str)]) -> KeyMap { + pairs + .iter() + .map(|(k, v)| ((*k).to_string(), (*v).to_string())) + .collect() + } + + #[test] + fn parses_plain_and_renamed_keys() { + let target = parse_target( + &annotations(&[ + (PROVIDER_ANNOTATION, " github-triage "), + ( + CREDENTIALS_ANNOTATION, + "GITHUB_TOKEN,\n LINEAR_API_KEY = linear-key ,", + ), + ]), + "default", + ) + .unwrap(); + assert_eq!(target.workspace, "default"); + assert_eq!( + target.providers, + BTreeMap::from([( + "github-triage".to_string(), + keys(&[ + ("GITHUB_TOKEN", "GITHUB_TOKEN"), + ("LINEAR_API_KEY", "linear-key") + ]), + )]) + ); + } + + #[test] + fn one_secret_feeds_several_providers() { + let target = parse_target( + &annotations(&[ + (WORKSPACE_ANNOTATION, "pm"), + ( + CREDENTIALS_ANNOTATION, + "slack-pm:SLACK_BOT_TOKEN, slack-pm:SLACK_APP_TOKEN,\n linear : LINEAR_API_KEY=linear-key", + ), + ]), + "default", + ) + .unwrap(); + assert_eq!(target.workspace, "pm"); + assert_eq!( + target.providers, + BTreeMap::from([ + ( + "linear".to_string(), + keys(&[("LINEAR_API_KEY", "linear-key")]) + ), + ( + "slack-pm".to_string(), + keys(&[ + ("SLACK_APP_TOKEN", "SLACK_APP_TOKEN"), + ("SLACK_BOT_TOKEN", "SLACK_BOT_TOKEN") + ]), + ), + ]) + ); + } + + #[test] + fn unprefixed_entries_use_the_provider_annotation() { + let target = parse_target( + &annotations(&[ + (PROVIDER_ANNOTATION, "github"), + ( + CREDENTIALS_ANNOTATION, + "GITHUB_TOKEN, linear:LINEAR_API_KEY", + ), + ]), + "default", + ) + .unwrap(); + assert_eq!( + target.providers.keys().collect::>(), + ["github", "linear"] + ); + } + + #[test] + fn workspace_annotation_overrides_default() { + let target = parse_target( + &annotations(&[ + (PROVIDER_ANNOTATION, "p"), + (WORKSPACE_ANNOTATION, "oncall"), + (CREDENTIALS_ANNOTATION, "TOKEN"), + ]), + "default", + ) + .unwrap(); + assert_eq!(target.workspace, "oncall"); + } + + #[test] + fn requires_credentials_and_a_provider_for_each_entry() { + assert_eq!( + parse_target(&annotations(&[(CREDENTIALS_ANNOTATION, "TOKEN")]), "d"), + Err(MappingError::MissingAnnotation(PROVIDER_ANNOTATION)) + ); + assert_eq!( + parse_target( + &annotations(&[(PROVIDER_ANNOTATION, "p"), (CREDENTIALS_ANNOTATION, " , ")]), + "d" + ), + Err(MappingError::MissingAnnotation(CREDENTIALS_ANNOTATION)) + ); + assert_eq!( + parse_target(&annotations(&[(PROVIDER_ANNOTATION, "p")]), "d"), + Err(MappingError::MissingAnnotation(CREDENTIALS_ANNOTATION)) + ); + } + + #[test] + fn rejects_invalid_and_duplicate_entries() { + for raw in [ + "1TOKEN", + "TO-KEN", + "TOKEN=bad/key", + "A=x,B=x", + "A=x,A=y", + "a:A=x,b:B=x", + "a:A=x,a:A=y", + ":TOKEN", + "bad/name:TOKEN", + ] { + let result = parse_target( + &annotations(&[(PROVIDER_ANNOTATION, "p"), (CREDENTIALS_ANNOTATION, raw)]), + "d", + ); + assert!( + matches!(result, Err(MappingError::InvalidAnnotation { .. })), + "{raw}: {result:?}" + ); + } + assert!(matches!( + parse_target( + &annotations(&[ + (PROVIDER_ANNOTATION, "bad name"), + (CREDENTIALS_ANNOTATION, "A") + ]), + "d" + ), + Err(MappingError::InvalidAnnotation { + annotation: PROVIDER_ANNOTATION, + .. + }) + )); + } + + #[test] + fn same_credential_key_on_two_providers_is_allowed() { + let target = parse_target( + &annotations(&[(CREDENTIALS_ANNOTATION, "a:TOKEN=one, b:TOKEN=two")]), + "d", + ) + .unwrap(); + assert_eq!(target.providers.len(), 2); + } + + #[test] + fn extracts_only_mapped_keys() { + let values = extract_values( + &keys(&[("A", "a")]), + Some(&data(&[("a", b"one"), ("unrelated", b"two")])), + ) + .unwrap(); + assert_eq!( + values, + Extracted::Present(CredentialValues(BTreeMap::from([( + "A".to_string(), + "one".to_string() + )]))) + ); + } + + #[test] + fn provider_with_no_keys_set_is_absent() { + let mapped = keys(&[("A", "a"), ("B", "b")]); + assert_eq!( + extract_values(&mapped, Some(&data(&[("a", b""), ("b", b"")]))), + Ok(Extracted::Absent) + ); + assert_eq!( + extract_values(&mapped, Some(&data(&[("other", b"x")]))), + Ok(Extracted::Absent) + ); + assert_eq!(extract_values(&mapped, None), Ok(Extracted::Absent)); + } + + #[test] + fn partly_set_or_binary_values_fail_the_provider() { + let mapped = keys(&[("A", "a"), ("B", "b")]); + assert!(matches!( + extract_values(&mapped, Some(&data(&[("a", b"one")]))), + Err(MappingError::MissingSecretKey { .. }) + )); + assert!(matches!( + extract_values(&mapped, Some(&data(&[("a", b"one"), ("b", b"")]))), + Err(MappingError::MissingSecretKey { .. }) + )); + assert!(matches!( + extract_values(&mapped, Some(&data(&[("a", b"one"), ("b", &[0xff, 0xfe])]))), + Err(MappingError::NotUtf8 { .. }) + )); + } + + #[test] + fn debug_never_prints_values() { + let values = CredentialValues(BTreeMap::from([( + "TOKEN".to_string(), + "super-secret".to_string(), + )])); + let rendered = format!("{values:?}"); + assert!(rendered.contains("TOKEN")); + assert!(!rendered.contains("super-secret")); + } +} diff --git a/crates/openshell-provider-sync/src/state.rs b/crates/openshell-provider-sync/src/state.rs new file mode 100644 index 0000000000..fece7bbf6a --- /dev/null +++ b/crates/openshell-provider-sync/src/state.rs @@ -0,0 +1,432 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! What has been pushed to the gateway, so only changed keys are sent. +//! +//! State is in memory only. Values are never stored: each applied value is +//! kept as a salted SHA-256 fingerprint, and the salt is random per process so +//! a fingerprint is useless outside it. After a restart every Secret is pushed +//! once more, which is a value-only update and harmless to running sandboxes. + +use std::collections::{BTreeMap, HashMap, HashSet}; +use std::fmt; + +use sha2::{Digest, Sha256}; + +use crate::mapping::{CredentialValues, KeyMap, SyncTarget}; + +/// `namespace/name` of a Secret. +#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)] +pub struct SecretRef { + pub namespace: String, + pub name: String, +} + +impl fmt::Display for SecretRef { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}/{}", self.namespace, self.name) + } +} + +/// One provider credential, which at most one Secret may feed. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +struct CredentialSlot { + workspace: String, + provider: String, + credential_key: String, +} + +impl fmt::Display for CredentialSlot { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "{}/{}:{}", + self.workspace, self.provider, self.credential_key + ) + } +} + +/// What a Secret last pushed to one provider. +#[derive(Clone)] +struct Applied { + workspace: String, + fingerprints: BTreeMap, +} + +/// A Secret and one provider it feeds. +type AppliedKey = (SecretRef, String); + +#[derive(Debug, thiserror::Error, PartialEq, Eq)] +#[error("credential {slot} is already synced from Secret {owner}")] +pub struct Conflict { + pub slot: String, + pub owner: SecretRef, +} + +pub struct SyncState { + salt: [u8; 32], + applied: HashMap, + owners: HashMap, +} + +impl Default for SyncState { + fn default() -> Self { + Self::new() + } +} + +fn slots(workspace: &str, provider: &str, keys: &KeyMap) -> Vec { + keys.keys() + .map(|credential_key| CredentialSlot { + workspace: workspace.to_string(), + provider: provider.to_string(), + credential_key: credential_key.clone(), + }) + .collect() +} + +impl SyncState { + pub fn new() -> Self { + Self { + salt: rand::random(), + applied: HashMap::new(), + owners: HashMap::new(), + } + } + + fn fingerprint(&self, value: &str) -> [u8; 32] { + let mut hasher = Sha256::new(); + hasher.update(self.salt); + hasher.update(value.as_bytes()); + hasher.finalize().into() + } + + /// Drop the claims and applied state `secret` holds outside `target`, as + /// when an entry is removed or the workspace changes. + pub fn retain(&mut self, secret: &SecretRef, target: &SyncTarget) { + let wanted: HashSet = target + .providers + .iter() + .flat_map(|(provider, keys)| slots(&target.workspace, provider, keys)) + .collect(); + self.release(secret, |slot| !wanted.contains(slot)); + self.applied.retain(|(owner, provider), applied| { + owner != secret + || (applied.workspace == target.workspace + && target.providers.contains_key(provider)) + }); + } + + /// Claim one provider's credentials for `secret`. + /// + /// `owner_exists` reports whether a Secret that holds a claim still + /// exists and is still labelled; a claim from a deleted Secret is taken + /// over rather than blocking its replacement. + pub fn claim( + &mut self, + secret: &SecretRef, + workspace: &str, + provider: &str, + keys: &KeyMap, + owner_exists: impl Fn(&SecretRef) -> bool, + ) -> Result<(), Conflict> { + let slots = slots(workspace, provider, keys); + for slot in &slots { + if let Some(owner) = self.owners.get(slot) + && owner != secret + && owner_exists(owner) + { + return Err(Conflict { + slot: slot.to_string(), + owner: owner.clone(), + }); + } + } + for slot in slots { + if let Some(previous) = self.owners.insert(slot, secret.clone()) + && &previous != secret + { + self.applied.remove(&(previous, provider.to_string())); + } + } + Ok(()) + } + + /// The values that differ from what `secret` last applied to `provider`. + pub fn changed( + &self, + secret: &SecretRef, + workspace: &str, + provider: &str, + values: &CredentialValues, + ) -> CredentialValues { + let applied = self + .applied + .get(&(secret.clone(), provider.to_string())) + .filter(|applied| applied.workspace == workspace); + CredentialValues( + values + .0 + .iter() + .filter(|(key, value)| { + applied + .and_then(|applied| applied.fingerprints.get(*key)) + .is_none_or(|fingerprint| *fingerprint != self.fingerprint(value)) + }) + .map(|(key, value)| (key.clone(), value.clone())) + .collect(), + ) + } + + /// Record that `values` are now what the gateway holds for `provider`. + pub fn commit( + &mut self, + secret: &SecretRef, + workspace: &str, + provider: &str, + values: &CredentialValues, + ) { + let fingerprints = values + .0 + .iter() + .map(|(key, value)| (key.clone(), self.fingerprint(value))) + .collect(); + self.applied.insert( + (secret.clone(), provider.to_string()), + Applied { + workspace: workspace.to_string(), + fingerprints, + }, + ); + } + + /// Forget what `secret` pushed to `provider`, so the next values it holds + /// are sent in full. The claim is kept. + pub fn clear_applied(&mut self, secret: &SecretRef, provider: &str) { + self.applied.remove(&(secret.clone(), provider.to_string())); + } + + /// Drop everything known about a Secret that was deleted, unlabelled or + /// made invalid. Provider credentials are left as they are. + pub fn forget(&mut self, secret: &SecretRef) { + self.applied.retain(|(owner, _), _| owner != secret); + self.release(secret, |_| true); + } + + fn release( + &mut self, + secret: &SecretRef, + mut should_release: impl FnMut(&CredentialSlot) -> bool, + ) { + self.owners + .retain(|slot, owner| owner != secret || !should_release(slot)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn secret(name: &str) -> SecretRef { + SecretRef { + namespace: "agents".to_string(), + name: name.to_string(), + } + } + + fn keys(names: &[&str]) -> KeyMap { + names + .iter() + .map(|k| ((*k).to_string(), (*k).to_lowercase())) + .collect() + } + + fn target(providers: &[(&str, &[&str])]) -> SyncTarget { + SyncTarget { + workspace: "default".to_string(), + providers: providers + .iter() + .map(|(provider, names)| ((*provider).to_string(), keys(names))) + .collect(), + } + } + + fn values(pairs: &[(&str, &str)]) -> CredentialValues { + CredentialValues( + pairs + .iter() + .map(|(k, v)| ((*k).to_string(), (*v).to_string())) + .collect(), + ) + } + + fn changed_keys(values: &CredentialValues) -> Vec<&str> { + values.keys().map(String::as_str).collect() + } + + #[test] + fn first_sight_pushes_everything_then_only_changes() { + let mut state = SyncState::new(); + let s = secret("gh"); + let v1 = values(&[("A", "1"), ("B", "2")]); + state + .claim(&s, "default", "github", &keys(&["A", "B"]), |_| true) + .unwrap(); + assert_eq!( + changed_keys(&state.changed(&s, "default", "github", &v1)), + ["A", "B"] + ); + state.commit(&s, "default", "github", &v1); + assert!(state.changed(&s, "default", "github", &v1).is_empty()); + + let v2 = values(&[("A", "1"), ("B", "3")]); + assert_eq!( + changed_keys(&state.changed(&s, "default", "github", &v2)), + ["B"] + ); + } + + #[test] + fn providers_of_one_secret_are_tracked_apart() { + let mut state = SyncState::new(); + let s = secret("pm"); + let token = values(&[("TOKEN", "1")]); + state.commit(&s, "default", "slack", &token); + assert!(state.changed(&s, "default", "slack", &token).is_empty()); + assert_eq!( + changed_keys(&state.changed(&s, "default", "linear", &token)), + ["TOKEN"] + ); + } + + #[test] + fn failed_push_is_retried() { + let state = SyncState::new(); + let v = values(&[("A", "1")]); + // No commit: the gateway call failed. + assert_eq!( + changed_keys(&state.changed(&secret("gh"), "default", "github", &v)), + ["A"] + ); + } + + #[test] + fn removed_provider_releases_its_claims_and_state() { + let mut state = SyncState::new(); + let s = secret("gh"); + let v = values(&[("A", "1")]); + state + .claim(&s, "default", "github", &keys(&["A"]), |_| true) + .unwrap(); + state.commit(&s, "default", "github", &v); + + state.retain(&s, &target(&[("github-2", &["A"])])); + // The old slot is free for another Secret. + assert!( + state + .claim(&secret("other"), "default", "github", &keys(&["A"]), |_| { + true + }) + .is_ok() + ); + // And returning to it pushes in full. + assert_eq!( + changed_keys(&state.changed(&s, "default", "github", &v)), + ["A"] + ); + } + + #[test] + fn workspace_change_pushes_in_full() { + let mut state = SyncState::new(); + let s = secret("gh"); + let v = values(&[("A", "1")]); + state.commit(&s, "default", "github", &v); + let mut moved = target(&[("github", &["A"])]); + moved.workspace = "oncall".to_string(); + state.retain(&s, &moved); + assert_eq!( + changed_keys(&state.changed(&s, "oncall", "github", &v)), + ["A"] + ); + } + + #[test] + fn two_live_secrets_cannot_feed_one_credential() { + let mut state = SyncState::new(); + state + .claim(&secret("first"), "default", "github", &keys(&["A"]), |_| { + true + }) + .unwrap(); + let err = state + .claim( + &secret("second"), + "default", + "github", + &keys(&["A"]), + |_| true, + ) + .unwrap_err(); + assert_eq!(err.owner, secret("first")); + // A different key of the same provider is fine. + assert!( + state + .claim( + &secret("second"), + "default", + "github", + &keys(&["B"]), + |_| { true } + ) + .is_ok() + ); + } + + #[test] + fn claim_of_a_deleted_secret_is_taken_over() { + let mut state = SyncState::new(); + let v = values(&[("A", "1")]); + state + .claim(&secret("first"), "default", "github", &keys(&["A"]), |_| { + true + }) + .unwrap(); + state.commit(&secret("first"), "default", "github", &v); + state + .claim( + &secret("second"), + "default", + "github", + &keys(&["A"]), + |owner| owner != &secret("first"), + ) + .unwrap(); + assert_eq!( + changed_keys(&state.changed(&secret("second"), "default", "github", &v)), + ["A"] + ); + } + + #[test] + fn forget_releases_claims() { + let mut state = SyncState::new(); + state + .claim(&secret("first"), "default", "github", &keys(&["A"]), |_| { + true + }) + .unwrap(); + state.forget(&secret("first")); + assert!( + state + .claim( + &secret("second"), + "default", + "github", + &keys(&["A"]), + |_| { true } + ) + .is_ok() + ); + } +} diff --git a/deploy/docker/Dockerfile.provider-sync b/deploy/docker/Dockerfile.provider-sync new file mode 100644 index 0000000000..c096a038df --- /dev/null +++ b/deploy/docker/Dockerfile.provider-sync @@ -0,0 +1,24 @@ +# syntax=docker/dockerfile:1.4 + +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +# Monad fork: the provider sync controller, which keeps provider credentials +# current with labelled Kubernetes Secrets. See +# crates/openshell-provider-sync/README.md. +# +# The GNU-linked binary is staged at: +# deploy/docker/.build/prebuilt-binaries//openshell-provider-sync +# +# Distroless Debian provides the glibc runtime required by the binary. + +ARG PROVIDER_SYNC_BASE_IMAGE=gcr.io/distroless/cc-debian13:nonroot@sha256:e792ab3d241a468a4fd7519ddbbebe66b49b5f365771716ea688ad40b6c6f1c2 +FROM ${PROVIDER_SYNC_BASE_IMAGE} AS provider-sync + +ARG TARGETARCH + +COPY --chmod=0555 deploy/docker/.build/prebuilt-binaries/${TARGETARCH}/openshell-provider-sync /usr/local/bin/openshell-provider-sync + +USER 65532:65532 + +ENTRYPOINT ["/usr/local/bin/openshell-provider-sync"] diff --git a/skills/debug-openshell-cluster/SKILL.md b/skills/debug-openshell-cluster/SKILL.md index ed2429bd3c..2d8289347e 100644 --- a/skills/debug-openshell-cluster/SKILL.md +++ b/skills/debug-openshell-cluster/SKILL.md @@ -943,6 +943,39 @@ done Any `no` result can strand recovery before workload Pod creation. Upgrade the OpenShell chart rather than treating missing workload Pods as Pod failures. +#### Provider sync controller + +Some deployments run `openshell-provider-sync` (image `provider-sync`) as its +own Deployment. It watches Secrets labelled `openshell.nvidia.com/provider-sync=true` +and pushes changed values to gateway providers with value-only `UpdateProvider` +calls. It never creates or deletes providers. When a rotated credential does not +reach a sandbox, read the Events on the Secret first. They name each provider +the Secret feeds and the outcome: + +```bash +kubectl -n get secret -o jsonpath='{.metadata.annotations}{"\n"}' +kubectl -n get events --field-selector involvedObject.name= --sort-by=.lastTimestamp | tail -n 20 +kubectl -n logs deployment/ --tail=200 +``` + +| Event reason | Meaning | +|---|---| +| `ProviderSynced` | Changed keys were sent; the log line carries the gateway `mutation_id` | +| `ProviderSyncSkipped` | Every key of that provider is missing or empty in the Secret, as with an unused alternative credential | +| `ProviderNotFound` | The provider does not exist in that workspace yet; retried every 30s | +| `ProviderSyncConflict` | Another live Secret already feeds the same workspace, provider and key | +| `InvalidProviderSync` | Bad annotations, or only some of a provider's keys are set, or a value is not UTF-8 | +| `ProviderSyncFailed` | The gateway rejected the update or was unreachable | + +No Events at all means the controller is not watching the Secret: check the +label, its `OPENSHELL_PROVIDER_SYNC_NAMESPACE`, and its RBAC (`get`, `list` and +`watch` on Secrets, `create` and `patch` on `events.k8s.io` Events). A +`ProviderSynced` event without the change reaching a running process usually +means the credential's profile lacks `stable_placeholder: true`; only processes +started after the update see the new value. Avoid rotating while gateways are +rolling: a supervisor whose provider refetch fails revokes its static +credentials. + #### Corporate upstream proxy When the deployment routes sandbox egress through a corporate HTTP forward @@ -1079,6 +1112,7 @@ credential failures. | Gateway fails before serving health after enabling an interceptor | Interceptor endpoint unavailable or manifest/binding validation failed | Gateway and interceptor logs; interceptor socket; `binding_policy`, phases, and failure policy | | Authenticated interceptor or middleware rejects gateway calls | Private CA or hostname mismatch, expected audience or issuer mismatch, stale/unknown `kid`, or malformed extension token | `tls_ca_cert_path`, registration `audience`, service verifier config and logs; fetch well-known metadata only through the already-trusted gateway TLS endpoint | | Provider profiles disappear after enabling an interceptor catalog | `provider_profile_sources` selected only an authoritative interceptor or returned invalid/duplicate IDs | Inspect source list and interceptor `Describe`/catalog logs; include `user` when composition with imported profiles is intended | +| A rotated Kubernetes Secret never reaches the provider | `openshell-provider-sync` is not watching it (label, namespace, RBAC), the provider is missing, or the annotations are invalid | Events on the Secret and the provider-sync logs; see [Provider sync controller](#provider-sync-controller) | | `provider list-profiles` is empty on a new gateway | Profiles are import-only and nothing has been imported | Import with `openshell provider profile import --from providers --global`; an empty catalog is a valid ready state, not a failure | | Sandbox create or provider attach fails naming a missing profile | The provider's profile was never imported, was deleted, or lives at another scope | Import it at the scope the provider uses; the error names the profile ID and the command | | Gateway fails after registering supervisor middleware | Service unavailable, invalid manifest, duplicate binding, reserved name, or invalid payload/timeout limit | Middleware service and gateway logs; `[[openshell.supervisor.middleware]]`; `Describe` response |