faucet-auth

Shared, single-flight authentication providers for the faucet-stream ecosystem. Each provider implements faucet_core::AuthProvider — a live entity that owns a token cache and refresh lifecycle.

The point of this crate is token sharing: one provider instance, wrapped in an Arc, is handed to every connector that authenticates against the same identity provider. So N connectors (or N matrix rows) hitting one IdP share a single token with single-flight refresh, instead of each racing to mint or rotate its own. That is the difference between one token request per run and one per connector per refresh window — and, for rotating refresh tokens, the difference between working and invalidating each other.

Feature highlights

• Four provider types — a fixed static credential, OAuth2 client_credentials, OAuth2 refresh_token with rotation capture, and a generic JSONPath-extracting token endpoint.
• Single-flight refresh — concurrent callers during a refresh await the one in-flight fetch; they don't stampede the token endpoint.
• Force-refresh on rejection — a connector that gets a 401 calls invalidate(stale) and receives a freshly-fetched token. Concurrent invalidations of the same token collapse into one refresh via compare-and-swap.
• Refresh-token rotation capture — oauth2_refresh captures a rotated refresh_token from each response in place, so a single active access token plus a rotating refresh token can be shared safely across many connectors.
• Secret-safe Debug — every provider's Debug impl renders secrets (client_secret, refresh token, request body, cached access token) as ***; only non-secret identifiers stay visible.
• Bounded fetch timeout — providers hold a single-flight mutex across the network call, so the internal HTTP client has a 30 s request timeout: a hung IdP fails and releases the lock instead of wedging every connector that shares the provider.
• Validated config at load time — expiry_ratio is checked to be a finite number in (0, 1]; unknown provider types and missing required fields surface as FaucetError::Config before any run starts.

Installation

# As a library:
cargo add faucet-auth

# In the CLI, the shared `auth:` catalog is always available — no feature flag needed.
# Library callers who want the umbrella crate to build providers enable the `auth` feature:
cargo add faucet-stream --features auth

The faucet-cli binary always links faucet-auth to power the top-level auth: catalog, so config-driven users get every provider type out of the box.

What it provides

build_provider(&Value) is the entry point: it reads a { type, config } spec and returns a SharedAuthProvider (Arc<dyn AuthProvider>).

Provider configuration reference

Every provider is built from a { type, config } object — the project-wide adjacently-tagged auth shape. The fields below are the keys inside each provider's config.

static

A fixed credential. Exactly one of the three shapes below must be present.

oauth2 (client-credentials)

oauth2_refresh

token_endpoint

CLI usage — the top-level auth: catalog

Define a provider once in the top-level auth: catalog, then reference it from any connector via auth: { ref: <name> }. The CLI builds each named provider a single time and injects the same Arc into every connector that references it, so all the rows that share a ref share one token.

# faucet run pipeline.yaml
version: 1
auth:
  sf:
    type: oauth2_refresh
    config:
      token_url: https://<acct>.snowflakecomputing.com/oauth/token-request
      client_id: ${secret:SF_CLIENT_ID}
      client_secret: ${secret:SF_CLIENT_SECRET}
      refresh_token: ${secret:SF_REFRESH_TOKEN}
pipeline:
  sources:
    sf_table:
      type: snowflake
      config:
        account: ${vars.account}
        auth: { ref: sf }      # every row using this template shares ONE token
  sink:
    type: jsonl
    config:
      path: ./out.jsonl

