使用certbot获取Let's Encrypt通配符证书并更新到Openshift集群路由

原理

当我们使用 certbot 申请通配符证书时，需要手动添加 TXT 记录。每个 certbot 申请的证书有效期为 3 个月，虽然 certbot 提供了自动续期命令，但是当我们把自动续期命令配置为定时任务时，我们无法手动添加新的 TXT 记录用于 certbot 验证。

好在 certbot 提供了一个 hook，可以编写一个 Shell 脚本。在续期的时候让脚本调用 DNS 服务商的 API 接口动态添加 TXT 记录，验证完成后再删除此记录。

参考Github项目certbot-dns-aliyun

1、准备工作

一个阿里云的域名，要求备案
阿里云的ACCESS_KEY_ID和ACCESS_KEY_SECRET并保证有添加域名解析的权限
一台linix服务器

2、certbot安装

sudo apt update
sudo apt install certbot

3、通过aliyun-dns验证并生成证书

1、安装 aliyun cli 工具

wget https://aliyuncli.alicdn.com/aliyun-cli-linux-latest-amd64.tgz
tar xzvf aliyun-cli-linux-latest-amd64.tgz
sudo cp aliyun /usr/local/bin
rm aliyun

2、安装 certbot-dns-aliyun 插件

wget https://cdn.jsdelivr.net/gh/justjavac/certbot-dns-aliyun@main/alidns.sh
sudo cp alidns.sh /usr/local/bin
sudo chmod +x /usr/local/bin/alidns.sh
sudo ln -s /usr/local/bin/alidns.sh /usr/local/bin/alidns
rm alidns.sh

3、申请证书

测试是否能正确申请：

certbot certonly -d *.example.com --manual --preferred-challenges dns --manual-auth-hook "alidns" --manual-cleanup-hook "alidns clean" --dry-run

正式申请时去掉 --dry-run 参数：

certbot certonly -d *.example.com --manual --preferred-challenges dns --manual-auth-hook "alidns" --manual-cleanup-hook "alidns clean"

4、证书续期

certbot renew --manual --preferred-challenges dns --manual-auth-hook "alidns" --manual-cleanup-hook "alidns clean" --dry-run

如果以上命令没有错误，把 --dry-run 参数去掉。

5、自动续期

0 1 0 0 0 root certbot renew --manual --preferred-challenges dns --manual-auth-hook "alidns" --manual-cleanup-hook "alidns clean"

证书会在到期前30天自动续期，续期的原理是对证书重新颁发，私钥会变，所以证书续期后还要在集群的路由中重新更新

4、集群内批量更新路由证书(Python脚本)

import os
import subprocess
import base64
import json
import shutil

# 配置变量
CERT_PATH = "/etc/letsencrypt/live/example.com/"


def check_file(cert_path, key_path, ca_path):
    # 检查证书和私钥文件是否存在
    if not os.path.isfile(cert_path):
        print(f"证书文件不存在: {cert_path}")
        exit(1)

    if not os.path.isfile(key_path):
        print(f"私钥文件不存在: {key_path}")
        exit(1)
        
    if not os.path.isfile(ca_path):
        print(f"CA证书文件不存在: {ca_path}")
        exit(1)


def get_cert(cert_path):
    # 读取证书内容
    with open(cert_path, 'r') as cert_file:
        cert_content = cert_file.read()
    return cert_content


def get_key(key_path):
    # 读取私钥内容
    with open(key_path, 'r') as key_file:
        key_content = key_file.read()
    return key_content


def get_ca(ca_path):
    # 读取CA证书内容
    with open(ca_path, 'r') as ca_file:
        ca_content = ca_file.read()
    return ca_content


def update(routes, namespace="default"):
    cert_path = f"{CERT_PATH}/cert.pem" # 证书
    key_path = f"{CERT_PATH}/privkey.pem" # 私钥
    ca_path = f"{CERT_PATH}/chain.pem" # CA
    check_file(cert_path, key_path, ca_path)

    cert = get_cert(cert_path).replace("\n", "\\n")
    key = get_key(key_path).replace("\n", "\\n")
    ca = get_ca(ca_path).replace("\n", "\\n")

    # 读取 Route 列表并更新证书
    for route in routes:
        # 构建 patch 命令
        patch_command = [
            "oc", "patch", "route", route, "-n", namespace,
            "--type=json", 
            "-p", f'['
                f'{{"op": "replace", "path": "/spec/tls/certificate", "value": "{cert}"}}, '
                f'{{"op": "replace", "path": "/spec/tls/key", "value": "{key}"}}, '
                f'{{"op": "replace", "path": "/spec/tls/caCertificate", "value": "{ca}"}}'
                f']'
        ]

        # 执行命令
        try:
            subprocess.run(patch_command, check=True)
        except subprocess.CalledProcessError as e:
            print(f"更新 Route: {route} 失败: {e}\n错误信息: {e.stderr}")


def get_route_names(namespace="default"):
    try:
        # 调用 oc 命令获取所有路由
        result = subprocess.run(
            ["oc", "get", "routes", "-n", namespace, "-o", "json"],
            capture_output=True,
            text=True,
            check=True
        )
        
        # 解析 JSON 输出
        routes = json.loads(result.stdout)
        
        route_names = [item['metadata']['name'] for item in routes['items']]
        
        return route_names

    except subprocess.CalledProcessError as e:
        print(f"Error executing oc command: {e}")
        return []



if __name__ == "__main__":
    # 集群路由
    ocp_routes = get_route_names()
    update("ocp", routes=ocp_routes)