Managing Kubernetes certificates with Python

I run into a small stumbling block the other evening while working on my ‘site domain manager’ project (for want of a better name). This is essentially a REST API running in a daemon service that manages the mappings of domains to websites, and uses ‘agents’ to automate the configuration via API calls to the various services involved (domain registrars, DNS servers, WAF providers, SSL certs etc.)

The problem arose because the manager class I was writing to interact with kubernetes needed to manage certificates. The kubernetes python client library has a whole bunch of useful higher-level API and model classes for listing, creating, updating the main models I needed to manage, such as the ConfigMap, DaemonSet, Service and Ingress, but because the Certificate is part of the cert-manager package, it doesn’t have the equivalent higher-level methods I needed.

In the end, we solved it by using the lower-level call_api method, as follows:

        siteid = sitename.split(".")[0]
        rawyaml = """
apiVersion: certmanager.k8s.io/v1alpha1
kind: Certificate
metadata:
  name: {{siteid}}-cert
  namespace: {{namespace}}
spec:
  acme:
    config:
    - dns01:
        provider: route53
      domains:
      - {{mainhostname}}
  commonName: {{mainhostname}}
  dnsNames:
  - {{mainhostname}}
  issuerRef:
    kind: ClusterIssuer
    name: letsencrypt-production
  secretName: {{siteid}}-tls
"""
        rawyaml = rawyaml.replace("{{siteid}}", siteid)
        rawyaml = rawyaml.replace("{{namespace}}", self.namespace)
        rawyaml = rawyaml.replace("{{mainhostname}}", mainhostname)
        model = yaml.load(rawyaml, Loader=yaml.SafeLoader)
        model['spec']['acme']['config'][0]['domains'] = aliases
        model['spec']['dnsNames'] = aliases

        path_params = {}
        auth_settings = ['BearerToken']
        header_params = {
            "Content-Type": "application/json"
        }
        query_params = []

        # Fetch latest state of resource to apply changes to
        response = self.api_client.call_api(f"/apis/certmanager.k8s.io/v1alpha1/namespaces/{self.namespace}/certificates", 'GET',
            path_params,
            query_params,
            header_params,
            auth_settings=auth_settings,
        )
        r = json.loads(self.api_client.last_response.data)
        certs = {x['metadata']['name']:x for x in r['items']}

        # If it doesn't exist, create it...
        certid = f"{siteid}-cert"
        try:
            if certid not in certs.keys():
                _logger.info(f"Creating certificate '{certid}' with {len(aliases)} hostnames...")
                response = self.api_client.call_api(f"/apis/certmanager.k8s.io/v1alpha1/namespaces/{self.namespace}/certificates", 'POST',
                    path_params,
                    query_params,
                    header_params,
                    body=model,
                    auth_settings=auth_settings,
                )
            else:
                _logger.info(f"Updating certificate '{certid}' with {len(aliases)} hostnames...")
                # Transplant metadata to allow update to work
                model['metadata'] = certs[certid]['metadata']

                # Attempt to update...
                response = self.api_client.call_api(f"/apis/certmanager.k8s.io/v1alpha1/namespaces/{self.namespace}/certificates/{certid}", 'PUT',
                    path_params,
                    query_params,
                    header_params,
                    body=model,
                    auth_settings=auth_settings,
                )
        except Exception as e:
            _logger.exception(e)
            return str(e)

You can see this in context here.