Close search's active-tenant write-routing gap with a polled allowlist

search/src/consumer.rs's write-routing (built last pass) had no active-
tenant check at all: IndexRegistry.resolve() would open-or-create an
index directory for any syntactically-valid tenant_id, active or not --
unlike ClickHouse's chwriter.Registry (an active-tenants-only snapshot
built at enterprise-ingest startup) or the read side (gated by
searchclient.TenantChecker, a direct rbacstore query). search is AGPL
core with no Postgres access and no enterprise/ import allowed, so it
needed a network boundary instead -- the same shape ingest's
TenantResolver already uses against enterprise-auth, just Rust calling
Go instead of Go calling Go.

New GET /internal/active-tenants endpoint on enterprise-auth
(rbacstore.ListActiveTenantIDs + authhandler.handleActiveTenants),
gated on a RoleService Bearer credential -- server-to-server auth, the
same shape alerting presents to api, minted via the already-generic
enterprise-auth -mint-service-token search. search/src/tenants.rs's
ActiveTenantTracker polls it every 60s, blocking startup on the first
fetch succeeding (fail-closed cold start -- a control-plane outage at
boot must not silently accept every tenant_id) and keeping the last-
known-good set on any later refresh failure (a transient blip shouldn't
stop every tenant's indexing, only prevent the allowlist from growing/
shrinking until connectivity resumes). consumer.rs refuses any tagged
record whose tenant isn't in the polled set, before ever calling
resolve() -- IndexRegistry itself stays policy-free, matching the same
mechanism/policy split clickhousewriter.Writer vs. chwriter.Registry
already draws on the ClickHouse side.

Off unless ENTERPRISE_AUTH_URL/ENTERPRISE_AUTH_SERVICE_TOKEN are both
set (search/src/config.rs rejects exactly one being set) -- every
existing deployment is unaffected.

Verified with real HTTP round trips in this environment: tenants.rs's
tests exercise real reqwest requests (actual Authorization: Bearer
header, actual JSON parsing) against a hand-rolled dependency-free TCP
test server, including both fail-closed paths (rejected first fetch,
unreachable server). authhandler's new tests cover the credential-kind
distinction this endpoint exists to enforce -- a real human session,
even for a genuine Owner, must not satisfy a check meant for a service
identity.

One asymmetry remains, disclosed rather than fixed: chwriter.Registry's
snapshot still never refreshes (stale until enterprise-ingest restarts),
while ActiveTenantTracker's 60s poll gives Tantivy a materially tighter
staleness window. Neither is a live per-write check -- that would mean
a database/HTTP round trip per record, a throughput cost neither
implementation accepts -- so both have some staleness window by design;
the gap between the two windows is what's disclosed, not a claim either
is fully live.

search/src/config.rs

