base

mlflow_secrets_auth.base

Base classes and abstractions for MLflow secrets-backed authentication providers.

This module defines

• Lightweight requests.auth.AuthBase implementations for Bearer, Basic, and custom-header auth.
• SecretsBackedAuthProvider, an abstract base for MLflow RequestAuthProviders that obtain credentials from secret managers and cache them with a TTL.

Design notes

• Providers implement _fetch_secret, _get_cache_key, _get_auth_mode, and _get_ttl.
• Caching is delegated to cached_fetch and TTL validation to validate_ttl.
• Secrets are parsed centrally via parse_secret_json and must resolve to either:
  • {"token": ""} OR
  • {"username": "...", "password": "..."}
• Header name can be configured; "Authorization" is normalized to the canonical header.

All logging goes through safe_log to avoid leaking sensitive values.

BasicAuth(username, password, header_name=DEFAULT_AUTH_HEADER)

Bases: AuthBase

HTTP Basic authentication for requests.

If a non-standard header is configured, the base64 credentials are put into that header.

Attributes:

Name	Type	Description
username		Basic auth username.
password		Basic auth password.
header_name		Target header (defaults to "Authorization").

Initialize basic authentication.

Parameters:

Name	Type	Description	Default
username	str	Username for basic auth.	required
password	str	Password for basic auth.	required
header_name	str	HTTP header to inject (defaults to "Authorization").	DEFAULT_AUTH_HEADER

Source code in src/mlflow_secrets_auth/base.py

def __init__(self, username: str, password: str, header_name: str = DEFAULT_AUTH_HEADER) -> None:
    """Initialize basic authentication.

    Args:
        username: Username for basic auth.
        password: Password for basic auth.
        header_name: HTTP header to inject (defaults to "Authorization").

    """
    self.username = username
    self.password = password
    self.header_name = header_name

__call__(r)

Attach the basic auth header to the outgoing request.

Source code in src/mlflow_secrets_auth/base.py

def __call__(self, r: requests.PreparedRequest) -> requests.PreparedRequest:
    """Attach the basic auth header to the outgoing request."""
    creds = f"{self.username}:{self.password}".encode()
    r.headers[self.header_name] = f"Basic {base64.b64encode(creds).decode()}"
    return r

BearerAuth(token, header_name=DEFAULT_AUTH_HEADER)

Bases: AuthBase

Bearer token authentication for requests.

The token is injected as: <header_name>: Bearer <token>

Attributes:

Name	Type	Description
token		Opaque bearer token.
header_name		Target header (defaults to "Authorization").

Initialize bearer authentication.

Parameters:

Name	Type	Description	Default
token	str	Bearer token string.	required
header_name	str	HTTP header to inject (defaults to "Authorization").	DEFAULT_AUTH_HEADER

Source code in src/mlflow_secrets_auth/base.py

def __init__(self, token: str, header_name: str = DEFAULT_AUTH_HEADER) -> None:
    """Initialize bearer authentication.

    Args:
        token: Bearer token string.
        header_name: HTTP header to inject (defaults to "Authorization").

    """
    self.token = token
    self.header_name = header_name

__call__(r)

Attach the bearer token header to the outgoing request.

Source code in src/mlflow_secrets_auth/base.py

def __call__(self, r: requests.PreparedRequest) -> requests.PreparedRequest:
    """Attach the bearer token header to the outgoing request."""
    r.headers[self.header_name] = f"Bearer {self.token}"
    return r

CustomHeaderAuth(token, header_name)

Bases: AuthBase

Custom header authentication for requests (token placed as-is).

Attributes:

Name	Type	Description
token		Opaque token to inject.
header_name		Target header.

Initialize custom header authentication.

Parameters:

Name	Type	Description	Default
token	str	Token value to inject directly.	required
header_name	str	Header name where the token is placed.	required

Source code in src/mlflow_secrets_auth/base.py

def __init__(self, token: str, header_name: str) -> None:
    """Initialize custom header authentication.

    Args:
        token: Token value to inject directly.
        header_name: Header name where the token is placed.

    """
    self.token = token
    self.header_name = header_name

__call__(r)

Attach the opaque token to the configured header.

Source code in src/mlflow_secrets_auth/base.py

def __call__(self, r: requests.PreparedRequest) -> requests.PreparedRequest:
    """Attach the opaque token to the configured header."""
    r.headers[self.header_name] = self.token
    return r

SecretData

Bases: TypedDict

Structured representation of parsed secret material.

SecretsBackedAuthProvider(provider_name, default_ttl=DEFAULT_TTL_SECONDS)

Bases: RequestAuthProvider, ABC

Abstract base class for secrets-backed MLflow auth providers.

