kubex.configuration

Auto-generated reference for the kubex.configuration module.

ClientConfiguration model

kubex.configuration.configuration

AuthInfo

Bases: ConfigModel

AuthInfo contains information that describes identity information. This is use to tell the kubernetes cluster who you are.

as_ class-attribute instance-attribute

as_: str | None = Field(None, alias='as')

Impersonate is the username to impersonate. The name matches the flag.

as_groups class-attribute instance-attribute

as_groups: list[str] | None = Field(None, alias="as-groups")

ImpersonateGroups is the groups to impersonate.

as_uid class-attribute instance-attribute

as_uid: str | None = Field(None, alias='as-uid')

ImpersonateUID is the uid to impersonate.

as_user_extra class-attribute instance-attribute

as_user_extra: dict[str, list[str]] | None = Field(
    None, alias="as-user-extra"
)

ImpersonateUserExtra contains additional information for impersonated user.

auth_provider class-attribute instance-attribute

auth_provider: AuthProviderConfig | None = Field(
    None, alias="auth-provider"
)

AuthProvider specifies a custom authentication plugin for the kubernetes cluster.

client_certificate class-attribute instance-attribute

client_certificate: _OptionalPath = Field(
    None, alias="client-certificate"
)

ClientCertificate is the path to a client cert file for TLS.

client_certificate_data class-attribute instance-attribute

client_certificate_data: str | None = Field(
    None, alias="client-certificate-data"
)

ClientCertificateData contains PEM-encoded data from a client cert file for TLS. Overrides ClientCertificate.

client_key class-attribute instance-attribute

client_key: _OptionalPath = Field(None, alias='client-key')

ClientKey is the path to a client key file for TLS.

client_key_data class-attribute instance-attribute

client_key_data: str | None = Field(
    None, alias="client-key-data", repr=False
)

ClientKeyData contains PEM-encoded data from a client key file for TLS. Overrides ClientKey.

password class-attribute instance-attribute

password: _CredentialSecret = None

Password is the password for basic authentication to the kubernetes cluster.

token class-attribute instance-attribute

token: _CredentialStr = Field(None, repr=False)

Token is the bearer token for authentication to the kubernetes cluster.

token_file class-attribute instance-attribute

token_file: _OptionalPath = Field(None, alias='tokenFile')

TokenFile is the path to a file containing the bearer token for authentication to the kubernetes cluster.

username class-attribute instance-attribute

username: _CredentialStr = None

Username is the username for basic authentication to the kubernetes cluster.

AuthProviderConfig

Bases: ConfigModel

AuthProviderConfig holds the configuration for a custom authentication plugin.

config class-attribute instance-attribute

config: OIDCConfig | dict[str, str] = Field(
    default_factory=dict[str, str], repr=False
)

Config holds the auth provider configuration data.

name instance-attribute

name: str

Name is the name of the auth provider.

ClientConfiguration

ClientConfiguration(
    url: str | None = None,
    server_ca_file: Path | str | None = None,
    insecure_skip_tls_verify: bool | None = None,
    client_cert_file: Path | str | None = None,
    client_key_file: Path | str | None = None,
    token_file: Path | str | None = None,
    token: str | None = None,
    namespace: str | None = None,
    try_refresh_token: bool = False,
    refreshable_token: SupportsAuthorizationHeader
    | None = None,
    tls_server_name: str | None = None,
    username: str | None = None,
    password: str | None = None,
    impersonate_user: str | None = None,
    impersonate_uid: str | None = None,
    impersonate_groups: Sequence[str] | None = None,
    impersonate_extra: Mapping[str, Sequence[str]]
    | None = None,
)

Connection and credential settings for a client.

Parameters:

Name	Type	Description	Default
tls_server_name	str | None	Name to verify the server certificate against (and send as SNI) instead of the hostname in url; kubeconfig's tls-server-name.	None
username	str | None	Basic-auth username; used only when no token source is set.	None
password	str | None	Basic-auth password.	None
impersonate_user	str | None	Username to act as (Impersonate-User).	None
impersonate_uid	str | None	UID to act as (Impersonate-Uid).	None
impersonate_groups	Sequence[str] | None	Groups to act as (one Impersonate-Group each).	None
impersonate_extra	Mapping[str, Sequence[str]] | None	Extra user info to act with (Impersonate-Extra-<key>, one header per value).	None

