🔐 Automatiser la gestion des certificats Let’s Encrypt de Kubernetes vers Apache avec cert-manager

📅 PubliĂ© en avril 2025
📌 Mots-clĂ©s SEO : cert-manager, Let’s Encrypt, Apache, Kubernetes, certificat SSL automatique, reverse proxy, TLS, automatisation

🔎 Introduction

La sĂ©curitĂ© des services web repose en grande partie sur l’usage de certificats TLS/SSL. Dans Kubernetes, cert-manager permet de gĂ©nĂ©rer et renouveler des certificats automatiquement via Let’s Encrypt.

Mais lorsque l’on utilise un reverse proxy Apache en dehors du cluster, il est nĂ©cessaire de rĂ©cupĂ©rer et d’intĂ©grer ces certificats dans la configuration Apache. Cet article explique comment automatiser cette synchronisation grĂące Ă  un script Python personnalisable.

🎯 Objectifs

  • ✅ RĂ©cupĂ©rer les certificats depuis Kubernetes via l’API
  • ✅ VĂ©rifier leur expiration pour limiter les traitements inutiles
  • ✅ GĂ©nĂ©rer dynamiquement des VirtualHosts Apache avec support HTTPS
  • ✅ Recharger Apache uniquement en cas de changement dĂ©tectĂ©

1ïžâƒŁ Donner accĂšs Ă  l’API Kubernetes

🔧 Manifeste Kubernetes à appliquer (certificates.yaml)

apiVersion: v1
kind: ServiceAccount
metadata:
  name: cert-reader-sa
  namespace: your-namespace
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: cert-reader-role
  namespace: your-namespace
rules:
- apiGroups: ["cert-manager.io"]
  resources: ["certificates"]
  verbs: ["get", "list"]
