rfgen.credentials

Centralizes credential resolution so vendor SDKs receive a uniform interface. The module defines the BaseCredentialsProvider ABC, the ResolvedCredentials bundle returned by every resolve(scope) call, the StaticCredentialsConfig config block, and the concrete providers shipped by the core package and the per-cloud extras.

Warning

Pre-implementation. This page describes proposed contracts. Class signatures, parameter types, config field names, and behavior are subject to change before code lands. Once implementation exists, content here will be regenerated from docstrings or sourced from running tests.

Module summary

from rfgen.credentials import StaticCredentialsConfig, StaticCredentialsProvider

config = StaticCredentialsConfig(
    scopes={
        "storage": {"endpoint": "https://example.local/s3", "key": "AKIA..."},
    },
)
provider = StaticCredentialsProvider(config)
creds = provider.resolve("storage")
assert creds.provider_name == "static"
assert creds.scope == "storage"
# `creds.raw` is a string-keyed string-valued dict ready to hand to a backend.

How these classes relate

The module exposes four roles: one ABC contract, one shared return type, one static-only config input, and four concrete providers.

              CONTRACT (ABC)
              BaseCredentialsProvider
              - provider_name: ClassVar[str]
              - resolve(scope) -> ResolvedCredentials
                      |
    +-----------------+----------------+----------------+
    |                 |                |                |
StaticCredentials  GcpADCProvider  AwsDefault       AzureDefault
Provider           (rfgen[gcp])    Provider         Provider
   ^                               (rfgen[aws])     (rfgen[azure])
   |
StaticCredentialsConfig          every resolve() returns
(static-only config input)       ResolvedCredentials (SHARED RETURN TYPE)

Contract. BaseCredentialsProvider is the ABC every provider subclasses. It fixes resolve(scope), the provider_name registry key, and the redacted_summary() telemetry surface.

Shared return type. ResolvedCredentials is the frozen Pydantic model returned by every resolve(scope) call, on every provider. Downstream code reads raw, expires_at, provider_name, and scope without caring which subclass produced it.

Static-only config input. StaticCredentialsConfig pairs only with StaticCredentialsProvider. Cloud providers take no config block because their SDKs already discover credentials: gcloud ADC, the boto3 default chain, and azure-identity.DefaultAzureCredential.

Concrete implementations. The four subclasses (Static, GCP, AWS, Azure) differ only in where credentials come from. All satisfy the same ABC and all return the same ResolvedCredentials shape.

When you’d use which

Scenario

Provider

Local dev, CI without cloud auth, fixture tests

StaticCredentialsProvider

Production on GCP (Vertex, Dataproc, GCS)

GcpADCProvider

Production on AWS (SageMaker, EMR, S3)

AwsDefaultProvider

Production on Azure (Synapse, Blob)

AzureDefaultProvider

Downstream consumers (a store, an inference client, an executor) call provider.resolve(scope) and work against the returned ResolvedCredentials regardless of which subclass produced it. That is the point of the ABC.

Class index

Class

Kind

Notes

BaseCredentialsProvider

ABC

Every provider subclasses it. Documented under ABC below.

ResolvedCredentials

Pydantic v2 model (frozen)

Resolved credential bundle returned by every resolve call.

StaticCredentialsConfig

Pydantic v2 model (frozen)

Config block for StaticCredentialsProvider.

StaticCredentialsProvider

concrete

Reads scope-keyed credential dicts from a validated config block. Core (no cloud SDK).

GcpADCProvider

concrete

Application Default Credentials for GCP. Ships in rfgen[gcp].

AwsDefaultProvider

concrete

boto3 default credential chain. Ships in rfgen[aws].

AzureDefaultProvider

concrete

azure-identity DefaultAzureCredential. Ships in rfgen[azure].

Valid scopes

The closed set of valid scope strings is fixed at this layer:

_VALID_SCOPES = ("storage", "llm", "executor")