Source code in kubex/configuration/configuration.py

def __init__(
    self,
    url: str | None = None,
    server_ca_file: Path | str | None = None,
    insecure_skip_tls_verify: bool | None = None,
    client_cert_file: Path | str | None = None,
    client_key_file: Path | str | None = None,
    token_file: Path | str | None = None,
    token: str | None = None,
    namespace: str | None = None,
    try_refresh_token: bool = False,
    refreshable_token: SupportsAuthorizationHeader | None = None,
    tls_server_name: str | None = None,
    username: str | None = None,
    password: str | None = None,
    impersonate_user: str | None = None,
    impersonate_uid: str | None = None,
    impersonate_groups: typing.Sequence[str] | None = None,
    impersonate_extra: typing.Mapping[str, typing.Sequence[str]] | None = None,
) -> None:
    """Connection and credential settings for a client.

    Args:
        tls_server_name: Name to verify the server certificate against (and
            send as SNI) instead of the hostname in ``url``; kubeconfig's
            ``tls-server-name``.
        username: Basic-auth username; used only when no token source is set.
        password: Basic-auth password.
        impersonate_user: Username to act as (``Impersonate-User``).
        impersonate_uid: UID to act as (``Impersonate-Uid``).
        impersonate_groups: Groups to act as (one ``Impersonate-Group`` each).
        impersonate_extra: Extra user info to act with
            (``Impersonate-Extra-<key>``, one header per value).
    """
    if try_refresh_token and token_file is None:
        raise ValueError("Token file must be provided to refresh token")
    self.refreshable_token = refreshable_token
    self.base_url = url
    if insecure_skip_tls_verify is False:
        if server_ca_file is None:
            raise ValueError("Server CA file must be provided")
    self.insecure_skip_tls_verify = insecure_skip_tls_verify
    if isinstance(server_ca_file, str):
        server_ca_file = Path(server_ca_file)
    self.server_ca_file = server_ca_file
    if isinstance(client_cert_file, str):
        client_cert_file = Path(client_cert_file)
    self.client_cert_file = client_cert_file
    if isinstance(client_key_file, str):
        client_key_file = Path(client_key_file)
    self.client_key_file = client_key_file
    self.namespace = namespace

    if isinstance(token_file, str):
        token_file = Path(token_file)

    self.token_file = token_file
    self.try_refresh_token = try_refresh_token
    self._last_token_read: float | None = None
    self._current_token: str | None = None
    self._token = token
    self.tls_server_name = tls_server_name or None
    if (username is None) != (password is None):
        raise ValueError("username and password must be provided together")
    self.username = username
    self._password = password
    if impersonate_user is None and (impersonate_groups or impersonate_extra):
        # The API server rejects group/extra impersonation without a user.
        raise ValueError(
            "impersonate_groups/impersonate_extra require impersonate_user"
        )
    self.impersonate_user = impersonate_user
    self.impersonate_uid = impersonate_uid
    self.impersonate_groups: tuple[str, ...] = tuple(impersonate_groups or ())
    self.impersonate_extra: dict[str, tuple[str, ...]] = {
        key: tuple(values) for key, values in (impersonate_extra or {}).items()
    }

impersonation_headers property

impersonation_headers: list[tuple[str, str]]

Impersonate-* headers to send on every request (possibly repeated).

token property

token: str | None

The bearer token to send.

As in client-go, the last successfully read, non-empty token_file content wins over token, which is the fallback until the file has been read. A failed read raises only when there is nothing to fall back to.

get_authorization_header async

get_authorization_header() -> str | None

Return the Authorization header value to send, if any.

Prefers refreshable_token (exec/OIDC credential plugins) over the static token/token_file path, since a kubeconfig context only ever sets one credential source.

Source code in kubex/configuration/configuration.py

async def get_authorization_header(self) -> str | None:
    """Return the `Authorization` header value to send, if any.

    Prefers `refreshable_token` (exec/OIDC credential plugins) over the
    static `token`/`token_file` path, since a kubeconfig context only
    ever sets one credential source.
    """
    if self.refreshable_token is not None:
        return await self.refreshable_token.to_header()
    token = self.token
    if token is not None:
        return f"Bearer {token}"
    if self.username is not None and self._password is not None:
        credentials = f"{self.username}:{self._password}".encode()
        return f"Basic {base64.b64encode(credentials).decode('ascii')}"
    return None