- apiGroups: [""]
  resources: ["secrets"]
  verbs: ["get"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: cert-reader-binding
  namespace: your-namespace
subjects:
- kind: ServiceAccount
  name: cert-reader-sa
roleRef:
  kind: Role
  name: cert-reader-role
  apiGroup: rbac.authorization.k8s.io
---
apiVersion: v1
kind: Secret
metadata:
  name: cert-reader-token
  annotations:
    kubernetes.io/service-account.name: "cert-reader-sa"
type: kubernetes.io/service-account-token

Remplace your-namespace par le nom réel du namespace contenant tes certificats.

🔐 Commandes pour rĂ©cupĂ©rer les accĂšs

# RĂ©cupĂ©rer l’URL de l’API Kubernetes
kubectl config view --minify -o jsonpath='{.clusters[0].cluster.server}'

# RĂ©cupĂ©rer le token d’accĂšs
kubectl get secret cert-reader-token -n your-namespace -o jsonpath='{.data.token}' | base64 --decode

2ïžâƒŁ Principe de fonctionnement du script Python

Le script sync_apache_certs_vhosts.py :

  • RĂ©cupĂšre les objets Certificate depuis Kubernetes
  • RĂ©cupĂšre les Secrets TLS associĂ©s
  • VĂ©rifie si les certificats existants sur disque expirent bientĂŽt
  • GĂ©nĂšre automatiquement les VirtualHosts Apache
  • DĂ©commente le bloc HTTPS une fois le certificat valide
  • Recharge Apache si nĂ©cessaire

Il dĂ©tecte Ă©galement l’annotation personnalisĂ©e suivante pour dĂ©finir dynamiquement le backend d’un reverse proxy :

annotations:
  apache-proxy-backend: "https://backend.monapp.com/"

đŸ› ïž Exemple de certificat Kubernetes

apiVersion: cert-manager.io/v1
kind: Certificate
metadata:
  name: my-cert
  namespace: your-namespace
  annotations:
    apache-proxy-backend: "https://backend.monapp.com/"
spec:
  dnsNames:
    - monapp.mondomaine.com
  issuerRef:
    name: letsencrypt-prod
    kind: ClusterIssuer
  secretName: my-cert-secret

⚙ Exemple de VirtualHost Apache gĂ©nĂ©rĂ©

<VirtualHost *:80>
    ServerName monapp.mondomaine.com

    <Location /.well-known/acme-challenge/ >
        ProxyPreserveHost On
        RequestHeader set Host "monapp.mondomaine.com"
        ProxyPass http://ingress-k8s/.well-known/acme-challenge/
        ProxyPassReverse http://ingress-k8s/.well-known/acme-challenge/
    </Location>

    RewriteEngine On
    RewriteCond %{REQUEST_URI} !^/.well-known/acme-challenge/ [NC]
    RewriteRule ^(.*)$ https://%{HTTP_HOST}$1 [R=301,L]
</VirtualHost>

<VirtualHost *:443>
    ServerName monapp.mondomaine.com

    SSLEngine on
    SSLCertificateFile /etc/apache2/certs/monapp_mondomaine_com/monapp_mondomaine_com.crt
    SSLCertificateKeyFile /etc/apache2/certs/monapp_mondomaine_com/monapp_mondomaine_com.key

    ProxyPreserveHost On
    ProxyPass / https://backend.monapp.com/
    ProxyPassReverse / https://backend.monapp.com/

    SSLProxyEngine On
    SSLProxyVerify none
    SSLProxyCheckPeerCN off
    SSLProxyCheckPeerName off
</VirtualHost>

🔁 Automatiser l’exĂ©cution avec cron

Ajoute une tùche cron pour exécuter le script chaque heure :

echo '0 * * * * root /usr/bin/python3 /usr/local/bin/sync_apache_certs_vhosts.py' > /etc/cron.d/sync_apache_certs_vhosts
systemctl restart cron

VĂ©rifie l’exĂ©cution via les logs :

grep sync_apache_certs_vhosts /var/log/syslog

✅ RĂ©sumĂ©

  • 🔐 Certificats Let’s Encrypt gĂ©rĂ©s automatiquement via cert-manager
  • 🔄 IntĂ©gration continue avec Apache sans intervention manuelle
  • đŸ›Ąïž VirtualHosts HTTPS configurĂ©s et activĂ©s dynamiquement
  • ⚙ Apache rechargĂ© uniquement si une modification est dĂ©tectĂ©e

📎 Annexe : Script Python complet sync_apache_certs_vhosts.py

#!/usr/bin/env python3

import os
import json
import base64
import requests
import subprocess
import re
from datetime import datetime, timedelta

# Variables Ă  adapter
KUBE_API_SERVER = "https://your-k8s-api-server:6443"
KUBE_TOKEN = "TON_TOKEN_K8S_ICI"
NAMESPACE = "your-namespace"

CERT_BASE_DIR = "/etc/apache2/certs/"
VHOST_DIR = "/etc/apache2/sites-available/"

APACHE_TEST_CMD = "apachectl configtest"
APACHE_RELOAD_CMD = "systemctl reload apache2"
APACHE_ENABLE_SITE_CMD = "a2ensite"

DOMAIN_SUFFIX = ".mondomaine.com"
EXPIRATION_THRESHOLD = 72  # heures
RECENT_THRESHOLD_HOURS = 2

HEADERS = {
    "Authorization": f"Bearer {KUBE_TOKEN}",
    "Accept": "application/json"
}
REQUESTS_VERIFY = False

def format_name(domain):
    return domain.replace(".", "_")
def is_cert_expiring(cert_path):
    if not os.path.exists(cert_path):
        print(f"⚠ Certificat absent ({cert_path}), renouvellement nĂ©cessaire.")
        return True

    try:
        result = subprocess.run(
            ["openssl", "x509", "-enddate", "-noout", "-in", cert_path],
            capture_output=True,
            text=True,
            check=True
        )
        expiry_str = result.stdout.strip().split("=")[-1]
        expiry_date = datetime.strptime(expiry_str, "%b %d %H:%M:%S %Y %Z")
        time_remaining = expiry_date - datetime.utcnow()
        if time_remaining < timedelta(hours=EXPIRATION_THRESHOLD):
            print(f"⚠ Expiration proche : {time_remaining}")
            return True
        else:
            print(f"✅ Certificat valide jusqu’à {expiry_date}")
            return False
    except Exception as e:
        print(f"❌ Erreur vĂ©rif cert : {e}")
        return True

def get_certificates():
    url = f"{KUBE_API_SERVER}/apis/cert-manager.io/v1/namespaces/{NAMESPACE}/certificates"
    response = requests.get(url, headers=HEADERS, verify=REQUESTS_VERIFY)

    if response.status_code != 200:
        print(f"❌ Erreur API : {response.status_code} - {response.text}")
        return []

    certs = response.json().get("items", [])
    filtered = []
    for cert in certs:
        dns_names = cert["spec"].get("dnsNames", [])
        valid_dns = [dns for dns in dns_names if dns.endswith(DOMAIN_SUFFIX) and dns.count('.') == DOMAIN_SUFFIX.count('.')]
        if valid_dns:
            cert["filtered_dns_names"] = valid_dns
            filtered.append(cert)
    return filtered
def extract_cert(secret_name, domain):
    url = f"{KUBE_API_SERVER}/api/v1/namespaces/{NAMESPACE}/secrets/{secret_name}"
    response = requests.get(url, headers=HEADERS, verify=REQUESTS_VERIFY)

    if response.status_code != 200:
        print(f"❌ Erreur rĂ©cupĂ©ration Secret: {response.status_code}")
        return None, None

    secret = response.json()["data"]
    domain_safe = format_name(domain)
    cert_dir = os.path.join(CERT_BASE_DIR, domain_safe)
    cert_path = os.path.join(cert_dir, f"{domain_safe}.crt")
    key_path = os.path.join(cert_dir, f"{domain_safe}.key")

    if not is_cert_expiring(cert_path):
        return None, None

    os.makedirs(cert_dir, exist_ok=True)
    with open(cert_path, "wb") as f:
        f.write(base64.b64decode(secret["tls.crt"]))
    with open(key_path, "wb") as f:
        f.write(base64.b64decode(secret["tls.key"]))

    return cert_path, key_path

def generate_vhost(domain, backend=None, cert_path=None, key_path=None):
    domain_safe = format_name(domain)
    vhost_path = os.path.join(VHOST_DIR, f"{domain_safe}.conf")
    final_cert_path = os.path.join(CERT_BASE_DIR, domain_safe, f"{domain_safe}.crt")
    final_key_path = os.path.join(CERT_BASE_DIR, domain_safe, f"{domain_safe}.key")

    if os.path.exists(vhost_path):
        print(f"⚠ VHost {vhost_path} existe dĂ©jĂ .")
        return None

    with open(vhost_path, "w") as f:
        f.write(f"""<VirtualHost *:80>
    ServerName {domain}
    <Location /.well-known/acme-challenge/ >
        ProxyPreserveHost On
        RequestHeader set Host "{domain}"
        ProxyPass http://ingress/.well-known/acme-challenge/
        ProxyPassReverse http://ingress/.well-known/acme-challenge/
    </Location>
    RewriteEngine On
    RewriteCond %{{REQUEST_URI}} !^/.well-known/acme-challenge/ [NC]
    RewriteRule ^(.*)$ https://%{{HTTP_HOST}}$1 [R=301,L]
</VirtualHost>

# <VirtualHost *:443>
#     ServerName {domain}
#     SSLEngine on
#     SSLCertificateFile {final_cert_path}
#     SSLCertificateKeyFile {final_key_path}
""")
        if backend:
            f.write(f"""#     ProxyPreserveHost On
#     ProxyPass / {backend}
#     ProxyPassReverse / {backend}
""")
            if backend.startswith("https://"):
                f.write("""#     SSLProxyEngine On
#     SSLProxyVerify none
#     SSLProxyCheckPeerCN off
#     SSLProxyCheckPeerName off
""")
        f.write("# </VirtualHost>\n")

    subprocess.run([APACHE_ENABLE_SITE_CMD, f"{domain_safe}.conf"], check=True)
    print(f"✅ VHost gĂ©nĂ©rĂ© et activĂ© : {vhost_path}")
    return vhost_path
def activate_https_in_vhost(vhost_path):
    with open(vhost_path, "r") as f:
        content = f.read()

    pattern = r"(?ms)#\s*<VirtualHost \*:443>.*?#\s*</VirtualHost>"

    def uncomment_block(match):
        block = match.group(0)
        return re.sub(r"(?m)^#\s?", "", block)

    new_content, count = re.subn(pattern, uncomment_block, content)
    if count > 0:
        with open(vhost_path, "w") as f:
            f.write(new_content)
        print(f"✅ HTTPS activĂ© dans {vhost_path}.")
    else:
        print("â„č Aucun bloc HTTPS commentĂ© trouvĂ©.")

def is_cert_recent(cert_path, threshold_hours=RECENT_THRESHOLD_HOURS):
    if not os.path.exists(cert_path):
        return False
    try:
        result = subprocess.run(
            ["openssl", "x509", "-startdate", "-noout", "-in", cert_path],
            capture_output=True,
            text=True,
            check=True
        )
        start_date_str = result.stdout.strip().split("=")[-1]
        start_date = datetime.strptime(start_date_str, "%b %d %H:%M:%S %Y %Z")
        age = datetime.utcnow() - start_date
        return age < timedelta(hours=threshold_hours)
    except Exception as e:
        print(f"❌ Erreur rĂ©cence cert {cert_path}: {e}")
        return False

def test_apache_config():
    result = subprocess.run(APACHE_TEST_CMD, shell=True, capture_output=True, text=True)
    return "Syntax OK" in result.stdout + result.stderr

def reload_apache():
    if test_apache_config():
        subprocess.run(APACHE_RELOAD_CMD, shell=True, check=True)
        print("✅ Apache rechargĂ©.")
    else:
        print("❌ Erreur de config Apache. Pas de reload.")

def main():
    certs = get_certificates()
    if not certs:
        print("⚠ Aucun certificat trouvĂ©.")
        return

    apache_needs_reload = False

    for cert in certs:
        domain = cert["filtered_dns_names"][0]
        secret_name = cert["spec"]["secretName"]
        backend = cert["metadata"].get("annotations", {}).get("apache-proxy-backend")

        cert_ready = any(
            cond["type"] == "Ready" and cond["status"] == "True"
            for cond in cert.get("status", {}).get("conditions", [])
        )

        cert_path, key_path = None, None
        if cert_ready:
            cert_path, key_path = extract_cert(secret_name, domain)
            if cert_path:
                apache_needs_reload = True

        vhost_path = generate_vhost(domain, backend, cert_path, key_path)
        if vhost_path:
            apache_needs_reload = True
            if cert_ready and cert_path and is_cert_recent(cert_path):
                activate_https_in_vhost(vhost_path)
                apache_needs_reload = True

    if apache_needs_reload:
        reload_apache()

if __name__ == "__main__":
    main()
Étiquettes

Publications similaires