resolve(scope) raises ConfigError for any value outside this set. Adding a new scope is a documented schema change, not a runtime extension.

Provider registration

Providers are discovered via the entry-point group rfgen.credentials using EntryPointRegistry. The module exports a process-wide credentials_registry instance:

from rfgen.credentials import credentials_registry

names = [m.name for m in credentials_registry.discover()]

First-party providers (static, gcp_adc, aws_default, azure_default) are registered through the same entry-point group.

Logging policy

Resolved credentials are never logged at INFO or above. Downstream telemetry calls redacted_summary on the provider instance, which returns a string-keyed string-valued dict containing only the provider name and provider class. Concrete subclasses that override redacted_summary MUST NOT include credential material.

Lazy SDK imports

Importing rfgen.credentials never imports google-auth, boto3, or azure-identity. Cloud SDKs are imported lazily inside their concrete provider modules at construction time. Constructing a cloud provider without the matching extra installed raises BackendUnavailableError whose message names the missing extra. Example for the GCP path:

GcpADCProvider requires the 'gcp' extra. Install it with `pip install rfgen[gcp]`

The same shape applies to AwsDefaultProvider (aws) and AzureDefaultProvider (azure).

Shipped rfgen.workload_identity contract

rfgen.workload_identity is the shipped, secret-free boundary for describing federated workload identity. It is separate from the proposed rfgen.credentials provider hierarchy above: it does not resolve a backend credential bundle and it does not import or call a cloud SDK. A caller supplies the provider-specific acquisition adapter, which may hold SDK credential objects privately, and rfgen returns only auditable identity metadata.

from datetime import UTC, datetime, timedelta

from rfgen.workload_identity import (
    IdentityDescriptorV1,
    IdentityProvider,
    WorkloadIdentityResolver,
)

now = datetime.now(UTC)

def fetch(provider: IdentityProvider, audience: str, issued_at: datetime) -> IdentityDescriptorV1:
    return IdentityDescriptorV1(
        provider=provider,
        subject="service-account:dataset-builder",
        audience=audience,
        tenant_or_account="project-123",
        credential_kind="federated",
        issued_at=issued_at,
        expires_at=issued_at + timedelta(hours=1),
        refresh_after=issued_at + timedelta(minutes=48),
        source_fingerprint="issuer-config-sha256:example",
    )

resolver = WorkloadIdentityResolver(fetch, environment="PRODUCTION")
identity = resolver.resolve(
    IdentityProvider.GCP_ADC,
    "https://service.example",
    now=now,
)

Supported provider mechanisms

IdentityProvider is a closed StrEnum. Its valid values are "aws_web_identity", "gcp_adc", and "azure_managed_identity". These values name the federated mechanism; they do not select an SDK adapter or create a cloud session.

IdentityDescriptorV1

IdentityDescriptorV1 is a frozen Pydantic v2 model with extra="forbid". It carries provider, subject, audience, tenant_or_account, credential_kind, issued_at, expires_at, refresh_after, and source_fingerprint. All string fields must be non-blank and must not contain token or secret material. The descriptor is the only identity object that crosses rfgen logging, manifest, or API boundaries.

The timestamps must be timezone-aware. expires_at must be after issued_at. refresh_after is required to equal the earlier of 80% of the credential lifetime and five minutes before expiration. This makes refresh timing deterministic rather than adapter-defined. cache_key is a secret-free tuple of provider, audience, tenant or account, and source fingerprint.

WorkloadIdentityResolver

WorkloadIdentityResolver(fetcher, *, environment="DEVELOPMENT", static_credentials=None) keeps descriptors only in a process-local in-memory cache. In PRODUCTION, passing non-empty static_credentials raises IdentityConfigurationError; it is a guard against configuring static keys, not a static-key implementation or a persistence policy.