Subclasses implement secret retrieval for a specific backend (e.g., Vault, AWS, Azure) and supply configuration inputs (cache key, auth mode, TTL).

This class handles

• Provider enablement checks.
• Host allowlisting for get_request_auth.
• Cache + TTL validation.
• Secret parsing and Auth object construction.

Parameters:

Name	Type	Description	Default
provider_name	str	Stable identifier used for logging and configuration.	required
default_ttl	int	Fallback TTL in seconds if configured TTL is invalid.	DEFAULT_TTL_SECONDS

Attributes:

Name	Type	Description
provider_name		Provider identifier.
default_ttl		Default TTL for cache.
logger		Namespaced logger instance.

Initialize the provider base.

Parameters:

Name	Type	Description	Default
provider_name	str	Identifier of the provider (e.g., "vault").	required
default_ttl	int	Default cache TTL in seconds.	DEFAULT_TTL_SECONDS

Source code in src/mlflow_secrets_auth/base.py

def __init__(self, provider_name: str, default_ttl: int = DEFAULT_TTL_SECONDS) -> None:
    """Initialize the provider base.

    Args:
        provider_name: Identifier of the provider (e.g., "vault").
        default_ttl: Default cache TTL in seconds.

    """
    self.provider_name = provider_name
    self.default_ttl = default_ttl
    self.logger = setup_logger(f"mlflow_secrets_auth.{provider_name}")

get_auth()

Return a requests Auth object (no URL filtering).

This method is used by MLflow when a per-request URL is not available.

Returns:

Type	Description
AuthBase | None	A requests.auth.AuthBase instance or None when disabled/unavailable.

Source code in src/mlflow_secrets_auth/base.py

def get_auth(self) -> requests.auth.AuthBase | None:
    """Return a `requests` Auth object (no URL filtering).

    This method is used by MLflow when a per-request URL is not available.

    Returns:
        A `requests.auth.AuthBase` instance or None when disabled/unavailable.

    """
    if not self._is_enabled():
        safe_log(self.logger, logging.DEBUG, DEBUG_PROVIDER_NOT_ENABLED.format(provider=self.provider_name))
        return None

    try:
        secret_data = self._fetch_secret_cached()
        return None if not secret_data else self._create_auth(secret_data)
    except Exception as e:  # pragma: no cover — defensive guard
        safe_log(self.logger, logging.ERROR, ERROR_UNEXPECTED_PROVIDER.format(provider=self.provider_name, error=e))
        return None

get_name()

Return the provider name (instance method in recent MLflow versions).

Returns:

Type	Description
str	Provider name for MLflow plugin discovery.

Source code in src/mlflow_secrets_auth/base.py

def get_name(self) -> str:
    """Return the provider name (instance method in recent MLflow versions).

    Returns:
        Provider name for MLflow plugin discovery.

    """
    return self.provider_name

get_request_auth(url)

Return a requests Auth object for a given MLflow request URL.

Applies host allowlisting to avoid credential leakage.

Parameters:

Name	Type	Description	Default
url	str	Full request URL for an MLflow call.	required

Returns:

Type	Description
AuthBase | None	A requests.auth.AuthBase instance or None if not allowed/available.

Source code in src/mlflow_secrets_auth/base.py

def get_request_auth(self, url: str) -> requests.auth.AuthBase | None:
    """Return a `requests` Auth object for a given MLflow request URL.

    Applies host allowlisting to avoid credential leakage.

    Args:
        url: Full request URL for an MLflow call.

    Returns:
        A `requests.auth.AuthBase` instance or None if not allowed/available.

    """
    if not self._is_enabled():
        safe_log(self.logger, logging.DEBUG, DEBUG_PROVIDER_NOT_ENABLED.format(provider=self.provider_name))
        return None

    allowed_hosts = get_allowed_hosts()
    if not is_host_allowed(url, allowed_hosts):
        hostname = urlparse(url).hostname or "<unknown>"
        safe_log(
            self.logger,
            logging.INFO,
            INFO_HOST_NOT_ALLOWED.format(hostname=hostname),
        )
        return None

    try:
        secret_data = self._fetch_secret_cached()
        if not secret_data:
            safe_log(self.logger, logging.WARNING, WARNING_FETCH_FAILED.format(provider=self.provider_name))
            return None
        return self._create_auth(secret_data)
    except ValueError as e:
        # Configuration or parsing error — not fatal to the request.
        safe_log(self.logger, logging.WARNING, WARNING_CONFIG_ERROR.format(provider=self.provider_name, error=e))
        return None
    except Exception as e:  # pragma: no cover — defensive
        safe_log(self.logger, logging.ERROR, ERROR_UNEXPECTED_PROVIDER.format(provider=self.provider_name, error=e))
        return None