@@ -13,14 +13,35 @@ pub struct Config {
     pub commit_interval: Duration,
     /// Phase 4: per-tenant index directories live under here, one
     /// subdirectory per tenant_id, opened on demand by
-    /// registry::IndexRegistry -- distinct from `index_path` above,
-    /// which stays the single shared index every ingest-written record
-    /// lands in regardless of tenant (see registry.rs's doc comment and
-    /// /docs/security/threat-model.md's ingest-tenancy caveat). Default
+    /// registry::IndexRegistry for both the read side (SearchRequest.
+    /// tenant_id) and the write side (consumer.rs, routing on each
+    /// record's tenant_id Kafka header) -- `index_path` above stays the
+    /// single shared index an untagged record, or any deployment that
+    /// never turned on ingest's TenantResolver, still lands in. Default
     /// matches the path convention deploy/operator's Tenant controller
     /// and enterprise/internal/rbacstore's seeded default data source
     /// already assume (`/var/lib/sentry-search/tenants/<id>`).
     pub tenants_index_path: PathBuf,
+    /// Base URL of enterprise-auth's HTTP API, e.g.
+    /// `http://enterprise-auth:8082` -- same env var name and "empty
+    /// means off" shape as ingest/internal/config's own
+    /// ENTERPRISE_AUTH_URL. When set (together with
+    /// ENTERPRISE_AUTH_SERVICE_TOKEN below), tenants::ActiveTenantTracker
+    /// gates consumer.rs's write-routing on a polled active-tenant
+    /// allowlist -- see that module's doc comment for why this needed a
+    /// network call instead of direct Postgres access. When unset,
+    /// write-routing behaves exactly as it did before that tracker
+    /// existed: any syntactically-valid tenant_id is trusted.
+    pub enterprise_auth_url: Option<String>,
+    /// RoleService Bearer credential this process presents to
+    /// enterprise-auth's GET /internal/active-tenants -- minted via the
+    /// existing `enterprise-auth -mint-service-token search` (the flag
+    /// is already generic over caller name, no backend change needed to
+    /// mint one for a new caller). Required together with
+    /// enterprise_auth_url above; Config::load fails if exactly one of
+    /// the two is set, rather than silently running with the tracker
+    /// half-configured.
+    pub enterprise_auth_service_token: Option<String>,
 }
 
 impl Config {
@@ -29,6 +50,14 @@ impl Config {
             .parse()
             .context("COMMIT_INTERVAL_MS must be a number")?;
 
+        let enterprise_auth_url = getenv_opt("ENTERPRISE_AUTH_URL");
+        let enterprise_auth_service_token = getenv_opt("ENTERPRISE_AUTH_SERVICE_TOKEN");
+        if enterprise_auth_url.is_some() != enterprise_auth_service_token.is_some() {
+            anyhow::bail!(
+                "ENTERPRISE_AUTH_URL and ENTERPRISE_AUTH_SERVICE_TOKEN must be set together, or neither -- got exactly one"
+            );
+        }
+
         Ok(Self {
             // Rust's SocketAddr parser needs a full address, unlike Go's
             // net package (ingest/api's ":PORT" convention won't parse
@@ -49,6 +78,8 @@ impl Config {
                 "TENANTS_INDEX_PATH",
                 "/var/lib/sentry-search/tenants",
             )),
+            enterprise_auth_url,
+            enterprise_auth_service_token,
         })
     }
 }