resolve(provider, audience, *, now, force_refresh=False) validates its inputs, uses one in-flight fetch per (provider, audience), and returns a cached descriptor only while it remains safe. It refreshes when the supplied time is before issuance, reaches refresh_after, or is within five minutes of expiration. force_refresh=True requests a refresh, while concurrent callers share the result instead of stampeding the provider. clear() drops all process-local cached descriptors.

The injected fetcher must return a descriptor for the requested provider and audience. It must not return an already unsafe descriptor. Those violations raise IdentityConfigurationError or IdentityExpiredError and do not enter the cache. Provider failures are classified without retaining their message: authorization failures raise IdentityAuthorizationError, invalid setup raises IdentityConfigurationError, expired assertions raise IdentityExpiredError, and transport or other provider failures raise IdentityUnavailableError.

Boundary with cloud integrations

This module deliberately has no AWS, Google Cloud, or Azure SDK dependency and does not persist descriptors. Optional cloud integrations may implement the injected fetcher behind their own dependency boundary. They must keep raw tokens, assertions, and SDK credential objects out of IdentityDescriptorV1, logs, manifests, and the resolver cache.

ABC: BaseCredentialsProvider

class BaseCredentialsProvider(ABC):
    provider_name: ClassVar[str]

    @abstractmethod
    def resolve(self, scope: str) -> ResolvedCredentials: ...

    def redacted_summary(self) -> dict[str, str]: ...

The abstract base class every concrete provider subclasses. The class attribute provider_name: ClassVar[str] is the stable registry key referenced by config (credentials.provider) and by audit logs.

Subclass enforcement

BaseCredentialsProvider.__init_subclass__ runs at class-definition time and raises TypeError if a non-abstract subclass does not declare a non-empty provider_name. A subclass that forgets to set the identifier fails before any instance is constructed.

Method: resolve

@abstractmethod
def resolve(self, scope: str) -> ResolvedCredentials

Resolves credentials for the requested scope and returns a ResolvedCredentials bundle.

Parameters

Name

Type

Required

Default

Description

scope

str

yes

-

One of "storage", "llm", "executor". Any other value raises ConfigError.

Returns

A populated ResolvedCredentials whose provider_name matches the provider’s class attribute and whose scope matches the input.

Raises

  • ConfigError if scope is not in the documented set, or if the provider cannot satisfy the scope (for example, StaticCredentialsProvider raising on a scope absent from its config).

Spec drift note

Plan line 53 originally typed the return as dict[str, str]. Plan line 58 required an expires_at: datetime | None slot, which a flat dict cannot carry. The shipped surface returns ResolvedCredentials so both pieces of the contract hold. The plan log records the reconciliation.

Method: redacted_summary

def redacted_summary(self) -> dict[str, str]

Returns a telemetry-safe summary. The default implementation returns {"provider_name": ..., "provider_class": ...}. Subclasses MAY override to add non-secret fields (region, project ID); they MUST NOT include credential material.

Extension contract

Every BaseCredentialsProvider subclass MUST satisfy:

  1. provider_name declared. Concrete subclasses declare provider_name: ClassVar[str] with a non-empty value. Enforced via __init_subclass__.

  2. Scope validation. resolve(scope) raises ConfigError for any value outside {"storage", "llm", "executor"}. Use the inherited helper BaseCredentialsProvider._check_scope(scope).

  3. Idempotency. Repeated calls to resolve(scope) with the same scope return semantically equivalent credentials. Tokens MUST NOT be expired at the time of return; callers refresh on demand using expires_at.

  4. Lazy imports. Cloud-specific SDK imports MUST be deferred to construction (or the first resolve call). Importing rfgen.credentials MUST NOT pull a cloud SDK.

  5. No secrets in provider_name. The class attribute appears in config and logs. It MUST NOT contain credential material.

  6. No secrets in logs. Resolved credentials are never logged at INFO or above. Telemetry uses redacted_summary.

class rfgen.credentials.ResolvedCredentials