Cluster

Bases: ConfigModel

Cluster contains information about how to communicate with a kubernetes cluster.

Fields are validated leniently so that one stale entry does not break loading the whole kubeconfig; configure_from_kubeconfig() checks the selected context's cluster (server URL, referenced files) when it is used.

certificate_authority class-attribute instance-attribute

certificate_authority: _OptionalPath = Field(
    None, alias="certificate-authority"
)

CertificateAuthority is the path to a cert file for the certificate authority.

certificate_authority_data class-attribute instance-attribute

certificate_authority_data: str | None = Field(
    None, alias="certificate-authority-data"
)

CertificateAuthorityData contains PEM-encoded certificate authority certificates. Overrides CertificateAuthority.

disable_compression class-attribute instance-attribute

disable_compression: bool = Field(
    False, alias="disable-compression"
)

DisableCompression allows client to opt-out of response compression for all requests to the server. This is useful to speed up requests (specifically lists) when client-server network bandwidth is ample, by saving time on compression (server-side) and decompression (client-side): https://github.com/kubernetes/kubernetes/issues/112296.

extensions class-attribute instance-attribute

extensions: list[NamedExtension] | None = None

Extensions holds additional information. This is useful for extenders so that reads and writes don't clobber unknown fields.

insecure_skip_tls_verify class-attribute instance-attribute

insecure_skip_tls_verify: bool = Field(
    False, alias="insecure-skip-tls-verify"
)

InsecureSkipTLSVerify skips the validity check for the server's certificate. This will make your HTTPS connections insecure.

proxy_url class-attribute instance-attribute

proxy_url: str | None = Field(None, alias='proxy-url')

ProxyURL is the URL to the proxy to be used for all requests made by this client. URLs with "http", "https", and "socks5" schemes are supported. If this configuration is not provided or the empty string, the client attempts to construct a proxy configuration from http_proxy and https_proxy environment variables. If these environment variables are not set, the client does not attempt to proxy requests. socks5 proxying does not currently support spdy streaming endpoints (exec, attach, port forward).

server class-attribute instance-attribute

server: _OptionalStr = None

