Merge remote-tracking branch 'origin/main' into ragnorc/explore-api

# Conflicts:
#	CONTRIBUTING.md

crates/omnigraph-server/src/api.rs

@@ -280,6 +280,11 @@ pub struct SchemaApplyOutput {
     pub steps: Vec<SchemaMigrationStep>,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct SchemaOutput {
+    pub schema_source: String,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 pub struct IngestRequest {
     pub branch: Option<String>,

crates/omnigraph-server/src/auth.rs

@@ -0,0 +1,310 @@
+//! Bearer token sources.
+//!
+//! A `TokenSource` loads `(actor_id, token)` pairs that the server uses to
+//! authenticate incoming bearer tokens. Plaintext tokens returned here are
+//! hashed immediately by `AppState` on ingest — see `hash_bearer_token` —
+//! and never persist past startup/refresh.
+//!
+//! The trait exists so that additional backends (AWS Secrets Manager,
+//! HashiCorp Vault, etc.) can plug in behind feature flags without
+//! touching the server wiring.
+
+use async_trait::async_trait;
+use color_eyre::eyre::{Result, bail};
+
+use crate::server_bearer_tokens_from_env;
+
+/// Environment variable that, when set, selects AWS Secrets Manager as the
+/// token source. Its value is the secret ID or ARN. Only honored when the
+/// binary is compiled with `--features aws`.
+pub const AWS_SECRET_ENV: &str = "OMNIGRAPH_SERVER_BEARER_TOKENS_AWS_SECRET";
+
+/// A source of bearer tokens, returned as `(actor_id, token)` pairs in
+/// plaintext. The caller is expected to hash tokens before storing them.
+#[async_trait]
+pub trait TokenSource: Send + Sync {
+    /// Fetch the current set of actor → token pairs.
+    ///
+    /// Called once at startup. Implementations that support rotation may
+    /// also be polled periodically.
+    async fn load(&self) -> Result<Vec<(String, String)>>;
+
+    /// Whether this source can be re-fetched for rotation without restart.
+    /// Default: false (one-shot sources).
+    fn supports_refresh(&self) -> bool {
+        false
+    }
+
+    /// Human-readable name for logs and error messages.
+    fn name(&self) -> &'static str;
+}
+
+/// Reads bearer tokens from environment variables and / or files, matching
+/// the long-standing server configuration:
+///
+/// - `OMNIGRAPH_SERVER_BEARER_TOKEN` — a single token assigned to the
+///   implicit actor `default`.
+/// - `OMNIGRAPH_SERVER_BEARER_TOKENS_JSON` — a JSON object of
+///   `{"actor_id": "token", …}`.
+/// - `OMNIGRAPH_SERVER_BEARER_TOKENS_FILE` — a path to a JSON file of the
+///   same shape.
+///
+/// Does not support refresh — reloading means restarting the process.
+#[derive(Debug, Default, Clone)]
+pub struct EnvOrFileTokenSource;
+
+#[async_trait]
+impl TokenSource for EnvOrFileTokenSource {
+    async fn load(&self) -> Result<Vec<(String, String)>> {
+        server_bearer_tokens_from_env()
+    }
+
+    fn name(&self) -> &'static str {
+        "env-or-file"
+    }
+}
+
+/// Pick the token source based on configuration.
+///
+/// Preference order:
+/// 1. If `OMNIGRAPH_SERVER_BEARER_TOKENS_AWS_SECRET` is set AND the binary was
+///    built with `--features aws`, returns an AWS Secrets Manager source.
+/// 2. If that env var is set but the binary was built without the feature,
+///    errors with a clear rebuild instruction rather than silently falling
+///    back to the env/file source (which would hide the misconfiguration).
+/// 3. Otherwise, returns `EnvOrFileTokenSource`.
+pub async fn resolve_token_source() -> Result<Box<dyn TokenSource>> {
+    if let Ok(secret_id) = std::env::var(AWS_SECRET_ENV) {
+        let secret_id = secret_id.trim().to_string();
+        if !secret_id.is_empty() {
+            #[cfg(feature = "aws")]
+            {
+                let source = aws::SecretsManagerTokenSource::new(secret_id).await?;
+                return Ok(Box::new(source));
+            }
+            #[cfg(not(feature = "aws"))]
+            {
+                bail!(
+                    "{} is set but this binary was not built with --features aws. \
+                     Rebuild: cargo build --release --features aws",
+                    AWS_SECRET_ENV
+                );
+            }
+        }
+    }
+    Ok(Box::new(EnvOrFileTokenSource))
+}
+
+/// Parse a JSON secret payload (from AWS Secrets Manager or any equivalent
+/// source) into actor → token pairs.
+///
+/// Payload shape: `{"actor_id_1": "token_1", "actor_id_2": "token_2", ...}`.
+/// Extracted as a free function so it can be unit-tested without the AWS SDK.
+#[cfg(any(test, feature = "aws"))]
+pub(crate) fn parse_json_secret_payload(payload: &str) -> Result<Vec<(String, String)>> {
+    use std::collections::HashMap;
+
+    let map: HashMap<String, String> = serde_json::from_str(payload).map_err(|err| {
+        color_eyre::eyre::eyre!(
+            "bearer-token secret payload is not a JSON object of actor→token: {}",
+            err
+        )
+    })?;
+
+    let mut pairs: Vec<(String, String)> = Vec::with_capacity(map.len());
+    for (actor, token) in map {
+        let actor = actor.trim().to_string();
+        let token = token.trim().to_string();
+        if actor.is_empty() {
+            bail!("bearer-token secret contains a blank actor id");
+        }
+        if token.is_empty() {
+            bail!("bearer-token secret has a blank token for actor '{}'", actor);
+        }
+        pairs.push((actor, token));
+    }
+    pairs.sort_by(|(a, _), (b, _)| a.cmp(b));
+    Ok(pairs)
+}
+
+#[cfg(feature = "aws")]
+pub mod aws {
+    //! AWS Secrets Manager bearer-token backend.
+    //!
+    //! Fetches a JSON payload from a named secret on startup. Credentials are
+    //! resolved via the AWS default chain — env vars, shared config, IMDSv2
+    //! instance role, or ECS task role — so no explicit credential plumbing
+    //! is needed when running under an IAM role.
+    //!
+    //! Background refresh for rotation is a follow-up.
+    use super::TokenSource;
+    use async_trait::async_trait;
+    use color_eyre::eyre::{Result, WrapErr, eyre};
+
+    /// Loads bearer tokens from a named AWS Secrets Manager secret.
+    pub struct SecretsManagerTokenSource {
+        client: aws_sdk_secretsmanager::Client,
+        secret_id: String,
+    }
+
+    impl SecretsManagerTokenSource {
+        /// Construct a new source. Resolves AWS credentials + region via the
+        /// default chain — no explicit configuration needed on EC2/ECS/EKS.
+        pub async fn new(secret_id: impl Into<String>) -> Result<Self> {
+            let config =
+                aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await;
+            let client = aws_sdk_secretsmanager::Client::new(&config);
+            Ok(Self {
+                client,
+                secret_id: secret_id.into(),
+            })
+        }
+    }
+
+    #[async_trait]
+    impl TokenSource for SecretsManagerTokenSource {
+        async fn load(&self) -> Result<Vec<(String, String)>> {
+            let output = self
+                .client
+                .get_secret_value()
+                .secret_id(&self.secret_id)
+                .send()
+                .await
+                .wrap_err_with(|| {
+                    format!("fetch AWS Secrets Manager secret '{}'", self.secret_id)
+                })?;
+
+            let payload = output.secret_string().ok_or_else(|| {
+                eyre!(
+                    "secret '{}' has no SecretString — binary secrets are not supported",
+                    self.secret_id
+                )
+            })?;
+
+            super::parse_json_secret_payload(payload)
+        }
+
+        fn supports_refresh(&self) -> bool {
+            true
+        }
+
+        fn name(&self) -> &'static str {
+            "aws-secrets-manager"
+        }
+    }
+}
+
+#[cfg(feature = "aws")]
+pub use aws::SecretsManagerTokenSource;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::env;
+    use serial_test::serial;
+
+    fn clear_env() {
+        unsafe {
+            env::remove_var("OMNIGRAPH_SERVER_BEARER_TOKEN");
+            env::remove_var("OMNIGRAPH_SERVER_BEARER_TOKENS_JSON");
+            env::remove_var("OMNIGRAPH_SERVER_BEARER_TOKENS_FILE");
+        }
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn env_or_file_source_returns_empty_when_nothing_configured() {
+        clear_env();
+        let source = EnvOrFileTokenSource;
+        let tokens = source.load().await.unwrap();
+        assert!(tokens.is_empty());
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn env_or_file_source_reads_single_token_as_default_actor() {
+        clear_env();
+        unsafe {
+            env::set_var("OMNIGRAPH_SERVER_BEARER_TOKEN", "some-token");
+        }
+        let source = EnvOrFileTokenSource;
+        let tokens = source.load().await.unwrap();
+        unsafe {
+            env::remove_var("OMNIGRAPH_SERVER_BEARER_TOKEN");
+        }
+        assert_eq!(tokens, vec![("default".to_string(), "some-token".to_string())]);
+    }
+
+    #[tokio::test]
+    async fn env_or_file_source_does_not_support_refresh() {
+        let source = EnvOrFileTokenSource;
+        assert!(!source.supports_refresh());
+        assert_eq!(source.name(), "env-or-file");
+    }
+
+    #[test]
+    fn parse_json_secret_payload_reads_actor_token_map() {
+        let pairs = parse_json_secret_payload(r#"{"alice": "tok-a", "bob": "tok-b"}"#).unwrap();
+        assert_eq!(
+            pairs,
+            vec![
+                ("alice".to_string(), "tok-a".to_string()),
+                ("bob".to_string(), "tok-b".to_string()),
+            ]
+        );
+    }
+
+    #[test]
+    fn parse_json_secret_payload_trims_whitespace() {
+        let pairs = parse_json_secret_payload(r#"{"  alice  ": "  tok-a  "}"#).unwrap();
+        assert_eq!(pairs, vec![("alice".to_string(), "tok-a".to_string())]);
+    }
+
+    #[test]
+    fn parse_json_secret_payload_rejects_blank_actor() {
+        let err = parse_json_secret_payload(r#"{"   ": "tok"}"#).unwrap_err();
+        assert!(err.to_string().contains("blank actor"));
+    }
+
+    #[test]
+    fn parse_json_secret_payload_rejects_blank_token() {
+        let err = parse_json_secret_payload(r#"{"alice": "  "}"#).unwrap_err();
+        assert!(err.to_string().contains("blank token"));
+    }
+
+    #[test]
+    fn parse_json_secret_payload_rejects_non_object() {
+        let err = parse_json_secret_payload("[1, 2, 3]").unwrap_err();
+        assert!(err.to_string().contains("not a JSON object"));
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn resolve_token_source_falls_back_to_env_or_file_when_aws_var_unset() {
+        clear_env();
+        unsafe {
+            env::remove_var(AWS_SECRET_ENV);
+        }
+        let source = resolve_token_source().await.unwrap();
+        assert_eq!(source.name(), "env-or-file");
+    }
+
+    #[cfg(not(feature = "aws"))]
+    #[tokio::test]
+    #[serial]
+    async fn resolve_token_source_errors_when_aws_var_set_without_feature() {
+        clear_env();
+        unsafe {
+            env::set_var(AWS_SECRET_ENV, "some-secret-id");
+        }
+        let result = resolve_token_source().await;
+        unsafe {
+            env::remove_var(AWS_SECRET_ENV);
+        }
+        let err = match result {
+            Ok(_) => panic!("expected resolve_token_source to error without aws feature"),
+            Err(err) => err,
+        };
+        assert!(err.to_string().contains("--features aws"));
+    }
+}

crates/omnigraph-server/src/lib.rs

@@ -1,4 +1,5 @@
 pub mod api;
+pub mod auth;
 pub mod config;
 pub mod policy;
 
@@ -14,7 +15,7 @@ use api::{
     BranchMergeOutput, BranchMergeRequest, ChangeOutput, ChangeRequest, CommitListOutput,
     CommitListQuery, ErrorCode, ErrorOutput, ExportRequest, HealthOutput, IngestOutput,
     IngestRequest, ReadOutput, ReadRequest, RunListOutput, SchemaApplyOutput, SchemaApplyRequest,
-    SnapshotQuery, ingest_output, schema_apply_output, snapshot_payload,
+    SchemaOutput, SnapshotQuery, ingest_output, schema_apply_output, snapshot_payload,
 };
 use axum::body::{Body, Bytes};
 use axum::extract::DefaultBodyLimit;
@@ -37,11 +38,14 @@ use omnigraph::error::{ManifestErrorKind, OmniError};
 use omnigraph_compiler::json_params_to_param_map;
 use omnigraph_compiler::query::parser::parse_query;
 use omnigraph_compiler::{JsonParamMode, ParamMap};
+pub use auth::{AWS_SECRET_ENV, EnvOrFileTokenSource, TokenSource, resolve_token_source};
 pub use policy::{
     PolicyAction, PolicyCompiler, PolicyConfig, PolicyDecision, PolicyEngine, PolicyExpectation,
     PolicyRequest, PolicyTestConfig,
 };
 use serde_json::Value;
+use sha2::{Digest, Sha256};
+use subtle::ConstantTimeEq;
 use tokio::net::TcpListener;
 use tokio::sync::{RwLock, mpsc};
 use tower_http::trace::TraceLayer;
@@ -50,6 +54,15 @@ use tracing_subscriber::EnvFilter;
 use utoipa::OpenApi;
 use utoipa::openapi::security::{Http, HttpAuthScheme, SecurityScheme};
 
+type BearerTokenHash = [u8; 32];
+
+fn hash_bearer_token(token: &str) -> BearerTokenHash {
+    let digest = Sha256::digest(token.as_bytes());
+    let mut out = [0u8; 32];
+    out.copy_from_slice(&digest);
+    out
+}
+
 #[derive(OpenApi)]
 #[openapi(
     info(
@@ -63,6 +76,7 @@ use utoipa::openapi::security::{Http, HttpAuthScheme, SecurityScheme};
         server_export,
         server_change,
         server_schema_apply,
+        server_schema_get,
         server_ingest,
         server_branch_list,
         server_branch_create,
@@ -109,7 +123,7 @@ pub struct ServerConfig {
 pub struct AppState {
     uri: String,
     db: Arc<RwLock<Omnigraph>>,
-    bearer_tokens: Arc<HashMap<Arc<str>, Arc<str>>>,
+    bearer_tokens: Arc<[(BearerTokenHash, Arc<str>)]>,
     policy_engine: Option<Arc<PolicyEngine>>,
 }
 
@@ -174,14 +188,14 @@ impl AppState {
         bearer_tokens: Vec<(String, String)>,
         policy_engine: Option<PolicyEngine>,
     ) -> Self {
-        let bearer_tokens = bearer_tokens
+        let bearer_tokens: Vec<(BearerTokenHash, Arc<str>)> = bearer_tokens
             .into_iter()
-            .map(|(actor, token)| (Arc::<str>::from(token), Arc::<str>::from(actor)))
+            .map(|(actor, token)| (hash_bearer_token(&token), Arc::<str>::from(actor)))
             .collect();
         Self {
             uri,
             db: Arc::new(RwLock::new(db)),
-            bearer_tokens: Arc::new(bearer_tokens),
+            bearer_tokens: Arc::from(bearer_tokens),
             policy_engine: policy_engine.map(Arc::new),
         }
     }
@@ -241,7 +255,17 @@ impl AppState {
     }
 
     fn authenticate_bearer_token(&self, provided_token: &str) -> Option<Arc<str>> {
-        self.bearer_tokens.get(provided_token).cloned()
+        // Hash the incoming token and compare against every stored digest in
+        // constant time. Iterate all entries unconditionally so total work —
+        // and therefore response timing — doesn't depend on which slot matches.
+        let provided_hash = hash_bearer_token(provided_token);
+        let mut matched: Option<Arc<str>> = None;
+        for (hash, actor) in self.bearer_tokens.iter() {
+            if bool::from(hash.ct_eq(&provided_hash)) && matched.is_none() {
+                matched = Some(Arc::clone(actor));
+            }
+        }
+        matched
     }
 
     fn policy_engine(&self) -> Option<&PolicyEngine> {
@@ -407,6 +431,7 @@ pub fn build_app(state: AppState) -> Router {
         .route("/export", post(server_export))
         .route("/read", post(server_read))
         .route("/change", post(server_change))
+        .route("/schema", get(server_schema_get))
         .route("/schema/apply", post(server_schema_apply))
         .route(
             "/ingest",
@@ -439,9 +464,11 @@ pub fn build_app(state: AppState) -> Router {
 }
 
 pub async fn serve(config: ServerConfig) -> Result<()> {
+    let token_source = resolve_token_source().await?;
+    info!(source = token_source.name(), "loaded bearer token source");
     let state = AppState::open_with_bearer_tokens_and_policy(
         config.uri.clone(),
-        server_bearer_tokens_from_env()?,
+        token_source.load().await?,
         config.policy_file.as_ref(),
     )
     .await?;
@@ -553,7 +580,7 @@ fn log_policy_decision(actor_id: &str, request: &PolicyRequest, decision: &Polic
 fn authorize_request(
     state: &AppState,
     actor: Option<&AuthenticatedActor>,
-    request: PolicyRequest,
+    mut request: PolicyRequest,
 ) -> std::result::Result<(), ApiError> {
     let Some(engine) = state.policy_engine() else {
         return Ok(());
@@ -561,6 +588,10 @@ fn authorize_request(
     let Some(actor) = actor else {
         return Err(ApiError::unauthorized("missing bearer token"));
     };
+    // Authoritative actor_id is the authenticated session, not whatever the
+    // handler put in the request. Prevents an empty-string default at any
+    // call site from ever reaching the engine as a policy subject.
+    request.actor_id = actor.as_str().to_string();
     let decision = engine
         .authorize(&request)
         .map_err(|err| ApiError::internal(format!("policy: {err}")))?;
@@ -801,6 +832,42 @@ async fn server_change(
     }))
 }
 
+#[utoipa::path(
+    get,
+    path = "/schema",
+    tag = "schema",
+    operation_id = "getSchema",
+    responses(
+        (status = 200, description = "Current schema source", body = SchemaOutput),
+        (status = 401, description = "Unauthorized", body = ErrorOutput),
+        (status = 403, description = "Forbidden", body = ErrorOutput),
+    ),
+    security(("bearer_token" = [])),
+)]
+async fn server_schema_get(
+    State(state): State<AppState>,
+    actor: Option<Extension<AuthenticatedActor>>,
+) -> std::result::Result<Json<SchemaOutput>, ApiError> {
+    authorize_request(
+        &state,
+        actor.as_ref().map(|Extension(actor)| actor),
+        PolicyRequest {
+            actor_id: actor
+                .as_ref()
+                .map(|Extension(actor)| actor.as_str().to_string())
+                .unwrap_or_default(),
+            action: PolicyAction::Read,
+            branch: None,
+            target_branch: None,
+        },
+    )?;
+    let schema_source = {
+        let db = Arc::clone(&state.db).read_owned().await;
+        db.schema_source().to_string()
+    };
+    Ok(Json(SchemaOutput { schema_source }))
+}
+
 #[utoipa::path(
     post,
     path = "/schema/apply",
@@ -1461,13 +1528,43 @@ fn server_bearer_tokens_from_env() -> Result<Vec<(String, String)>> {
 #[cfg(test)]
 mod tests {
     use super::{
-        load_server_settings, normalize_bearer_token, parse_bearer_tokens_json,
+        hash_bearer_token, load_server_settings, normalize_bearer_token, parse_bearer_tokens_json,
         server_bearer_tokens_from_env,
     };
     use std::env;
     use std::fs;
     use tempfile::tempdir;
 
+    #[test]
+    fn hash_bearer_token_produces_32_byte_output() {
+        let hash = hash_bearer_token("any-token");
+        assert_eq!(hash.len(), 32);
+    }
+
+    #[test]
+    fn hash_bearer_token_is_deterministic() {
+        assert_eq!(
+            hash_bearer_token("stable-input"),
+            hash_bearer_token("stable-input"),
+        );
+    }
+
+    #[test]
+    fn hash_bearer_token_differs_for_different_inputs() {
+        assert_ne!(hash_bearer_token("token-a"), hash_bearer_token("token-b"));
+    }
+
+    #[test]
+    fn hash_bearer_token_matches_known_sha256_vector() {
+        // SHA-256("abc"). If this ever fails, the hash function was swapped.
+        let hash = hash_bearer_token("abc");
+        let hex: String = hash.iter().map(|b| format!("{:02x}", b)).collect();
+        assert_eq!(
+            hex,
+            "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"
+        );
+    }
+
     #[test]
     fn server_settings_load_from_yaml_config() {
         let temp = tempdir().unwrap();