class ResolvedCredentials(BaseModel):
    model_config = ConfigDict(frozen=True, extra="forbid", arbitrary_types_allowed=True)

    raw: dict[str, str]
    expires_at: datetime | None = None
    provider_name: str = ""
    scope: str = ""

Frozen Pydantic v2 model returned by every resolve call. Treated as immutable after resolution.

Fields

Field

Type

Description

raw

dict[str, str]

Cross-provider credential material. Concrete cloud providers MAY also stash SDK-native objects on a sibling slot; raw is the portable contract.

expires_at

datetime | None

UTC expiry timestamp. None for credentials that do not expire. Callers refresh on demand instead of caching forever.

provider_name

str

The provider_name of the producing provider. Copied here for downstream audit.

scope

str

The scope passed to resolve(...).

class rfgen.credentials.StaticCredentialsConfig

class StaticCredentialsConfig(BaseModel):
    model_config = ConfigDict(frozen=True, extra="forbid")

    scopes: dict[str, dict[str, str]] = Field(default_factory=dict)
    expires_at: datetime | None = None

Pydantic v2 config block consumed by StaticCredentialsProvider. Holds a per-scope mapping from scope name to credential dict.

Fields

Field

Type

Description

scopes

dict[str, dict[str, str]]

Mapping from scope name ("storage", "llm", "executor") to a string-keyed credential dict. Unknown scope keys are rejected at validation time.

expires_at

datetime | None

Optional UTC expiry shared by all scopes; primarily for tests.

Validation

A model_validator(mode="after") runs at construction and raises Pydantic’s ValidationError if any key in scopes is outside _VALID_SCOPES. The error message lists the unknown keys and the expected set.

class rfgen.credentials.StaticCredentialsProvider

class StaticCredentialsProvider(BaseCredentialsProvider):
    provider_name: ClassVar[str] = "static"

    def __init__(self, config: StaticCredentialsConfig) -> None: ...
    def resolve(self, scope: str) -> ResolvedCredentials: ...

Reads credentials from a validated StaticCredentialsConfig and returns them wrapped in a ResolvedCredentials. Ships in core; sufficient for tests, CI, and local runs that do not touch a cloud. Imports no cloud SDK.

Constructor

Name

Type

Required

Description

config

StaticCredentialsConfig

yes

Validated config block whose scopes field already passed scope-key validation.

Method: resolve

resolve(scope) validates the scope, looks it up in config.scopes, and returns a ResolvedCredentials populated with raw (a copy of the configured dict), expires_at (from the config), provider_name = "static", and scope (the input).

Missing-scope behavior

A scope that is valid (in _VALID_SCOPES) but not present in config.scopes raises ConfigError, never a silent empty dict. The error message names the requested scope and the configured scope keys.

config = StaticCredentialsConfig(scopes={"storage": {"endpoint": "..."}})
provider = StaticCredentialsProvider(config)

provider.resolve("storage")        # OK
provider.resolve("llm")            # raises ConfigError: no credentials for scope 'llm'
provider.resolve("nonexistent")    # raises ConfigError: unknown scope

class rfgen.credentials.GcpADCProvider

class GcpADCProvider(BaseCredentialsProvider):
    provider_name: ClassVar[str] = "gcp_adc"

    def __init__(
        self,
        *,
        project: str | None = None,
        location: str | None = None,
        quota_project: str | None = None,
        impersonate_service_account: str | None = None,
        scopes: tuple[str, ...] | None = None,
        request_timeout_s: float = 30.0,
    ) -> None: ...

Application Default Credentials provider for GCP. Ships in rfgen[gcp]. The class is declared in rfgen.credentials so the registry can list the provider name even when the extra is absent; the constructor itself runs the lazy import gate.

Constructor parameters

Name

Type

Required

Default

Description

project

str | None

no

None

Vertex AI project passed to Gemini clients. Non-None values are stripped and must not be blank. If None, resolve("llm") uses the ADC project returned by google.auth.default(...), then falls back to quota_project.

