AWS IAM RDS

RdsIamProvider and fetch_rds_iam_token for AWS IAM RDS Postgres authentication, backed by boto3. Part of the taskq[aws] extra. See the Managed Identities guide for prerequisites and a full worker example.

aws

AWS IAM database authentication providers for Amazon RDS Postgres.

This module is part of the taskq[aws] optional extra. It provides a :class:~taskq.auth.PgCredentialProvider implementation backed by AWS IAM RDS authentication, plus the raw token fetcher for users building their own providers. Install with::

pip install 'taskq-py[aws]'

Usage

::

from taskq.auth import make_pg_pool_factory
from taskq.aws import RdsIamProvider

provider = RdsIamProvider(settings.pg_dsn_direct, region="us-east-1")

WorkerConnections(
    dispatcher_pool_factory=make_pg_pool_factory(
        settings.pg_dsn_direct, provider, max_size=settings.dispatcher_pool_size,
    ),
)

How AWS IAM RDS auth works

AWS RDS Postgres supports IAM database authentication: instead of a static password, you request a SigV4-signed auth token from the RDS API (generate_db_auth_token) and use it as the Postgres password. The token is valid for 15 minutes (:data:RDS_TOKEN_LIFETIME_SECONDS), so each call to a factory fetches a fresh token. For long-lived workers, send SIGHUP to the worker process on a schedule shorter than 15 minutes — the factory is re-invoked automatically to rebuild the pool with a fresh token (see taskq.worker.deps.reload_credentials); no restart needed.

boto3 is synchronous. generate_db_auth_token itself is local SigV4 signing, but resolving the ambient credential chain (get_frozen_credentials) may perform blocking STS/IMDS HTTPS calls when credentials are near expiry — so the provider offloads the call to a thread rather than stalling the event loop. This module never imports boto3 at module top level — the import is deferred so import taskq.aws is safe without the extra installed.

Prerequisites

• Enable IAM database authentication on the RDS instance.
• Create a database user mapped to an IAM principal (CREATE USER myiamuser; GRANT rds_iam TO myiamuser;).
• Grant the IAM principal (user/role) permission to call rds-db:connect via an IAM policy.
• The DSN's user must be the IAM-mapped database user.
• sslmode=require is enforced by :func:~taskq.auth.enrich_pg_dsn.

__all__ module-attribute

__all__ = [
    "RDS_TOKEN_LIFETIME_SECONDS",
    "RdsIamProvider",
    "fetch_rds_iam_token",
]

RDS_TOKEN_LIFETIME_SECONDS module-attribute

RDS_TOKEN_LIFETIME_SECONDS = 900

RdsIamProvider

RdsIamProvider(
    dsn: str,
    *,
    region: str | None = None,
    client: Any | None = None,
    username: str | None = None,
)

Bases: PgCredentialProvider

:class:~taskq.auth.PgCredentialProvider backed by AWS IAM RDS auth.

Returns the IAM auth token as the Postgres password; the DSN's existing user (the IAM-mapped DB user) is preserved.

client defaults to a boto3.client('rds') from the ambient credential chain; pass region to pin it. username defaults to the DSN's userinfo user.

Source code in src/taskq/aws.py

def __init__(
    self,
    dsn: str,
    *,
    region: str | None = None,
    client: Any | None = None,
    username: str | None = None,
) -> None:
    hostname, port, parsed_username = _parse_dsn(dsn)
    self._hostname = hostname
    self._port = port
    self._username = username if username is not None else parsed_username
    if not self._username:
        # Deliberately no DSN in the message — it may carry a userinfo
        # password, which must not land in tracebacks / log aggregation.
        raise ValueError(
            "DSN has no username; AWS IAM RDS auth requires the "
            "IAM-mapped database user in the DSN userinfo (or pass username=)."
        )
    self._region = region
    self._client = client

get_pg_credential async

get_pg_credential() -> PgCredential

Source code in src/taskq/aws.py

async def get_pg_credential(self) -> PgCredential:
    # Offloaded to a thread: although generate_db_auth_token is local
    # SigV4 signing, the ambient credential chain may perform blocking
    # STS/IMDS HTTPS refreshes, which must not stall the event loop.
    token = await asyncio.to_thread(
        fetch_rds_iam_token,
        hostname=self._hostname,
        port=self._port,
        username=self._username,
        region=self._region,
        client=self._client,
    )
    return PgCredential(password=token)

fetch_rds_iam_token

fetch_rds_iam_token(
    *,
    hostname: str,
    port: int,
    username: str,
    region: str | None = None,
    client: Any | None = None,
) -> str

Fetch an AWS RDS IAM auth token for use as the Postgres password.

client defaults to a boto3.client('rds') built with the ambient AWS credential chain (env vars, instance role, etc.). Pass region to pin the region when the client is not supplied.

The token is a SigV4-signed URL valid for 15 minutes (:data:RDS_TOKEN_LIFETIME_SECONDS). region is passed through to botocore untouched — None lets botocore fall back to the client's ambient region (an empty string would produce a signature scoped to date//rds-db/aws4_request, which RDS rejects).

This is a synchronous function and may perform blocking network I/O (STS/IMDS credential refresh); async callers should offload it to a thread, as :class:RdsIamProvider does.

Source code in src/taskq/aws.py

def fetch_rds_iam_token(
    *,
    hostname: str,
    port: int,
    username: str,
    region: str | None = None,
    client: Any | None = None,
) -> str:
    """Fetch an AWS RDS IAM auth token for use as the Postgres password.

    ``client`` defaults to a ``boto3.client('rds')`` built with the
    ambient AWS credential chain (env vars, instance role, etc.). Pass
    ``region`` to pin the region when the client is not supplied.

    The token is a SigV4-signed URL valid for 15 minutes
    (:data:`RDS_TOKEN_LIFETIME_SECONDS`). ``region`` is passed through to
    botocore untouched — ``None`` lets botocore fall back to the client's
    ambient region (an empty string would produce a signature scoped to
    ``date//rds-db/aws4_request``, which RDS rejects).

    This is a synchronous function and may perform blocking network I/O
    (STS/IMDS credential refresh); async callers should offload it to a
    thread, as :class:`RdsIamProvider` does.
    """
    if client is None:
        boto3 = _require_boto3()
        client_kwargs: dict[str, Any] = {}
        if region is not None:
            client_kwargs["region_name"] = region
        resolved = boto3.client("rds", **client_kwargs)
    else:
        resolved = client
    return resolved.generate_db_auth_token(
        DBHostname=hostname,
        Port=port,
        DBUsername=username,
        Region=region,
    )