@@ -56,3 +87,7 @@ impl Config {
 fn getenv(key: &str, fallback: &str) -> String {
     std::env::var(key).unwrap_or_else(|_| fallback.to_string())
 }
+
+fn getenv_opt(key: &str) -> Option<String> {
+    std::env::var(key).ok().filter(|v| !v.is_empty())
+}

search/src/consumer.rs

@@ -10,6 +10,7 @@ use crate::config::Config;
 use crate::logsv1;
 use crate::offsets::OffsetStore;
 use crate::registry::IndexRegistry;
+use crate::tenants::ActiveTenantTracker;
 
 /// Kafka message header a resolved tenant ID rides in, attached by
 /// ingest's gRPC front end. Mirrors `ingest/internal/grpcserver.
@@ -30,7 +31,12 @@ const TENANT_ID_HEADER_KEY: &str = "tenant_id";
 /// discovered dynamically, since it has to match what
 /// /transport/provision-topics.sh actually created anyway (documented
 /// cross-component contract, same as the topic name already is).
-pub async fn run(cfg: Arc<Config>, registry: Arc<IndexRegistry>, partition_count: i32) -> Result<()> {
+pub async fn run(
+    cfg: Arc<Config>,
+    registry: Arc<IndexRegistry>,
+    partition_count: i32,
+    active_tenants: Option<Arc<ActiveTenantTracker>>,
+) -> Result<()> {
     let client = ClientBuilder::new(cfg.redpanda_brokers.clone())
         .build()
         .await
@@ -65,9 +71,10 @@ pub async fn run(cfg: Arc<Config>, registry: Arc<IndexRegistry>, partition_count
         let client = Arc::clone(&client);
         let registry = Arc::clone(&registry);
         let offsets = Arc::clone(&offsets);
+        let active_tenants = active_tenants.clone();
         let topic = cfg.redpanda_topic.clone();
         handles.push(tokio::spawn(async move {
-            consume_partition(client, topic, partition, start_offset, registry, offsets).await
+            consume_partition(client, topic, partition, start_offset, registry, offsets, active_tenants).await
         }));
     }
 
@@ -100,6 +107,7 @@ async fn consume_partition(
     start_offset: i64,
     registry: Arc<IndexRegistry>,
     offsets: Arc<Mutex<OffsetStore>>,
+    active_tenants: Option<Arc<ActiveTenantTracker>>,
 ) -> Result<()> {
     let partition_client = client
         .partition_client(topic.clone(), partition, UnknownTopicHandling::Error)
@@ -143,17 +151,33 @@ async fn consume_partition(
             }
 
             let tenant_id = tenant_id_from_headers(&record_and_offset.record.headers);
+
+            // Fail-closed active-tenant gate (see tenants.rs's doc
+            // comment) -- only applies to tagged records and only when
+            // a tracker is actually configured, matching resolve()'s
+            // own "empty tenant_id always means the default index"
+            // rule and this codebase's "off unless configured" default
+            // everywhere else. This is the check registry.rs's `resolve`
+            // doc comment used to name as missing entirely.
+            if !tenant_id.is_empty() {
+                if let Some(tracker) = &active_tenants {
+                    if !tracker.is_active(&tenant_id).await {
+                        tracing::warn!(record_id = %rec.record_id, tenant_id, "skipping record: tenant is not active");
+                        continue;
+                    }
+                }
+            }
+
             let index = match registry.resolve(&tenant_id).await {
                 Ok(index) => index,
                 Err(e) => {
                     // Shouldn't normally happen -- ingest/grpcserver only
                     // ever attaches a tenant_id it validated against a
-                    // real credential -- but an unsafe/malformed
-                    // tenant_id is a hard skip, never a silent fall-back
-                    // to the default or any other tenant's index. See
-                    // registry.rs's doc comment for the one residual gap
-                    // this consumer doesn't close (no active-tenant
-                    // check, since this process has no Postgres access).
+                    // real credential, and the active-tenant gate above
+                    // already refused anything not currently active when
+                    // configured -- but an unsafe/malformed tenant_id is
+                    // a hard skip regardless, never a silent fall-back to
+                    // the default or any other tenant's index.
                     tracing::error!(error = %e, record_id = %rec.record_id, tenant_id, "skipping record: failed to resolve tenant index");
                     continue;
                 }

search/src/main.rs

@@ -4,6 +4,7 @@ mod grpc;
 mod index;
 mod offsets;
 mod registry;
+mod tenants;
 
 pub mod logsv1 {
     tonic::include_proto!("sentry.logs.v1");
@@ -32,6 +33,24 @@ async fn main() -> Result<()> {
 
     let cfg = Arc::new(Config::load().context("loading config")?);
 
+    // Off unless ENTERPRISE_AUTH_URL/ENTERPRISE_AUTH_SERVICE_TOKEN are
+    // both set (Config::load already rejects exactly one being set).
+    // Blocks startup entirely on failure, same "fail hard, let the
+    // orchestrator restart" posture enterprise-ingest's main.go already
+    // uses when its own required startup fetch (rbacstore.
+    // ListProvisionedDataSources) fails -- see tenants.rs's doc comment
+    // for why a partial/degraded startup isn't the safer choice here.
+    let active_tenants = match (&cfg.enterprise_auth_url, &cfg.enterprise_auth_service_token) {
+        (Some(url), Some(token)) => {
+            tracing::info!(url, "active-tenant write-routing gate enabled");
+            Some(tenants::ActiveTenantTracker::start(url, token).await?)
+        }
+        _ => {
+            tracing::info!("ENTERPRISE_AUTH_URL not set -- write-routing has no active-tenant gate");
+            None
+        }
+    };
+
     let index = Arc::new(
         SearchIndex::open_or_create(&cfg.index_path).context("opening tantivy index")?,
     );
@@ -48,8 +67,9 @@ async fn main() -> Result<()> {
 
     let consumer_cfg = Arc::clone(&cfg);
     let consumer_registry = Arc::clone(&registry);
+    let consumer_active_tenants = active_tenants.clone();
     let consumer_handle = tokio::spawn(async move {
-        if let Err(e) = consumer::run(consumer_cfg, consumer_registry, partition_count).await {
+        if let Err(e) = consumer::run(consumer_cfg, consumer_registry, partition_count, consumer_active_tenants).await {
             tracing::error!(error = %e, "redpanda consumer exited with error");
         }
     });

search/src/registry.rs

@@ -22,29 +22,27 @@ use crate::index::SearchIndex;
 /// path every Phase 0-3 deployment, and every untagged record, still
 /// uses.
 ///
-/// **Known residual gap, disclosed rather than silently accepted**:
-/// unlike the read side (gated by `enterprise/internal/searchclient`'s
-/// `TenantChecker`, which refuses to even issue a search for a tenant
-/// that isn't `active` in `rbacstore`) and unlike ClickHouse's write
-/// side (`enterprise/internal/chwriter.Registry`, built once at startup
-/// from `rbacstore.ListProvisionedDataSources` -- `active` tenants
-/// only, so an unrecognized `tenant_id` has no writer and the whole
-/// batch is refused), this registry's `resolve` has no equivalent gate
-/// on the write path: `consumer.rs` calls it directly, with no Postgres
-/// access to check tenant status against, the same reason this
-/// module's doc comment used to give for the old read-side gap
-/// `TenantChecker` was built to close. A syntactically-valid `tenant_id`
-/// on an ingest credential that's still valid but should have been
-/// revoked (deprovisioning does not yet revoke `ingest_credentials`
-/// rows -- see `/CLAUDE.md`'s Phase 4 non-goals) can therefore cause an
-/// index directory to be silently created here for a tenant that isn't
-/// really active. The blast radius is narrow -- an orphan, isolated,
-/// empty-except-for-that-tenant's-own-traffic index directory, not
-/// cross-tenant data exposure, and only reachable with a real signed
-/// ingest credential, not by an arbitrary caller -- but it is real, not
-/// hypothetical. Closing it fully would mean giving `search` (AGPL
-/// core, no `enterprise/` import allowed) some way to learn which
-/// tenants are actually active; not designed yet.
+/// `resolve` itself still has no active-tenant gate of its own -- it
+/// will happily open-or-create an index for any syntactically-valid
+/// `tenant_id`, active or not. That's deliberate: this struct's job is
+/// managing index lifecycles, not policy, the same separation
+/// `clickhousewriter.Writer` (mechanism) vs. `chwriter.Registry`
+/// (policy: which tenants get a writer at all) draws on the ClickHouse
+/// side. The gate lives one layer up, at each caller:
+/// `enterprise/internal/searchclient.TenantChecker` for the read side
+/// (refuses to even issue a search for a tenant that isn't `active` in
+/// `rbacstore`, via a direct Postgres-backed query since that code runs
+/// in `enterprise/`), and `consumer.rs`'s `tenants::ActiveTenantTracker`
+/// for the write side (a polled allowlist fetched from a new
+/// `enterprise-auth` endpoint over HTTP -- `search` is AGPL core with no
+/// Postgres access and no `enterprise/` import allowed, so it needed a
+/// network boundary instead of an import one, the same shape
+/// `ingest/internal/grpcserver.TenantResolver` already uses against the
+/// same service). Both gates are optional at this layer -- `resolve`
+/// itself works identically whether or not either caller happens to
+/// gate it -- so a future caller that forgets to gate would silently
+/// reopen this exact class of gap; see `consumer.rs`'s call site for
+/// the write side's enforcement.
 pub struct IndexRegistry {
     default_index: Arc<SearchIndex>,
     tenants_root: PathBuf,

search/src/tenants.rs

@@ -0,0 +1,223 @@
+use anyhow::{Context, Result};
+use std::collections::HashSet;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::RwLock;
+
+/// How often the tracker re-fetches the active-tenant list after its
+/// first successful fetch. Not configurable -- no deployment has needed
+/// to tune this yet, and a hardcoded value keeps Config's surface
+/// smaller; revisit if that changes.
+const REFRESH_INTERVAL: Duration = Duration::from_secs(60);
+
+/// Tracks which tenant_ids are currently `active` in enterprise-auth's
+/// `tenants` table, polled from a new `GET /internal/active-tenants`
+/// endpoint -- this closes the one gap registry.rs's `resolve` doc
+/// comment used to name: `search` (AGPL core) has no Postgres access,
+/// so unlike `chwriter.Registry` (an active-tenants-only snapshot built
+/// from `rbacstore.ListProvisionedDataSources` at `enterprise-ingest`
+/// startup) or the read side (gated by `enterprise/internal/
+/// searchclient.TenantChecker`, backed by `rbacstore.TenantIsActive`
+/// directly), `consumer.rs`'s write-routing had no allowlist at all --
+/// any syntactically-valid `tenant_id` on a still-valid-but-should-
+/// have-been-revoked ingest credential could get an index directory
+/// created for it.
+///
+/// Network boundary, not import boundary -- same shape
+/// `ingest/internal/grpcserver`'s `TenantResolver` already uses against
+/// this exact service, just Rust calling Go instead of Go calling Go,
+/// and authenticated the same way `/alerting` authenticates to `/api`:
+/// a long-lived RoleService Bearer credential
+/// (`enterprise-auth -mint-service-token search`), not a tenant-scoped
+/// one -- this tracker proves "I am the search service," never "I may
+/// act as tenant X."
+///
+/// Off unless configured: only constructed when both
+/// `ENTERPRISE_AUTH_URL` and `ENTERPRISE_AUTH_SERVICE_TOKEN` are set
+/// (see config.rs). When they aren't, `consumer.rs` holds `None` and
+/// skips the gate entirely -- every tagged write is routed exactly as
+/// it was before this tracker existed, the same "off unless configured"
+/// default every other optional integration point in this codebase
+/// uses.
+pub struct ActiveTenantTracker {
+    tenants: RwLock<HashSet<String>>,
+}
+
+impl ActiveTenantTracker {
+    /// Blocks until the first fetch succeeds. A cold start with
+    /// enterprise-auth unreachable must not silently accept every
+    /// tenant_id it sees -- that's the exact gap this tracker exists to
+    /// close -- so there is deliberately no empty-set-and-keep-going
+    /// fallback here; callers should refuse to start the write-routing
+    /// consumer at all if this returns an error. Once constructed,
+    /// periodic refreshes are best-effort: a transient failure logs and
+    /// keeps serving the last-known-good set rather than clearing it
+    /// (see the spawned task below) -- only the very first fetch is
+    /// fail-closed-to-refusing-startup.
+    pub async fn start(base_url: &str, service_token: &str) -> Result<Arc<Self>> {
+        let client = reqwest::Client::new();
+        let initial = fetch_active_tenants(&client, base_url, service_token)
+            .await
+            .context("fetching initial active-tenant list from enterprise-auth")?;
+        tracing::info!(count = initial.len(), "loaded initial active-tenant list");
+
+        let tracker = Arc::new(Self {
+            tenants: RwLock::new(initial),
+        });
+
+        let refresh_tracker = Arc::clone(&tracker);
+        let base_url = base_url.to_string();
+        let service_token = service_token.to_string();
+        tokio::spawn(async move {
+            let mut ticker = tokio::time::interval(REFRESH_INTERVAL);
+            ticker.tick().await; // fires immediately -- start() already fetched once, skip it
+            loop {
+                ticker.tick().await;
+                match fetch_active_tenants(&client, &base_url, &service_token).await {
+                    Ok(fresh) => {
+                        let count = fresh.len();
+                        *refresh_tracker.tenants.write().await = fresh;
+                        tracing::debug!(count, "refreshed active-tenant list");
+                    }
+                    Err(e) => {
+                        // No staleness ceiling: a prolonged enterprise-auth
+                        // outage means the allowlist just doesn't grow or
+                        // shrink until connectivity resumes, disclosed here
+                        // rather than degrading further (e.g. clearing the
+                        // set, which would stop every tenant's indexing on
+                        // one control-plane blip -- a worse blast radius
+                        // than staleness).
+                        tracing::error!(error = %e, "failed to refresh active-tenant list, keeping last-known-good set");
+                    }
+                }
+            }
+        });
+
+        Ok(tracker)
+    }
+
+    pub async fn is_active(&self, tenant_id: &str) -> bool {
+        self.tenants.read().await.contains(tenant_id)
+    }
+}
+
+#[derive(serde::Deserialize)]
+struct ActiveTenantsResponse {
+    tenant_ids: Vec<String>,
+}
+
+async fn fetch_active_tenants(
+    client: &reqwest::Client,
+    base_url: &str,
+    service_token: &str,
+) -> Result<HashSet<String>> {
+    let resp = client
+        .get(format!("{base_url}/internal/active-tenants"))
+        .bearer_auth(service_token)
+        .send()
+        .await
+        .context("sending request")?
+        .error_for_status()
+        .context("non-2xx response")?
+        .json::<ActiveTenantsResponse>()
+        .await
+        .context("parsing response body")?;
+    Ok(resp.tenant_ids.into_iter().collect())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    use tokio::net::TcpListener;
+
+    /// Minimal hand-rolled HTTP/1.1 server -- one dependency-free helper
+    /// rather than pulling in a mocking crate for the one endpoint this
+    /// module ever calls. Reads one request, hands it (as raw bytes) to
+    /// `respond`, writes back exactly what `respond` returns, then
+    /// closes -- enough to exercise real reqwest request construction
+    /// (the Bearer header, the URL path) and real response parsing, not
+    /// a fake client substituted in.
+    async fn spawn_fake_server(
+        respond: impl Fn(&str) -> String + Send + Sync + 'static,
+    ) -> String {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            loop {
+                let (mut stream, _) = match listener.accept().await {
+                    Ok(v) => v,
+                    Err(_) => return,
+                };
+                let mut buf = vec![0u8; 8192];
+                let n = stream.read(&mut buf).await.unwrap_or(0);
+                let request = String::from_utf8_lossy(&buf[..n]).to_string();
+                let response = respond(&request);
+                let _ = stream.write_all(response.as_bytes()).await;
+            }
+        });
+        format!("http://{addr}")
+    }
+
+    fn json_response(status_line: &str, body: &str) -> String {
+        format!(
+            "{status_line}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
+            body.len()
+        )
+    }
+
+    #[tokio::test]
+    async fn start_fetches_and_serves_the_initial_list() {
+        let base_url = spawn_fake_server(|_req| {
+            json_response("HTTP/1.1 200 OK", r#"{"tenant_ids":["acme","globex"]}"#)
+        })
+        .await;
+
+        let tracker = ActiveTenantTracker::start(&base_url, "test-token")
+            .await
+            .expect("start should succeed against a healthy fake server");
+
+        assert!(tracker.is_active("acme").await);
+        assert!(tracker.is_active("globex").await);
+        assert!(!tracker.is_active("initech").await);
+    }
+
+    #[tokio::test]
+    async fn start_sends_the_bearer_token() {
+        let base_url = spawn_fake_server(|req| {
+            if req.contains("authorization: Bearer secret-token") {
+                json_response("HTTP/1.1 200 OK", r#"{"tenant_ids":["acme"]}"#)
+            } else {
+                json_response("HTTP/1.1 401 Unauthorized", r#"{"error":"no credentials presented"}"#)
+            }
+        })
+        .await;
+
+        let tracker = ActiveTenantTracker::start(&base_url, "secret-token")
+            .await
+            .expect("start should succeed once the fake server sees the right token");
+        assert!(tracker.is_active("acme").await);
+    }
+
+    #[tokio::test]
+    async fn start_fails_closed_when_the_first_fetch_fails() {
+        let base_url = spawn_fake_server(|_req| {
+            json_response("HTTP/1.1 401 Unauthorized", r#"{"error":"invalid or expired credentials"}"#)
+        })
+        .await;
+
+        let result = ActiveTenantTracker::start(&base_url, "wrong-token").await;
+        assert!(
+            result.is_err(),
+            "expected start() to fail (not silently start with an empty/permissive allowlist) when the first fetch fails"
+        );
+    }
+
+    #[tokio::test]
+    async fn start_fails_closed_when_the_server_is_unreachable() {
+        // Port 1 is (almost certainly) not listening -- connection refused,
+        // not a slow timeout, so this test stays fast.
+        let result = ActiveTenantTracker::start("http://127.0.0.1:1", "any-token").await;
+        assert!(result.is_err());
+    }
+}