load_balancer.py

import base64
from urllib.request import urlopen

import pulumi
import pulumi_aws as aws
import pulumi_kubernetes as k8s
import pulumi_tls as tls

# This is an internal module that exposes a EKS cluster at `cluster`.
import eks

def eks_role_policy(oidc_provider, namespace, service_account):
    return Output.from_input(
        {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Principal": {
                        "Federated": oidc_provider.arn,
                    },
                    "Action": "sts:AssumeRoleWithWebIdentity",
                    "Condition": {
                        "StringEquals": {
                            Output.concat(oidc_provider.url, ":sub"): Output.concat(
                                "system:serviceaccount:",
                                namespace,
                                ":",
                                service_account,
                            )
                        }
                    },
                }
            ],
        }
    ).apply(json.dumps)

aws_config = pulumi.Config("aws")

base_name = f"{pulumi.get_project()}-{pulumi.get_stack()}-aws-load-balancer-controller"

policy_url = "https://raw.githubusercontent.com/kubernetes-sigs/aws-load-balancer-controller/v2.1.3/docs/install/iam_policy.json"

policy = aws.iam.Policy(base_name, policy=urlopen(policy_url).read().decode("utf8"))

role = aws.iam.Role(
    base_name,
    assume_role_policy=eks_role_policy(
        eks.cluster.core.oidc_provider, "default", "aws-load-balancer-controller"
    ),
)

aws.iam.RolePolicyAttachment(base_name, policy_arn=policy.arn, role=role)

ca_key = tls.PrivateKey(
    "aws-load-balancer-controller-ca-key",
    algorithm="RSA",
)

ca = tls.SelfSignedCert(
    "aws-load-balancer-controller-ca",
    subjects=[
        tls.SelfSignedCertSubjectArgs(common_name="aws-load-balancer-controller-ca")
    ],
    key_algorithm="RSA",
    private_key_pem=ca_key.private_key_pem,
    allowed_uses=["key_encipherment", "digital_signature", "cert_signing"],
    validity_period_hours=10 * 365 * 24,
    is_ca_certificate=True,
)

cert_key = tls.PrivateKey(
    "aws-load-balancer-controller-cert-key",
    algorithm="RSA",
)

cert_request = tls.CertRequest(
    "aws-load-balancer-controller-cert-request",
    key_algorithm="RSA",
    private_key_pem=cert_key.private_key_pem,
    subjects=[
        tls.SelfSignedCertSubjectArgs(common_name="aws-load-balancer-controller")
    ],
    dns_names=[
        "aws-load-balancer-webhook-service.default",
        "aws-load-balancer-webhook-service.default.svc",
    ],
)

cert = tls.LocallySignedCert(
    "aws-load-balancer-controller-cert",
    ca_key_algorithm="RSA",
    ca_cert_pem=ca.cert_pem,
    ca_private_key_pem=ca_key.private_key_pem,
    cert_request_pem=cert_request.cert_request_pem,
    allowed_uses=["key_encipherment", "digital_signature"],
    validity_period_hours=10 * 365 * 24,
)

ca_cert_encoded = ca.cert_pem.apply(lambda s: base64.b64encode(s.encode()).decode())
cert_encoded = cert.cert_pem.apply(lambda s: base64.b64encode(s.encode()).decode())
cert_key_encoded = cert_key.private_key_pem.apply(
    lambda s: base64.b64encode(s.encode()).decode()
)


def fix_chart(args, opts):
    if args["kind"] == "CustomResourceDefinition":
        # The chart has an errant `status` field in its CRD.
        # https://github.com/pulumi/pulumi-kubernetes/issues/800
        del args["status"]
    elif args["kind"] == "Secret":
        # The Helm chart generates new certificates on every apply.
        # To workaround, we generate our own certificates with Pulumi and inject
        # them here. We may one day be able to inject them via the `values`
        # map instead.
        # See: https://github.com/aws/eks-charts/issues/347
        args["data"]["ca.crt"] = ca_cert_encoded
        args["data"]["tls.crt"] = cert_encoded
        args["data"]["tls.key"] = cert_key_encoded
    elif args["kind"] in [
        "MutatingWebhookConfiguration",
        "ValidatingWebhookConfiguration",
    ]:
        # Ditto.
        for webhook in args["webhooks"]:
            webhook["clientConfig"]["caBundle"] = ca_cert_encoded


k8s.helm.v3.Chart(
    "aws-load-balancer-controller",
    k8s.helm.v3.ChartOpts(
        chart="aws-load-balancer-controller",
        version="1.1.6",
        namespace="default",
        fetch_opts=k8s.helm.v3.FetchOpts(repo="https://aws.github.io/eks-charts"),
        values={
            "clusterName": eks.cluster.name,
            "serviceAccount": {
                "annotations": {
                    "eks.amazonaws.com/role-arn": role.arn,
                }
            },
        },
        transformations=[fix_chart],
    ),
    opts=pulumi.ResourceOptions(provider=eks.cluster.provider),
)