location

str | None

no

None

Vertex AI location passed to Gemini clients, for example us-central1. Non-None values are stripped and must not be blank. If unset, clients use their own default location behavior.

quota_project

str | None

no

None

GCP project to bill quota and API usage to. Non-None values are stripped and must not be blank. Passed to google.auth.default(quota_project_id=...) and used as a fallback project value when ADC does not return a project.

impersonate_service_account

str | None

no

None

Email of a service account to impersonate via short-lived tokens. Non-None values are stripped and must not be blank. Requires the caller to hold roles/iam.serviceAccountTokenCreator on the target.

scopes

tuple[str, ...] | None

no

None

OAuth2 scopes to request. If None, scopes are inferred at resolve time from the credentials scope.

request_timeout_s

float

no

30.0

Positive finite timeout, in seconds, for ADC metadata-server and token-exchange HTTP calls made through google.auth.default(request=...). If google-auth supplies a stricter per-call timeout, rfgen preserves the stricter value.

Missing-extra behavior

Constructing without google-auth installed raises BackendUnavailableError with the message:

GcpADCProvider requires the 'gcp' extra. Install it with `pip install rfgen[gcp]`

The error’s context dict carries missing_module and extra so callers can disambiguate programmatically.

Method: resolve

resolve(scope) validates the scope and returns a ResolvedCredentials whose raw carries string metadata used by downstream SDK clients. For llm, ADC uses the cloud-platform scope and returns project, location when configured, and quota_project when configured. The SDK-native google-auth credentials object is stored in extras["credentials"] so Gemini can create a Vertex AI client without exposing tokens in raw.

The provider passes a timeout-bound google.auth.transport.requests.Request into google.auth.default(request=...). That timeout applies during ADC metadata-server probing and token exchange. If impersonate_service_account is set, the returned credentials object is wrapped with google.auth.impersonated_credentials.Credentials using the same resolved scopes.

class rfgen.credentials.AwsDefaultProvider

class AwsDefaultProvider(BaseCredentialsProvider):
    provider_name: ClassVar[str] = "aws_default"

    def __init__(
        self,
        *,
        profile_name: str | None = None,
        region_name: str | None = None,
    ) -> None: ...

Resolves credentials via the standard boto3 default chain (env vars, shared credentials file, IAM role on the EC2 / ECS / EKS host). Ships in rfgen[aws]. Imports boto3 lazily.

Constructor parameters

Name

Type

Required

Default

Description

profile_name

str | None

no

None

Named profile in ~/.aws/credentials. If None, the default profile chain applies.

region_name

str | None

no

None

AWS region for downstream session construction. If None, the boto3 region resolution chain applies.

Missing-extra behavior

Constructing without boto3 installed raises BackendUnavailableError:

AwsDefaultProvider requires the 'aws' extra. Install it with `pip install rfgen[aws]`

class rfgen.credentials.AzureDefaultProvider

class AzureDefaultProvider(BaseCredentialsProvider):
    provider_name: ClassVar[str] = "azure_default"

    def __init__(
        self,
        *,
        tenant_id: str | None = None,
        managed_identity_client_id: str | None = None,
    ) -> None: ...

Wraps azure.identity.DefaultAzureCredential, which walks the standard Azure chain (env vars, managed identity, Azure CLI login, Visual Studio Code login, in that order). Ships in rfgen[azure]. Imports azure-identity lazily.

Constructor parameters

Name

Type

Required

Default

Description

tenant_id

str | None

no

None

Azure AD tenant ID. If None, the default tenant resolution applies.

managed_identity_client_id

str | None

no

None

Client ID for a user-assigned managed identity. If None, the system-assigned identity (when present) is used.

Missing-extra behavior

Constructing without azure-identity installed raises BackendUnavailableError:

AzureDefaultProvider requires the 'azure' extra. Install it with `pip install rfgen[azure]`

See Also