Server is the address of the kubernetes cluster (https://hostname:port). A value without a scheme is treated as https://.

tls_server_name class-attribute instance-attribute

tls_server_name: str | None = Field(
    None, alias="tls-server-name"
)

TLSServerName is used to check server certificate. If TLSServerName is empty, the hostname used to contact the server is used.

ConfigModel

Bases: BaseK8sModel

Base for kubeconfig / credential-plugin models.

Unlike resource models, unknown keys are dropped: nothing round-trips them, and keeping them would surface unexpected (possibly secret) kubeconfig content in reprs.

Context

Bases: ConfigModel

Context holds user values, cluster values, and extension values.

cluster class-attribute instance-attribute

cluster: str | None = None

Cluster is the name of the cluster for this context.

extensions class-attribute instance-attribute

extensions: list[NamedExtension] | None = None

Extensions holds additional information. This is useful for extenders so that reads and writes don't clobber unknown fields.

namespace class-attribute instance-attribute

namespace: str | None = None

Namespace is the default namespace to use on unspecified requests.

user class-attribute instance-attribute

user: str | None = None

User is the name of the user for this context. Unset means anonymous access.

ExecConfig

Bases: ConfigModel

ExecConfig specifies a command to provide client credentials.

api_version class-attribute instance-attribute

api_version: str | None = Field(None, alias='apiVersion')

APIVersion is the version of the API: client.authentication.k8s.io/v1 or client.authentication.k8s.io/v1beta1.

args class-attribute instance-attribute

args: list[str] | None = Field(None, repr=False)

Args to pass to the command when executing it.

command class-attribute instance-attribute

command: str = ''

Command to execute. A relative path containing a path separator is resolved against the kubeconfig file's directory; a bare name is looked up on PATH.

env class-attribute instance-attribute

env: list[dict[str, str]] | None = Field(None, repr=False)

Env defines the environment variables to pass to the command.

install_hint class-attribute instance-attribute

install_hint: str | None = Field(None, alias='installHint')

InstallHint is shown to the user when the command cannot be found.

interactive_mode class-attribute instance-attribute

interactive_mode: ExecInteractiveMode | None = Field(
    None, alias="interactiveMode"
)

InteractiveMode controls whether the plugin's stdin is connected; unset means IfAvailable.

provide_cluster_info class-attribute instance-attribute

provide_cluster_info: bool | None = Field(
    None, alias="provideClusterInfo"
)

ProvideClusterInfo passes the cluster's connection details to the plugin in KUBERNETES_EXEC_INFO.

ExecInteractiveMode

Bases: str, Enum

ExecInteractiveMode define the interactity of the child process

ALWAYS class-attribute instance-attribute

ALWAYS = 'Always'

Always get interactive

IF_AVAILABLE class-attribute instance-attribute

IF_AVAILABLE = 'IfAvailable'

Get interactive if available

NEVER class-attribute instance-attribute

NEVER = 'Never'

Never get interactive

KubeConfig

Bases: ConfigModel

Config holds the information needed to build connect to remote kubernetes clusters as a given user.

clusters class-attribute instance-attribute

clusters: list[NamedClaster] = Field(default_factory=list)

Clusters is a map of referencable names to cluster configs.

contexts class-attribute instance-attribute

contexts: list[NamedContext] = Field(default_factory=list)

Contexts is a map of referencable names to context configs.

current_context class-attribute instance-attribute

current_context: str | None = Field(
    None, alias="current-context"
)

CurrentContext is the name of the context that you would like to use by default.

users class-attribute instance-attribute

users: list[NamedAuthInfo] = Field(default_factory=list)

AuthInfos is a map of referencable names to user configs.

NamedAuthInfo

Bases: ConfigModel

NamedAuthInfo holds an AuthInfo with name.

auth_info class-attribute instance-attribute

auth_info: AuthInfo = Field(
    default_factory=AuthInfo, alias="user"
)

AuthInfo holds the auth info.

name instance-attribute

name: str

Name is the name of the auth

NamedClaster

Bases: ConfigModel

NamedCluster holds a cluster with name.

cluster class-attribute instance-attribute

cluster: Cluster = Field(default_factory=Cluster)

Cluster holds the cluster information.

name instance-attribute

name: str

Name is the name of the cluster.

NamedContext

Bases: ConfigModel

NamedContext holds a context with name.

context class-attribute instance-attribute

context: Context = Field(default_factory=Context)

Context holds the context information.

name instance-attribute

name: str

Name is the name of the context.

NamedExtension

Bases: ConfigModel

NamedExtension holds an extension with name.

extension class-attribute instance-attribute

extension: Any = None

Extension holds the extension information.

name instance-attribute

name: str

Name is the name of the extension.

OIDCConfig

Bases: ConfigModel

client_id class-attribute instance-attribute

client_id: SecretStr = Field(alias='client-id')

ClientID is the client ID for the OpenID Connect client, as described in https://tools.ietf.org/html/rfc6749#section-2.2.

client_secret class-attribute instance-attribute

client_secret: SecretStr | None = Field(
    None, alias="client-secret"
)

ClientSecret is the client secret for the OpenID Connect client, as described in https://tools.ietf.org/html/rfc6749#section-2.3. Omitted for public clients.

id_token class-attribute instance-attribute

id_token: SecretStr | None = Field(None, alias='id-token')

IDToken is the current id_token; used as-is until its exp claim is reached.

idp_certificate_authority class-attribute instance-attribute

idp_certificate_authority: _OptionalPath = Field(
    None, alias="idp-certificate-authority"
)

Path to a CA bundle for the issuer's TLS certificate. Relative to the kubeconfig file's directory.

idp_certificate_authority_data class-attribute instance-attribute

idp_certificate_authority_data: str | None = Field(
    None, alias="idp-certificate-authority-data"
)

Base64-encoded CA bundle for the issuer's TLS certificate. Overrides idp-certificate-authority.

idp_issuer_url class-attribute instance-attribute

idp_issuer_url: str = Field(alias='idp-issuer-url')

IDPIssuerURL is the URL of the OpenID issuer, only HTTPS scheme will be accepted.

refresh_token class-attribute instance-attribute

refresh_token: SecretStr | None = Field(
    None, alias="refresh-token"
)

RefreshToken is the refresh token for the OpenID Connect client, as described in https://tools.ietf.org/html/rfc6749#section-6. Redeemed only when id-token is missing or expired.

RawExtension

Bases: ConfigModel

RawExtension is used to hold extensions in external versions

object_ class-attribute instance-attribute

object_: dict[str, Any] | None = Field(None, alias="object")

Object can hold a representation of this extension - useful for working with versioned structs.

raw class-attribute instance-attribute

raw: bytes | None = None

Raw is the underlying serialization of this object.

SupportsAuthorizationHeader

Bases: Protocol

Structural type for a token source that can format its own Authorization header

describe_validation_error

describe_validation_error(
    exc: ValidationError, data: Any = None
) -> str

Summarise exc by error locations and types only.

pydantic's own message embeds the offending input, which for kubeconfig and exec-plugin data can be a credential. Pass the validated data to name list entries (e.g. users['dev']) instead of numbering them.

Source code in kubex/configuration/configuration.py

def describe_validation_error(exc: ValidationError, data: typing.Any = None) -> str:
    """Summarise ``exc`` by error locations and types only.

    pydantic's own message embeds the offending input, which for kubeconfig
    and exec-plugin data can be a credential. Pass the validated ``data`` to
    name list entries (e.g. ``users['dev']``) instead of numbering them.
    """
    details = [
        f"{_format_location(error['loc'], data)} ({error['type']})"
        for error in exc.errors(include_url=False, include_input=False)
    ]
    return f"{exc.error_count()} validation error(s): {', '.join(details)}"

Kubeconfig file loading

kubex.configuration.file_config

configure_from_kubeconfig async

configure_from_kubeconfig(
    config: KubeConfig | None = None,
    use_context: str | None = None,
    *,
    config_file: str | PathLike[str] | None = None,
) -> ClientConfiguration

Create a ClientConfiguration from a kubeconfig.

Only the selected context's cluster and user are checked (server URL, referenced files), so a stale entry for another context does not matter.

Parameters:

Name	Type	Description	Default
config	KubeConfig | None	An already-parsed kubeconfig. Relative file paths inside it are used as-is (relative to the current directory); use config_file to load a file from disk.	None
use_context	str | None	Context to use instead of current-context.	None
config_file	str | PathLike[str] | None	Kubeconfig file to load instead of $KUBECONFIG / ~/.kube/config. Relative paths inside it resolve against its directory, as with kubectl.	None

Raises:

Type	Description
FileNotFoundError	No kubeconfig file exists.
ConfigurationError	The kubeconfig, or the selected cluster/user, is invalid (e.g. a referenced certificate file is missing).
ConfigurationValueError	The context, or the cluster/user it names, is not defined. Also a ConfigurationError and a ValueError.
ValueError	Both config and config_file are given.

Source code in kubex/configuration/file_config.py

async def configure_from_kubeconfig(
    config: KubeConfig | None = None,
    use_context: str | None = None,
    *,
    config_file: str | os.PathLike[str] | None = None,
) -> ClientConfiguration:
    """Create a ClientConfiguration from a kubeconfig.

    Only the selected context's cluster and user are checked (server URL,
    referenced files), so a stale entry for another context does not matter.

    Args:
        config: An already-parsed kubeconfig. Relative file paths inside it are
            used as-is (relative to the current directory); use `config_file`
            to load a file from disk.
        use_context: Context to use instead of `current-context`.
        config_file: Kubeconfig file to load instead of `$KUBECONFIG` /
            `~/.kube/config`. Relative paths inside it resolve against its
            directory, as with kubectl.

    Raises:
        FileNotFoundError: No kubeconfig file exists.
        ConfigurationError: The kubeconfig, or the selected cluster/user, is
            invalid (e.g. a referenced certificate file is missing).
        ConfigurationValueError: The context, or the cluster/user it names, is
            not defined. Also a `ConfigurationError` and a `ValueError`.
        ValueError: Both `config` and `config_file` are given.
    """
    if config is not None and config_file is not None:
        raise ValueError("Pass either config or config_file, not both")
    if config is None:
        config = _load_kube_config(
            Path(config_file) if config_file is not None else None
        )
    current_context = use_context or config.current_context
    if not current_context:
        raise ConfigurationValueError("No current context in kubeconfig")
    context = next(
        (c.context for c in config.contexts if c.name == current_context),
        None,
    )
    if context is None:
        raise ConfigurationValueError(
            f"Context {current_context} not found in kubeconfig"
        )
    if not context.cluster:
        raise ConfigurationError(
            f"kubeconfig context {current_context!r} does not name a cluster"
        )
    cluster = next(
        (c.cluster for c in config.clusters if c.name == context.cluster),
        None,
    )
    if cluster is None:
        raise ConfigurationValueError(
            f"Cluster {context.cluster} not found in kubeconfig"
        )
    user = AuthInfo()
    if context.user:
        found_user = next(
            (u.auth_info for u in config.users if u.name == context.user),
            None,
        )
        if found_user is None:
            raise ConfigurationValueError(
                f"User {context.user} not found in kubeconfig"
            )
        user = found_user
    cluster_owner = f"cluster {context.cluster!r}"
    user_owner = f"user {context.user!r}"
    server = _server_url(context.cluster, cluster.server)
    ca_file = _data_or_file(
        cluster.certificate_authority_data,
        cluster.certificate_authority,
        cluster_owner,
        "certificate-authority",
    )
    client_cert_file = _data_or_file(
        user.client_certificate_data,
        user.client_certificate,
        user_owner,
        "client-certificate",
    )
    client_key_file = _data_or_file(
        user.client_key_data, user.client_key, user_owner, "client-key"
    )
    if user.token_file is not None and not user.token:
        _require_file(user.token_file, user_owner, "tokenFile")
    refreshable_token: SupportsAuthorizationHeader | None = None
    if user.exec is not None:
        cluster_info = (
            _exec_cluster_info(server, cluster, ca_file)
            if user.exec.provide_cluster_info
            else None
        )
        refreshable_token = ExecRefreshableToken(
            ExecAuthProvider(user.exec, cluster_info=cluster_info)
        )
    elif user.auth_provider is not None:
        if user.auth_provider.name != "oidc":
            raise ConfigurationError(
                f"configure_from_kubeconfig() does not support the "
                f"{user.auth_provider.name!r} auth-provider; only 'oidc' with a "
                "valid OIDC config is supported."
            )
        refreshable_token = OidcRefreshableToken(
            OIDCAuthProvider(_oidc_config(context.user or "", user.auth_provider))
        )
    # Basic auth needs a username; client-go sends an empty password if unset.
    username = user.username or None
    password = None
    if username is not None:
        password = user.password.get_secret_value() if user.password else ""
    if cluster.proxy_url:
        warnings.warn(
            f"kubeconfig cluster {context.cluster!r} sets proxy-url, which "
            "configure_from_kubeconfig() does not apply (it belongs on "
            "ClientOptions, which this function has no access to). Pass the "
            "equivalent proxy via ClientOptions(proxy=...) when constructing "
            "the client.",
            UserWarning,
            stacklevel=2,
        )
    return ClientConfiguration(
        url=server,
        server_ca_file=ca_file,
        # Cluster.insecure_skip_tls_verify defaults to False (not None), and
        # ClientConfiguration raises when insecure_skip_tls_verify is False
        # (rather than None) with no server_ca_file -- so only forward True,
        # leaving the common "no CA, no insecure flag" kubeconfig (system
        # trust store) as ClientConfiguration's own default of None.
        insecure_skip_tls_verify=True if cluster.insecure_skip_tls_verify else None,
        client_cert_file=client_cert_file,
        client_key_file=client_key_file,
        token=user.token or None,
        token_file=user.token_file,
        # kubectl/client-go use "default" when the context sets no namespace.
        namespace=context.namespace or DEFAULT_NAMESPACE,
        try_refresh_token=user.token_file is not None,
        refreshable_token=refreshable_token,
        tls_server_name=cluster.tls_server_name,
        username=username,
        password=password,
        impersonate_user=user.as_,
        impersonate_uid=user.as_uid,
        impersonate_groups=user.as_groups,
        impersonate_extra=user.as_user_extra,
    )

In-cluster configuration

kubex.configuration.incluster_config

configure_from_pod_env async

configure_from_pod_env(
    token_filename: Path | None = None,
    cert_filename: Path | None = None,
    kubernetes_service_host: str | None = None,
    kubernetes_service_port: str | None = None,
    try_refresh_token: bool = True,
) -> ClientConfiguration

Build a ClientConfiguration from the pod's service-account mount.

Raises:

Type	Description
ConfigurationError	KUBERNETES_SERVICE_HOST/KUBERNETES_SERVICE_PORT are not set (and not passed explicitly), i.e. the process is not running inside a cluster.

Source code in kubex/configuration/incluster_config.py

async def configure_from_pod_env(
    token_filename: Path | None = None,
    cert_filename: Path | None = None,
    kubernetes_service_host: str | None = None,
    kubernetes_service_port: str | None = None,
    try_refresh_token: bool = True,
) -> ClientConfiguration:
    """Build a ClientConfiguration from the pod's service-account mount.

    Raises:
        ConfigurationError: `KUBERNETES_SERVICE_HOST`/`KUBERNETES_SERVICE_PORT`
            are not set (and not passed explicitly), i.e. the process is not
            running inside a cluster.
    """
    if token_filename is None:
        token_filename = DEFAULT_TOKEN_FILENAME
    if cert_filename is None:
        cert_filename = DEFAULT_CERT_FILENAME
    if kubernetes_service_host is None:
        kubernetes_service_host = os.environ.get(KUBERNETES_SERVICE_HOST_ENV)
    if kubernetes_service_port is None:
        kubernetes_service_port = os.environ.get(KUBERNETES_SERVICE_PORT_ENV)
    if not kubernetes_service_host or not kubernetes_service_port:
        raise ConfigurationError(
            "Unable to load in-cluster configuration: not running inside a "
            f"Kubernetes cluster ({KUBERNETES_SERVICE_HOST_ENV} and "
            f"{KUBERNETES_SERVICE_PORT_ENV} are not set). Outside a cluster, "
            "provide a kubeconfig file (~/.kube/config or $KUBECONFIG)."
        )
    return ClientConfiguration(
        url=f"https://{_join_host_port(kubernetes_service_host, kubernetes_service_port)}",
        server_ca_file=cert_filename,
        client_cert_file=None,
        client_key_file=None,
        namespace=_read_namespace(),
        token_file=token_filename,
        try_refresh_token=try_refresh_token,
    )

Exec provider authentication

kubex.configuration.auth.exec

ExecAuthProvider

ExecAuthProvider(
    config: ExecConfig,
    cluster_info: ExecCredentialCluster | None = None,
)

Run a client-go credential plugin.

Parameters:

Name	Type	Description	Default
config	ExecConfig	The kubeconfig exec block.	required
cluster_info	ExecCredentialCluster | None	Sent to the plugin in KUBERNETES_EXEC_INFO when config.provide_cluster_info is set.	None

Source code in kubex/configuration/auth/exec.py

def __init__(
    self,
    config: ExecConfig,
    cluster_info: ExecCredentialCluster | None = None,
) -> None:
    """Run a client-go credential plugin.

    Args:
        config: The kubeconfig `exec` block.
        cluster_info: Sent to the plugin in `KUBERNETES_EXEC_INFO` when
            `config.provide_cluster_info` is set.
    """
    if not config.command:
        raise ConfigurationError("exec: no command configured")
    if config.api_version not in SUPPORTED_EXEC_API_VERSIONS:
        raise ConfigurationError(
            f"exec: unsupported apiVersion {config.api_version!r}; expected one "
            f"of {list(SUPPORTED_EXEC_API_VERSIONS)}"
        )
    self.config = config
    self.cluster_info = cluster_info
    self.env = os.environ.copy()
    for item in config.env or []:
        self.env[item["name"]] = item["value"]

refresh_token async

refresh_token() -> tuple[str, str | None]

Return the credential token and its RFC 3339 expirationTimestamp, if the plugin reported one.

Source code in kubex/configuration/auth/exec.py

async def refresh_token(self) -> tuple[str, str | None]:
    """Return the credential token and its RFC 3339 `expirationTimestamp`, if the plugin reported one."""
    credential = await self.run()
    status = credential.status
    if status is not None and status.token:
        return status.token, status.expiration_timestamp
    if status is not None and (
        status.client_certificate_data or status.client_key_data
    ):
        raise ConfigurationError(
            f"exec: plugin {self.config.command!r} returned a client certificate "
            "(clientCertificateData/clientKeyData); certificate-based exec "
            "credentials are not supported yet, only token-based ones"
        )
    raise ConfigurationValueError("exec: token not found in response")

ExecCredentialCluster

Bases: ConfigModel

Cluster connection details passed to a plugin with provideClusterInfo: true.

certificate_authority_data class-attribute instance-attribute

certificate_authority_data: str | None = Field(
    None, alias="certificate-authority-data"
)

Base64-encoded PEM CA bundle.

config class-attribute instance-attribute

config: Any = Field(None, repr=False)

The cluster's client.authentication.k8s.io/exec extension, if any.

OIDC authentication

kubex.configuration.auth.oidc

OIDCAuthProvider

OIDCAuthProvider(config: OIDCConfig)

Source code in kubex/configuration/auth/oidc.py

def __init__(self, config: OIDCConfig):
    self.config = config
    self._token_endpoint: str | None = None
    self._auth_style: AuthStyle | None = None
    self._ssl_context = _build_ssl_context(
        config.idp_certificate_authority_data, config.idp_certificate_authority
    )

refresh_token async

refresh_token() -> tuple[str, float | None]

Return a usable id_token, redeeming the refresh token only when needed.

Like client-go, the configured id-token is returned as long as its exp claim is in the future; otherwise the refresh token is redeemed for a new id_token (rotating the in-memory refresh token if the IdP issues a new one). Raises ConfigurationError when there is neither a valid id-token nor a refresh token.

Returns the id_token and its own exp claim (Unix epoch seconds), if it is a decodable JWT carrying one. This is deliberately not the token response's expires_in field, which describes the access token's lifetime -- a separate token with its own, independently configured expiry -- not the id_token that kubex actually sends to the API server.

A rotated refresh token is kept only on self.config for the life of this process; it is not written back to the source kubeconfig file. Against an IdP configured for refresh-token rotation (e.g. Auth0, Okta, Keycloak with rotation enabled), the on-disk kubeconfig's refresh token is single-use and is invalidated by the first successful refresh, so a fresh process (or a concurrent kubectl using the same kubeconfig) must re-authenticate interactively once this token has been used here.

Source code in kubex/configuration/auth/oidc.py

async def refresh_token(self) -> tuple[str, float | None]:
    """Return a usable id_token, redeeming the refresh token only when needed.

    Like client-go, the configured `id-token` is returned as long as its `exp`
    claim is in the future; otherwise the refresh token is redeemed for a new
    id_token (rotating the in-memory refresh token if the IdP issues a new one).
    Raises `ConfigurationError` when there is neither a valid id-token nor a
    refresh token.

    Returns the id_token and its own `exp` claim (Unix epoch seconds), if it is a
    decodable JWT carrying one. This is deliberately not the token response's
    `expires_in` field, which describes the access token's lifetime -- a
    separate token with its own, independently configured expiry -- not the
    id_token that kubex actually sends to the API server.

    A rotated refresh token is kept only on `self.config` for the life of this
    process; it is not written back to the source kubeconfig file. Against an
    IdP configured for refresh-token rotation (e.g. Auth0, Okta, Keycloak with
    rotation enabled), the on-disk kubeconfig's refresh token is single-use and
    is invalidated by the first successful refresh, so a fresh process (or a
    concurrent `kubectl` using the same kubeconfig) must re-authenticate
    interactively once this token has been used here.
    """
    configured = self._configured_id_token()
    if configured is not None:
        return configured
    if self.config.refresh_token is None:
        raise ConfigurationError(
            "OIDC: no valid id-token (missing, expired or without an exp claim) "
            "and no refresh-token to obtain a new one; log in again to refresh "
            "the kubeconfig credentials"
        )
    content = await self._id_token()
    data = json.loads(content)
    id_token = data.get("id_token")
    if not isinstance(id_token, str):
        raise ConfigurationValueError("OIDC: id_token not found in token response")
    self.config.id_token = SecretStr(id_token)
    rotated_refresh_token = data.get("refresh_token")
    if isinstance(rotated_refresh_token, str) and rotated_refresh_token:
        self.config.refresh_token = SecretStr(rotated_refresh_token)
    return id_token, _jwt_exp_claim(id_token)

Refreshable token

kubex.configuration.auth.refreshable_token

BaseRefreshableToken

BaseRefreshableToken()

Bases: ABC

Source code in kubex/configuration/auth/refreshable_token.py

def __init__(self) -> None:
    self._lock = anyio.Lock()
    self._last_read_token: SecretStr | None = None
    self._expires_at: float = anyio.current_time()
    self._expires_at_wall: datetime = datetime.now(timezone.utc)