Eight connectors consume shared providers via auth: { ref }: rest, graphql, xml, grpc, websocket, sink-http, elasticsearch, and snowflake. (Kafka / BigQuery / GCS keep the same { type, config } wire shape but are not bearer/token-based, so they don't take auth: { ref }.)

Provider configs can hold ${env:…} / ${file:…} / ${secret:…} / ${vault:…} / ${aws-sm:…} indirection — the secrets pass walks the auth: catalog, so a single shared provider can hold a secret-manager reference and be reused across every row.

Inline auth vs shared provider

A connector's auth field accepts either an inline { type, config } block or a { ref: <name> } pointer. Use a ref (and the auth: catalog) whenever more than one connector / matrix row authenticates against the same IdP — that's when token sharing and single-flight refresh actually pay off. A one-off connector can keep its auth inline.

Library usage

Build a provider, wrap it in an Arc, and clone it into every source/sink that should share the token via with_auth_provider:

use std::sync::Arc;
use faucet_auth::{build_provider, OAuth2RefreshProvider};
use faucet_core::SharedAuthProvider;
use serde_json::json;

# fn main() -> Result<(), Box<dyn std::error::Error>> {
// Build directly from a typed config object…
let provider: SharedAuthProvider = Arc::new(OAuth2RefreshProvider::from_config(&json!({
    "token_url": "https://idp.example/token",
    "client_id": "my-client",
    "client_secret": "s3cr3t",
    "refresh_token": "initial-rt",
}))?);

// …or from a `{ type, config }` spec, the same shape the CLI catalog uses:
let provider2: SharedAuthProvider = build_provider(&json!({
    "type": "oauth2",
    "config": {
        "token_url": "https://idp.example/token",
        "client_id": "my-client",
        "client_secret": "s3cr3t",
        "scopes": ["read"]
    }
}))?;
# let _ = (provider2,);

// Clone the Arc into every connector that should share the one token:
// let source_a = RestStreamSource::new(cfg_a)?.with_auth_provider(provider.clone());
// let source_b = RestStreamSource::new(cfg_b)?.with_auth_provider(provider.clone());
# let _ = provider;
# Ok(())
# }

Asking the provider for a credential is then a one-liner; the cache, refresh, and single-flight coordination are internal:

use faucet_core::{AuthProvider, Credential};
# async fn use_provider(provider: &dyn AuthProvider) -> Result<(), Box<dyn std::error::Error>> {
let cred: Credential = provider.credential().await?;     // cached or freshly refreshed
// On a 401, force a single-flight refresh of exactly this stale token:
let fresh = provider.invalidate(&cred).await?;
# let _ = fresh;
# Ok(())
# }

How single-flight refresh works

Each fetching provider (oauth2, oauth2_refresh, token_endpoint) holds a Mutex-guarded token cache and performs the token-endpoint call with the lock held:

1. Cache hit — credential() returns the cached token until expires_in × expiry_ratio has elapsed.
2. Cache miss / expiry — the first caller takes the lock and fetches; concurrent callers block on the same lock and, when it's released, observe the freshly-cached token. Result: one network fetch serves an arbitrary number of concurrent callers.
3. Force-refresh on 401 — a connector whose request was rejected calls invalidate(stale). This is a compare-and-swap: it refetches only if the cache still holds the stale token. If a concurrent caller already refreshed (the cache now holds a different token), the stale caller gets that new token back without a second fetch.
4. Rotation capture (oauth2_refresh only) — each refresh response may carry a rotated refresh_token; the provider stores it in place, so the next refresh uses the latest rotated token. This is exactly the case that breaks when each connector keeps its own copy of a rotating refresh token.

The internal HTTP client has a bounded 30 s request timeout so a hung or unreachable IdP fails the fetch and releases the single-flight lock, rather than wedging every connector that shares the provider.

expiry_ratio (default 0.9) must be a finite number in (0, 1], validated at construction. A value ≤ 0 (or NaN) would expire every token immediately, defeating the cache and single-flight refresh; a value > 1 would treat the token as valid past its real expiry, causing 401s mid-use. Both are rejected at config-load time.

Feature flags

This crate has no optional Cargo features of its own. It is pulled in by:

• the faucet-cli binary (always — for the top-level auth: catalog);
• the faucet-stream umbrella crate's auth feature, for library callers who want build_provider available alongside the connectors.

Troubleshooting / FAQ

See also

• Authentication cookbook — the full { type, config } vs { ref } model, per-connector auth methods, and the shared auth: catalog.
• Secrets cookbook — feeding ${secret:…} / ${vault:…} references into provider configs.
• faucet-core — defines the AuthProvider trait, Credential enum, and SharedAuthProvider / AuthSpec types this crate implements against.

License

Licensed under either of Apache License, Version 2.0 or MIT license at your option.