Authentication¶

To access a Kubernetes cluster, an endpoint and some credentials are needed. They are usually taken either from the environment (environment variables), or from the ~/.kube/config file, or from external authentication services.

Kopf provides rudimentary authentication out of the box: it can authenticate with the Kubernetes API either via the service account or raw kubeconfig data (with no additional interpretation or parsing of those).

But this may not be enough in some setups and environments. Kopf does not attempt to maintain all the authentication methods possible. Instead, it allows the operator developers to implement their custom authentication methods and “piggybacks” the existing Kubernetes clients.

The latter ones can implement some advanced authentication techniques, such as the temporary token retrieval via the authentication services, token rotation, etc.

Custom authentication¶

In most setups, the normal authentication from one of the API client libraries is enough — it works out of the box if those clients are installed (see Piggybacking below). Custom authentication is only needed if the normal authentication methods do not work for some reason, such as if you have a specific and unusual cluster setup (e.g. your own auth tokens).

To implement a custom authentication method, one or a few login-handlers can be added. The login handlers should either return nothing (None) or an instance of kopf.ConnectionInfo:

import datetime
import kopf
from typing import Any

@kopf.on.login()
def login_fn(**_: Any) -> kopf.ConnectionInfo | None:
    return kopf.ConnectionInfo(
        server='https://localhost',
        proxy_url='http://localhost:8080',
        ca_path='/etc/ssl/ca.crt',
        ca_data=b'...',
        insecure=True,
        username='...',
        password='...',
        scheme='Bearer',
        token='...',
        certificate_path='~/.minikube/client.crt',
        private_key_path='~/.minikube/client.key',
        certificate_data=b'...',
        private_key_data=b'...',
        trust_env=True,
        expiration=datetime.datetime(2099, 12, 31, 23, 59, 59),
    )

Both TZ-naive and TZ-aware expiration times are supported. The TZ-naive timestamps are always treated as UTC.

As with any other handlers, the login handler can be async if the network communication is needed and async mode is supported:

import kopf
from typing import Any

@kopf.on.login()
async def login_fn(**_: Any) -> kopf.ConnectionInfo | None:
    pass

A kopf.ConnectionInfo is a container to bring the parameters necessary for making the API calls, but not the ways of retrieving them. Specifically:

TCP server host & port.
SSL verification/ignorance flag.
SSL certificate authority.
SSL client certificate and its private key.
HTTP Authorization: Basic username:password.
HTTP Authorization: Bearer token (or other schemes: Bearer, Digest, etc).
URL’s default namespace for cases where this is implied.
HTTP/HTTPS proxy url, possibly with credentials.

Note

Proxy support is limited to what aiohttp supports. Specifically, it supports plain HTTP proxies, some limited HTTPS proxies, but not SOCKS5. If you need more sophisticated proxying or tunneling, implement it as a custom HTTP session with a custom connector, for example using aiohttp_socks (Kopf claims no responsibility for the quality of this library; do your own due diligence).

No matter how the endpoints or credentials are retrieved, they are directly mapped to TCP/SSL/HTTPS protocols in the API clients. It is the responsibility of the authentication handlers to ensure that the values are consistent and valid (e.g. via internal verification calls). It is theoretically possible to mix all authentication methods at once or to have none at all. If the credentials are inconsistent or invalid, permanent re-authentication will occur.

Multiple handlers can be declared to retrieve different credentials or the same credentials via different libraries. All of the retrieved credentials will be used in random order with no specific priority.

The connection info does not respect environment variables by default, such as HTTP_PROXY, HTTPS_PROXY, NO_PROXY, or ~/.netrc. The easiest way to enable this is to set settings.networking.trust_env = True in a startup handler (see Configuration). The built-in login handlers will propagate this setting to kopf.ConnectionInfo automatically.

Custom login handlers can set trust_env=True directly on the returned kopf.ConnectionInfo (see example above). As a last resort, advanced users can provide a custom aiohttp session with trust_env=True (see examples below).

Custom HTTP sessions¶

Advanced users can provide their own aiohttp client session instead of the pre-built one by returning an instance of kopf.AiohttpSession from the login handlers.

You can, for example, inject extra headers, remove or replace the existing ones, add sophisticated authentication schemes, or set the networking parameters, such as timeouts or connection limits (for client-side rate-limiting).

However, if you provide a custom session, you must configure the authentication yourself. This includes the username/password, tokens, or SSL certificates. Kopf will not modify the provided session (except for injecting User-Agent), and cannot do so: most of these fields are either hidden by aiohttp, or are read-only, so they can only be set at the session creation.

For your convenience, kopf.ConnectionInfo from the existing login functions —see Piggybacking— provides the methods to convert it to the typical components of the HTTP sessions:

kopf.ConnectionInfo.as_aiohttp_basic_auth() for username/password.
kopf.ConnectionInfo.as_http_headers() for all tokens.
kopf.ConnectionInfo.as_ssl_context() for CA & SSL client certificates.

Note

