b2c_tooling_sdk.auth
Authentication strategies and helpers for the B2C tooling SDK.
Mirrors the @salesforce/b2c-tooling-sdk/auth subpath export. Each strategy implements the AuthStrategy protocol (an async fetch that injects credentials and handles retry/refresh). The persistent session store here reads and writes the same auth-sessions.json file as the B2C CLI, so tokens are shared across the Python and TypeScript tooling.
Classes
AccessTokenResponse
class AccessTokenResponseAccess token response from Account Manager.
Fields
| Name | Type |
|---|---|
access_token | str |
expires | datetime |
scopes | list[str] |
DecodedJWT
class DecodedJWTA decoded (unverified) JWT.
Fields
| Name | Type |
|---|---|
header | dict[str, Any] |
payload | dict[str, Any] |
AuthStrategy
class AuthStrategy(Protocol)Protocol implemented by every authentication strategy.
Implementations must inject the auth header and handle their own 401 retry/refresh inside fetch.
fetch method
async def fetch(url: str, *, method: str = 'GET', headers: dict[str, str] | None = None, content: Any = None, **kwargs: Any) -> httpx.ResponsePerform an authenticated request and return the response.
ScopedAuthStrategy
class ScopedAuthStrategy(AuthStrategy, Protocol)An AuthStrategy that can also mint/return tokens and manage scopes.
get_authorization_header method
async def get_authorization_header() -> strReturn the full Authorization header value (e.g. Bearer ...).
invalidate_token method
def invalidate_token() -> NoneInvalidate the cached token, forcing re-auth on the next request.
with_additional_scopes method
def with_additional_scopes(additional_scopes: list[str]) -> ScopedAuthStrategyReturn a copy of this strategy with additional_scopes merged in.
get_access_token_for_cascade method
async def get_access_token_for_cascade(candidates: list[list[str]]) -> strResolve a scope cascade, returning the first token Account Manager accepts.
BasicAuthConfig
class BasicAuthConfigBasic authentication (username / access-key). Used for WebDAV.
Fields
| Name | Type |
|---|---|
username | str |
password | str |
OAuthAuthConfig
class OAuthAuthConfigOAuth authentication configuration for OCAPI / platform APIs.
Fields
| Name | Type | Default |
|---|---|---|
client_id | str | |
client_secret | str | None | None |
scopes | list[str] | None | None |
account_manager_host | str | None | None |
jwt_cert_path | str | None | None |
jwt_key_path | str | None | None |
jwt_passphrase | str | None | None |
redirect_uri | str | None | None |
open_browser | Callable[[str], Awaitable[None]] | None | None |
ApiKeyAuthConfig
class ApiKeyAuthConfigAPI key authentication (MRT and external services).
Fields
| Name | Type | Default |
|---|---|---|
key | str | |
header_name | str | None | None |
AuthConfig
class AuthConfigCombined authentication configuration used by B2CInstance.
Fields
| Name | Type | Default |
|---|---|---|
basic | BasicAuthConfig | None | None |
oauth | OAuthAuthConfig | None | None |
api_key | ApiKeyAuthConfig | None | None |
auth_methods | list[AuthMethod] | None | None |
AuthCredentials
class AuthCredentialsFlat credential bundle accepted by resolve_auth_strategy.
Fields
| Name | Type | Default |
|---|---|---|
client_id | str | None | None |
client_secret | str | None | None |
scopes | list[str] | None | None |
account_manager_host | str | None | None |
username | str | None | None |
password | str | None | None |
api_key | str | None | None |
api_key_header_name | str | None | None |
redirect_uri | str | None | None |
open_browser | Callable[[str], Awaitable[None]] | None | None |
extra | dict[str, Any] | field(default_factory=dict) |
OAuthStrategy
class OAuthStrategyOAuth 2.0 client-credentials authentication strategy.
:example:
from b2c_tooling_sdk.auth import OAuthStrategy
auth = OAuthStrategy(OAuthConfig(
client_id="your-client-id",
client_secret="your-client-secret",
scopes=["sfcc.products"],
))
response = await auth.fetch("https://api.example.com/products")fetch method
async def fetch(url: str, *, method: str = 'GET', headers: dict[str, str] | None = None, content: Any = None, dispatcher: httpx.AsyncBaseTransport | None = None, **kwargs: Any) -> httpx.ResponsePerform an authenticated request, injecting a bearer token and retrying once on a post-success 401.
get_authorization_header method
async def get_authorization_header() -> strReturn the Authorization header value (Bearer <token>).
get_jwt method
async def get_jwt() -> DecodedJWTReturn the decoded (unverified) access-token JWT.
get_token_response method
async def get_token_response() -> AccessTokenResponseReturn the full token response (token + expiry + scopes), using the cache when valid.
invalidate_token method
def invalidate_token() -> NoneInvalidate every cached token for this client/method/AM-host identity.
with_additional_scopes method
def with_additional_scopes(additional_scopes: list[str]) -> OAuthStrategyReturn a new strategy with additional_scopes merged into the configured scopes.
get_access_token_for_cascade method
async def get_access_token_for_cascade(candidates: list[list[str]]) -> strResolve a scope cascade, returning the first token AM accepts.
Each candidate is merged with this strategy's base scopes. Pass 1 scans the cache for a token satisfying any candidate; pass 2 requests each candidate from AM in order, skipping invalid_scope rejections and rethrowing anything else.
OAuthConfig
class OAuthConfigConfiguration for OAuthStrategy (client-credentials grant).
JwtOAuthStrategy
class JwtOAuthStrategyOAuth 2.0 JWT Bearer authentication strategy (RFC 7523).
Differs from client credentials: uses a public/private key pair instead of a secret, sends a self-signed short-lived JWT as client_assertion in the POST body, and shares the module-level token cache under the jwt method.
fetch method
async def fetch(url: str, *, method: str = 'GET', headers: dict[str, str] | None = None, content: Any = None, dispatcher: httpx.AsyncBaseTransport | None = None, **kwargs: Any) -> httpx.ResponsePerform a request with JWT Bearer auth, retrying once on a post-success 401.
get_authorization_header method
async def get_authorization_header() -> strReturn the Authorization header value (Bearer <token>).
get_jwt method
async def get_jwt() -> DecodedJWTReturn the decoded (unverified) access-token JWT.
get_token_response method
async def get_token_response() -> AccessTokenResponseReturn the full token response, using the cache when valid.
invalidate_token method
def invalidate_token() -> NoneEvict every cached token for this client/AM-host JWT identity.
with_additional_scopes method
def with_additional_scopes(additional_scopes: list[str]) -> JwtOAuthStrategyReturn a new strategy with additional_scopes merged into the configured scopes.
get_access_token_for_cascade method
async def get_access_token_for_cascade(candidates: list[list[str]]) -> strResolve a scope cascade for the JWT flow (mirrors OAuthStrategy.get_access_token_for_cascade).
JwtOAuthConfig
class JwtOAuthConfigConfiguration for JwtOAuthStrategy.
PkceOAuthStrategy
class PkceOAuthStrategyOAuth 2.0 Authorization Code Flow with PKCE (public clients).
fetch method
async def fetch(url: str, *, method: str = 'GET', headers: dict[str, str] | None = None, content: Any = None, dispatcher: httpx.AsyncBaseTransport | None = None, **kwargs: Any) -> httpx.ResponsePerform a request with PKCE auth, retrying once on a post-success 401.
get_authorization_header method
async def get_authorization_header() -> strReturn the Authorization header value (Bearer <token>).
get_jwt method
async def get_jwt() -> DecodedJWTReturn the decoded (unverified) access-token JWT.
get_token_response method
async def get_token_response() -> AccessTokenResponseReturn the full token response, refreshing or running the browser flow as needed.
invalidate_token method
def invalidate_token() -> NoneDrop only the cached access token; the refresh token is preserved for silent renewal.
PkceOAuthConfig
class PkceOAuthConfigConfiguration for the OAuth Authorization Code + PKCE flow.
PkceGrantUnsupportedError
class PkceGrantUnsupportedError(Exception)Raised when the Authorization Code + PKCE flow fails because the client is not registered for that grant (e.g. a legacy implicit-only public client or a missing/mismatched redirect URI) rather than a transient or user-driven failure.
PkceWithImplicitFallbackStrategy keys its automatic fallback off this type so it retries with the legacy implicit flow ONLY for grant/registration failures — never for user-cancel, state mismatch, or a port-in-use error.
PkceWithImplicitFallbackStrategy
class PkceWithImplicitFallbackStrategyWraps a PkceOAuthStrategy, falling back to implicit on a grant error.
fetch method
async def fetch(url: str, *, method: str = 'GET', headers: dict[str, str] | None = None, content: Any = None, dispatcher: httpx.AsyncBaseTransport | None = None, **kwargs: Any) -> httpx.ResponseFetch via PKCE, falling back to implicit on a grant-unsupported error.
get_authorization_header method
async def get_authorization_header() -> strReturn the Authorization header, falling back to implicit on a grant error.
get_jwt method
async def get_jwt() -> DecodedJWTReturn the decoded access-token JWT, falling back to implicit on a grant error.
get_token_response method
async def get_token_response() -> AccessTokenResponseReturn the full token response, falling back to implicit on a grant error.
invalidate_token method
def invalidate_token() -> NoneInvalidate cached tokens on both the PKCE and (if present) implicit strategies.
ImplicitOAuthStrategy
class ImplicitOAuthStrategyOAuth 2.0 Implicit Grant flow (deprecated; public clients only).
fetch method
async def fetch(url: str, *, method: str = 'GET', headers: dict[str, str] | None = None, content: Any = None, dispatcher: httpx.AsyncBaseTransport | None = None, **kwargs: Any) -> httpx.ResponsePerform a request with implicit-flow auth, retrying once on a post-success 401.
get_authorization_header method
async def get_authorization_header() -> strReturn the Authorization header value (Bearer <token>).
get_jwt method
async def get_jwt() -> DecodedJWTReturn the decoded (unverified) access-token JWT.
get_token_response method
async def get_token_response() -> AccessTokenResponseReturn the full token response, running the browser flow when the cache is stale.
invalidate_token method
def invalidate_token() -> NoneInvalidate the cached token, forcing re-authentication on the next request.
ImplicitOAuthConfig
class ImplicitOAuthConfigConfiguration for the legacy implicit OAuth flow.
StatefulOAuthStrategy
class StatefulOAuthStrategyAuth strategy that uses a persisted access token from the unified store.
No refresh — on expiry/401, the session is cleared and the caller is expected to re-authenticate.
fetch method
async def fetch(url: str, *, method: str = 'GET', headers: dict[str, str] | None = None, content: Any = None, dispatcher: httpx.AsyncBaseTransport | None = None, **kwargs: Any) -> httpx.ResponsePerform a request with the stored token; on 401 clear the session.
get_authorization_header method
async def get_authorization_header() -> strReturn the Authorization header value (Bearer <token>).
get_token_response method
async def get_token_response() -> AccessTokenResponseReturn the current token as an AccessTokenResponse (expires/scopes from the JWT).
get_jwt method
async def get_jwt() -> DecodedJWTReturn the decoded (unverified) access-token JWT.
invalidate_token method
def invalidate_token() -> NoneDelete the persisted session and blank the in-memory access token.
StatefulOAuthStrategyOptions
class StatefulOAuthStrategyOptionsOptions for StatefulOAuthStrategy (kept for API parity with the TS SDK).
BasicAuthStrategy
class BasicAuthStrategyBasic authentication strategy.
:example:
from b2c_tooling_sdk.auth import BasicAuthStrategy
auth = BasicAuthStrategy("username", "access-key")
response = await auth.fetch("https://webdav.example.com/path")fetch method
async def fetch(url: str, *, method: str = 'GET', headers: dict[str, str] | None = None, content: Any = None, dispatcher: httpx.AsyncBaseTransport | None = None, **kwargs: Any) -> httpx.ResponsePerform a request with the Authorization: Basic header set.
get_authorization_header method
async def get_authorization_header() -> strReturn the Authorization: Basic header value.
ApiKeyStrategy
class ApiKeyStrategyAPI key authentication strategy.
:example:
# MRT API (Bearer token) -> Authorization: Bearer {key}
auth = ApiKeyStrategy(api_key, "Authorization")
# Custom header -> x-api-key: {key}
auth = ApiKeyStrategy(api_key, "x-api-key")fetch method
async def fetch(url: str, *, method: str = 'GET', headers: dict[str, str] | None = None, content: Any = None, dispatcher: httpx.AsyncBaseTransport | None = None, **kwargs: Any) -> httpx.ResponsePerform a request with the API-key header set.
get_authorization_header method
async def get_authorization_header() -> strReturn the header value (Bearer {key} for Authorization, else the raw key).
AvailableAuthMethods
class AvailableAuthMethodsResult of checking which auth methods have credentials available.
Fields
| Name | Type |
|---|---|
available | list[AuthMethod] |
unavailable | list[UnavailableAuthMethod] |
UnavailableAuthMethod
class UnavailableAuthMethodA method that is missing at least one required credential.
Fields
| Name | Type |
|---|---|
method | AuthMethod |
reason | str |
AuthMiddleware
class AuthMiddleware(Protocol)Middleware for authentication requests (analogous to openapi-fetch middleware).
on_request method
async def on_request(request: httpx.Request) -> httpx.Request | NoneCalled before the auth request is sent; may mutate or replace it.
on_response method
async def on_response(request: httpx.Request, response: httpx.Response) -> httpx.Response | NoneCalled after the auth response is received; may mutate or replace it.
AuthMiddlewareProvider
class AuthMiddlewareProvider(Protocol)Supplies AuthMiddleware for auth requests.
Fields
| Name | Type |
|---|---|
name | str |
get_middleware method
def get_middleware() -> AuthMiddleware | NoneReturn middleware to apply, or None to skip.
AuthMiddlewareRegistry
class AuthMiddlewareRegistryCollects middleware from providers, returning them in registration order.
Fields
| Name | Type | Description |
|---|---|---|
size | int | Number of registered providers. |
register method
def register(provider: AuthMiddlewareProvider) -> NoneRegister a middleware provider.
unregister method
def unregister(name: str) -> boolRemove a provider by name; return True if one was removed.
get_middleware method
def get_middleware() -> list[AuthMiddleware]Collect middleware from all providers, in registration order.
clear method
def clear() -> NoneClear all registered providers (primarily for testing).
get_provider_names method
def get_provider_names() -> list[str]Return the names of all registered providers.
AuthSession
class AuthSessionOne persisted authentication session, keyed by client_id.
Field names are snake_case in Python but serialize to the camelCase keys the TypeScript SDK writes (clientId, accessToken, refreshToken, ...).
Fields
| Name | Type | Default |
|---|---|---|
client_id | str | |
flow | AuthSessionFlow | |
access_token | str | |
pkce_unsupported | bool | None | None |
refresh_token | str | None | None |
sub | str | None | None |
expires_at | str | None | None |
scopes | list[str] | None | None |
account_manager_host | str | None | None |
last_used_at | str | None | None |
to_json method
def to_json() -> dict[str, Any]Serialize to a dict with camelCase keys, omitting None fields (matching JSON.stringify).
from_json method
def from_json(data: dict[str, Any]) -> AuthSessionBuild an AuthSession from a camelCase dict written by any backend.
AuthSessionBackend
class AuthSessionBackend(Protocol)Pluggable backend for the auth-session store.
FileAuthSessionBackend
class FileAuthSessionBackendDefault JSON-file backend at <data dir>/auth-sessions.json.
Writes atomically via a temp file + rename, with the directory created 0o700 and the file written 0o600 (matching the TS backend, since the file holds long-lived PKCE refresh tokens).
Fields
| Name | Type |
|---|---|
data_dir | Path |
InMemoryAuthSessionBackend
class InMemoryAuthSessionBackendIn-memory backend, useful for tests and IDE adapters.
Functions
create_user_auth_strategy
def create_user_auth_strategy(config: PkceOAuthConfig) -> PkceOAuthStrategy | PkceWithImplicitFallbackStrategyBuild the browser-based "user" auth strategy.
Returns a plain PkceOAuthStrategy when the fallback is disabled (SFCC_DISABLE_PKCE_FALLBACK), otherwise a PkceWithImplicitFallbackStrategy.
is_pkce_fallback_disabled
def is_pkce_fallback_disabled() -> boolTrue when SFCC_DISABLE_PKCE_FALLBACK is set to any truthy value.
resolve_auth_strategy
def resolve_auth_strategy(credentials: AuthCredentials, allowed_methods: list[AuthMethod] | None = None) -> AuthStrategyResolve and create the appropriate auth strategy.
Iterates through allowed methods in priority order and returns the first strategy for which the required credentials are available.
Parameters
| Name | Type | Description |
|---|---|---|
credentials | AuthCredentials | The available credentials. |
allowed_methods | list[AuthMethod] | None | Allowed methods in priority order (defaults to ALL_AUTH_METHODS, where PKCE-based user auth is preferred over the deprecated implicit flow). |
Raises
RuntimeError— if no allowed method has the required credentials.
check_available_auth_methods
def check_available_auth_methods(credentials: AuthCredentials, allowed_methods: list[AuthMethod] | None = None) -> AvailableAuthMethodsCheck which auth methods have the required credentials available.
Parameters
| Name | Type | Description |
|---|---|---|
credentials | AuthCredentials | The available credentials. |
allowed_methods | list[AuthMethod] | None | Methods to check (defaults to ALL_AUTH_METHODS). |
Returns: The available and unavailable methods.
encode_basic_client_credentials
def encode_basic_client_credentials(client_id: str, client_secret: str) -> strBuild the Base64 payload for Authorization: Basic per RFC 6749 §2.3.1.
Parameters
| Name | Type | Description |
|---|---|---|
client_id | str | The OAuth client identifier. |
client_secret | str | The OAuth client password/secret. |
Returns: The Base64 string to place after Basic in the header.
decode_jwt
def decode_jwt(token: str) -> DecodedJWTDecode a JWT into its header and payload without verifying the signature.
Raises
ValueError— if the token is not a well-formed three-part JWT.
decode_jwt_token_info
def decode_jwt_token_info(token: str) -> tuple[datetime, list[str]]Return (expires, scopes) for a token. Propagates decode errors.
extract_jwt_scopes
def extract_jwt_scopes(payload: dict[str, Any]) -> list[str]Extract scope from a decoded JWT payload (array or space-delimited string).
is_jwt_token_valid
def is_jwt_token_valid(token: str, required_scopes: list[str] | None = None, expiry_buffer_sec: int = DEFAULT_EXPIRY_BUFFER_SEC) -> boolReturn True if the token decodes, is unexpired (with buffer), and has all scopes.
get_oauth_cache_key
def get_oauth_cache_key(client_id: str, method: str, account_manager_host: str, scopes: list[str] | None = None) -> strBuild a token cache key. Includes the auth method to keep grants distinct.
get_cached_oauth_token
def get_cached_oauth_token(cache_key: str, required_scopes: list[str] | None = None) -> AccessTokenResponse | NoneReturn a cached token if present, unexpired, and covering required_scopes.
set_cached_oauth_token
def set_cached_oauth_token(cache_key: str, token_response: AccessTokenResponse) -> NoneStore a token in the global cache.
find_cached_token_satisfying
def find_cached_token_satisfying(identity_prefix: str, required_scopes: list[str]) -> AccessTokenResponse | NoneReturn the first non-expired cached token (matching identity_prefix) whose scopes ⊇ required_scopes.
Used by cascade resolution: a token granted with broader scopes automatically satisfies a later request needing a narrower scope, with no extra AM round trip.
invalidate_cached_tokens_for_identity
def invalidate_cached_tokens_for_identity(identity_prefix: str) -> NoneEvict every cached token for an identity prefix (host:clientId:method:).
Cascade-resolving strategies cache tokens under merged-scope keys, so deleting only the base key on a 401 would leave a rejected merged token cached. Clearing by identity prefix evicts all of them so the retry re-requests from AM.
reset_oauth_cache_for_testing
def reset_oauth_cache_for_testing() -> NoneClear the module-level token cache and pending-request map (tests only).
apply_auth_request_middleware
async def apply_auth_request_middleware(request: httpx.Request, middleware: list[AuthMiddleware]) -> httpx.RequestApply every on_request hook in order, accumulating modifications.
apply_auth_response_middleware
async def apply_auth_response_middleware(request: httpx.Request, response: httpx.Response, middleware: list[AuthMiddleware]) -> httpx.ResponseApply every on_response hook in order, accumulating modifications.
get_default_data_dir
def get_default_data_dir(*, data_directory: str | None = None, environment: dict[str, str] | None = None, home_directory: str | None = None, platform: str | None = None) -> PathResolve the shared oclif-compatible B2C data directory (the session store).
Mirrors @oclif/core's Config.dataDir — and the sibling get_b2c_config_directory — so the SDK reads the same auth-sessions.json the b2c CLI writes:
$B2C_DATA_DIR | $XDG_DATA_HOME | (win32 %LOCALAPPDATA%) | ~/.local/share then /b2c.
Note: oclif's data dir uses the XDG ~/.local/share base on macOS too — not ~/Library/Application Support (that path is only oclif's cache dir).
set_auth_session_backend
def set_auth_session_backend(backend: AuthSessionBackend | None) -> NoneRegister an auth-session backend. Pass None to fall back to the file backend.
get_auth_session_backend
def get_auth_session_backend() -> AuthSessionBackendReturn the active backend (lazily creating the file-backed default).
initialize_file_auth_session_store
def initialize_file_auth_session_store(data_dir: str | os.PathLike[str]) -> NoneInstall a FileAuthSessionBackend pointed at data_dir.
find_auth_session
def find_auth_session(client_id: str) -> AuthSession | NoneRead the stored session for client_id (or None).
save_auth_session
def save_auth_session(session: AuthSession) -> NoneWrite a session, replacing any prior record for the same client_id.
delete_auth_session
def delete_auth_session(client_id: str) -> NoneDelete the session for client_id.
list_auth_sessions
def list_auth_sessions() -> list[AuthSession]List all stored sessions (for diagnostics).
clear_all_auth_sessions
def clear_all_auth_sessions() -> NoneRemove every stored session. Used by auth logout.
is_auth_session_token_valid
def is_auth_session_token_valid(session: AuthSession, required_scopes: list[str] | None = None, expiry_buffer_sec: int = DEFAULT_EXPIRY_BUFFER_SEC, required_client_id: str | None = None) -> boolReturn True if the session's access token is present, unexpired, and in-scope.
Performs no network calls — validity is derived from the JWT exp/scope.
reset_auth_session_store_for_testing
def reset_auth_session_store_for_testing() -> NoneReset the active backend (tests). The next call falls back to the file default.
Attributes
AuthMethod
AuthMethod = Literal['client-credentials', 'jwt', 'user', 'implicit', 'basic', 'api-key']ALL_AUTH_METHODS
ALL_AUTH_METHODS: list[AuthMethod] = ['client-credentials', 'jwt', 'user', 'implicit', 'basic', 'api-key']DEFAULT_EXPIRY_BUFFER_SEC
DEFAULT_EXPIRY_BUFFER_SEC = 60global_auth_middleware_registry
global_auth_middleware_registry = AuthMiddlewareRegistry()AuthSessionFlow
AuthSessionFlow = Literal['pkce', 'implicit', 'client-credentials']