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.

flow (composable multi-step auth)

A small declarative auth program (#511): an optional login / pre-flight request chain whose responses are captured by JSONPath, credential placement across header / query / cookie / body, a pluggable HMAC request signer, and a dynamic per-session base-URL — on top of the single-flight machinery. It's the single biggest source-side unlock for session-cookie / multi-step ERP APIs (Bullhorn, Acumatica, SAP, Sage/Intacct, SkySlope, …).

Signing a login step (#541)

A login/pre-flight request can carry its own sign: block — the same SignSpec shape used in apply. It's computed with a fresh ${ts}/${nonce} clock per step and attached as a header, so APIs whose login call is itself HMAC-signed (e.g. SkySlope) are expressible from config. On the first step nothing has been captured yet, so its template may reference only ${param.*} / ${env:*} / ${ts} / ${nonce} (later steps also see values captured by earlier steps).

auth:
  skyslope:
    type: flow
    config:
      steps:
        - request:
            method: POST
            url: "${param.base_url}/auth/login"
            json: { clientId: "${param.client_id}" }
            sign:                                   # sign the login step itself
              alg: hmac_sha256
              key: "${param.secret_key}"
              template: "${param.client_id}:${param.client_secret}:${ts}"
              encoding: base64
              into: { header: "Authorization", format: "SS ${param.access_key}:${sig}" }
          capture: { session: "$.session.token" }
      apply:
        - { sign: { alg: hmac_sha256, key: "${param.secret_key}", template: "${param.client_id}:${param.client_secret}:${ts}", encoding: base64, into: { header: Authorization, format: "SS ${param.access_key}:${sig}" } } }
        - { into: header, name: "Session", value: "${session}" }

Capture sources (#542)

By default a capture entry is a JSONPath string into the JSON response body ({ session: "$.session.token" } — unchanged, back-compatible). It can also be a { from, … } struct to capture from other parts of the response:

steps:
  - request: { method: POST, url: "${param.base_url}/entity/auth/login",
               json: { name: "${param.user}", password: "${secret:pw}" } }
    capture:
      session_cookie: { from: set_cookie, name: "ASP.NET_SessionId" }   # Acumatica: Set-Cookie + 204 empty body
      sess_id:        { from: xml, path: "response.operation.result.data.api.sessionid" }  # Sage Intacct: XML session id
      loc:            { from: header, name: "Location" }                # any response header
apply:
  - { into: cookie, name: "ASP.NET_SessionId", value: "${session_cookie}" }

A shared cookie jar (auto-forwarding login cookies to data requests without an explicit capture) is future work; the from: set_cookie → into: cookie path covers the same case today.

auth:
  bullhorn:
    type: flow
    config:
      steps:
        - request: { method: POST, url: "https://auth/oauth/token",
                     form: { grant_type: refresh_token, refresh_token: "${env:RT}" } }
          capture: { access_token: "$.access_token" }
        - request: { method: GET, url: "https://login/rest-services/login",
                     query: { access_token: "${access_token}" } }
          capture: { bh_rest_token: "$.BhRestToken", base_url: "$.restUrl" }
      apply:
        - { into: query, name: BhRestToken, value: "${bh_rest_token}" }
      base_url_from: "${base_url}"
      reauth_on: [401]

Header placements also apply to xml / graphql sources (via credential()); query / cookie / body placement and dynamic base-URL are consumed by the rest source (which reads the richer request_auth).

static

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

oauth2 (client-credentials)

oauth2_refresh

persist example:

auth:
  graph:
    type: oauth2_refresh
    config:
      token_url: "https://login.microsoftonline.com/${param.tenant}/oauth2/v2.0/token"
      client_id: "${secret:graph_client_id}"
      client_secret: "${secret:graph_client_secret}"
      refresh_token: "${secret:graph_seed_refresh_token}"   # seed; used only until the first rotation
      scope: "https://graph.microsoft.com/.default offline_access"  # Graph requires scope on refresh
      persist:
        path: "./state/auth"        # rotated refresh_token survives across runs

token_endpoint

apply_as example — SAP Business One session cookie:

auth:
  sap:
    type: token_endpoint
    config:
      url: "https://host:50000/b1s/v1/Login"
      body: { CompanyDB: "${param.company_db}", UserName: "${param.user}", Password: "${secret:sap_pw}" }
      token_path: "$.SessionId"
      apply_as:
        header: "Cookie"
        template: "B1SESSION={token}; CompanyDB=${param.company_db}"

oauth1 (HMAC-SHA256 request signing)

Unlike the token providers, OAuth1 has no token to fetch — it signs every request individually (the signature covers the HTTP method, URL, and query parameters). Requires the oauth1 crate feature (cargo add faucet-auth --features oauth1, or cargo install faucet-cli --features oauth1).

auth:
  netsuite:
    type: oauth1
    config:
      consumer_key: "${secret:ns_consumer_key}"
      consumer_secret: "${secret:ns_consumer_secret}"
      token: "${secret:ns_token}"
      token_secret: "${secret:ns_token_secret}"
      realm: "${param.account}"   # NetSuite account id

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.