kopf.ConnectionInfo.as_ssl_context() will store the certificate and private key data blobs to the disk files temporarily for a brief time, since Python’s ssl can only load it from files, not from data blobs. It will delete the files as soon as the SSL context is constructed.

You do not need to worry about the session termination or closing — Kopf will own and manage the provided session and will close it when needed.

import aiohttp
import kopf
from typing import Any

@kopf.on.login()
async def login_fn(**_: Any) -> kopf.AiohttpSession:
    credentials = kopf.login_with_kubeconfig()  # or any other available method
    headers = {
        'X-Custom-Header': 'helloworld',
        'Authorization': 'VeryAdvancedAuthScheme xyz',
        'User-Agent': f'myoperator/1.2.3 kopf/{kopf.__version__}',
    }
    session = aiohttp.ClientSession(
        connector=aiohttp.TCPConnector(
            limit_per_host=10, limit=20,
            keepalive_timeout=30.
            ssl=credentials.as_ssl_context(),
        ),
        headers=credentials.as_http_headers() | headers,
        auth=credentials.as_aiohttp_basic_auth(),
        trust_env=True,  # respect HTTP_PROXY, HTTPS_PROXY, NO_PROXY, and ~/.netrc
    )
    return kopf.AiohttpSession(
        aiohttp_session=session,
        server=credentials.server,
        default_namespace=credentials.default_namespace,
    )

Warning

As a rule, aiohttp is the internal detail of the implementation and is normally not exposed to users except for very advanced use-cases. Kopf reserves the right to change its internal HTTP library without warning or backwards compatibility. In that case, Kopf will fail if this class is returned (will raise an exception) — to prevent the unnoticed accidental damage during such upgrades. Use at your own risk.

Piggybacking¶

In case no handlers are explicitly declared, Kopf attempts to authenticate with the existing Kubernetes libraries if they are installed. At the moment: pykube-ng and kubernetes. In the future, more libraries can be added for authentication piggybacking.

Note

Since kopf>=1.29, pykube-ng is not pre-installed implicitly. If needed, install it explicitly as a dependency of the operator, or via kopf[full-auth] (see Installation).

Piggybacking means that the config parsing and authentication methods of these libraries are used, and only the information needed for API calls is extracted.

If a few of the piggybacked libraries are installed, all of them will be attempted (as if multiple handlers are installed), and all the credentials will be utilised in random order.

If that is not the desired case, and only one of the libraries is needed, declare a custom login handler explicitly, and use only the preferred library by calling one of the piggybacking functions:

import kopf
from typing import Any

@kopf.on.login()
def login_fn(**kwargs: Any) -> kopf.ConnectionInfo | None:
    return kopf.login_via_pykube(**kwargs)

Or:

import kopf
from typing import Any

@kopf.on.login()
def login_fn(**kwargs: Any) -> kopf.ConnectionInfo | None:
    return kopf.login_via_client(**kwargs)

The same trick is also useful to limit the authentication attempts by time or by number of retries (by default, it tries forever until it succeeds, returns nothing, or explicitly fails):

import kopf
from typing import Any

@kopf.on.login(retries=3)
def login_fn(**kwargs: Any) -> kopf.ConnectionInfo | None:
    return kopf.login_via_pykube(**kwargs)

Similarly, if the libraries are installed and needed, but their credentials are not desired, the rudimentary login functions can be used directly:

import kopf
from typing import Any

@kopf.on.login()
def login_fn(**kwargs: Any) -> kopf.ConnectionInfo | None:
    return kopf.login_with_service_account(**kwargs) or kopf.login_with_kubeconfig(**kwargs)

Credentials lifecycle¶

Internally, all the credentials are gathered from all the active handlers (either the declared ones or all the fallback piggybacking ones) in no particular order, and are fed into a vault.

The Kubernetes API calls then use random credentials from that vault. The credentials that have reached their expiration are ignored and removed. If the API call fails with an HTTP 401 error, these credentials are marked invalid, excluded from further use, and the next random credentials are tried.

When the vault is fully depleted, it freezes all the API calls and triggers the login handlers for re-authentication. Only the new credentials are used. The credentials, which previously were known to be invalid, are ignored to prevent a permanent never-ending re-authentication loop.

There is no validation of credentials by making fake API calls. Instead, the real API calls validate the credentials by using them and reporting them back to the vault as invalid (or keeping them as valid), potentially causing new re-authentication activities.

In case the vault is depleted and no new credentials are provided by the login handlers, the API calls fail, and so does the operator.

This internal logic is hidden from the operator developers, but it is worth knowing how it works internally. See Vault.

If re-authentication is expected to happen frequently (e.g. every few minutes), you might want to disable the logging of re-authentication (whether this is a good idea or not is for you to decide based on the specifics of your system):

import kopf
import logging
from typing import Any

@kopf.on.startup()
def disable_auth_logs(**_: Any) -> None:
    logging.getLogger('kopf.activities.authentication').disabled = True
    logging.getLogger('kopf._core.engines.activities').disabled = True