如何通过kubernetes-python-client获取Kubernetes集群SSL证书过期时间

前置依赖 #

• 已安装所需Python包：执行 pip install kubernetes cryptography pyyaml 完成安装
• 本地kubeconfig已配置所有4个集群的上下文、凭证信息，网络策略允许客户端直连所有集群的API Server地址

核心实现步骤 #

1. 批量加载多集群独立配置 #

默认客户端的load_kube_config仅会读取当前激活的上下文，我们需要遍历kubeconfig中所有集群上下文，为每个集群生成独立的配置对象，避免全局上下文切换冲突。

import os
import yaml
from kubernetes import client, config
from kubernetes.client import Configuration
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from datetime import datetime, timezone

# 加载本地kubeconfig全量内容
with open(os.path.expanduser("~/.kube/config"), "r") as f:
    kubeconfig = yaml.safe_load(f)

# 遍历所有集群上下文，生成每个集群的独立配置
cluster_configs = {}
for context_entry in kubeconfig["contexts"]:
    context_name = context_entry["name"]
    cluster_config = config.load_kube_config_from_dict(
        config_dict=kubeconfig,
        context=context_name
    )
    cluster_configs[context_name] = cluster_config

2. 获取kube-apiserver服务端证书过期时间 #

无需调用K8s API，直接在建立HTTPS连接时从TLS会话中提取服务端证书解析即可，该方式即使在证书即将过期的状态下也能正常获取信息。

import ssl

def get_apiserver_cert_expiry(config: Configuration) -> datetime:
    # 解析配置中的API Server地址，去掉前缀的https://
    apiserver_url = config.host.replace("https://", "")
    # 拆分主机与端口
    if ":" in apiserver_url:
        host, port = apiserver_url.split(":")
        port = int(port)
    else:
        host = apiserver_url
        port = 443

    # 自定义TLS上下文，跳过证书校验（仅用于读取目标证书，无需验证受信状态）
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE

    # 建立TLS连接获取证书原始DER格式
    with ssl.create_connection((host, port), context=ctx, timeout=10) as sock:
        der_cert = sock.getpeercert(binary_form=True)

    # 解析证书获取过期时间
    cert = x509.load_der_x509_certificate(der_cert, default_backend())
    return cert.not_valid_after_utc

3. 获取集群内所有TLS类型Secret的证书过期时间 #

如果需要检测集群内部署应用的TLS证书（存储为kubernetes.io/tls类型Secret），通过调用K8s API遍历所有命名空间下的Secret，提取证书内容解析即可。

import base64

def get_all_tls_secret_expiry(config: Configuration) -> list[dict]:
    # 基于当前集群配置生成独立API客户端
    api_client = client.ApiClient(config)
    core_v1 = client.CoreV1Api(api_client)

    tls_cert_list = []
    # 遍历所有命名空间下的Secret
    for secret in core_v1.list_secret_for_all_namespaces(timeout_seconds=30).items:
        # 仅处理标准TLS类型的Secret
        if secret.type != "kubernetes.io/tls":
            continue
        # 提取base64编码的证书内容
        tls_crt_b64 = secret.data.get("tls.crt")
        if not tls_crt_b64:
            continue
        # 解码为PEM格式证书
        tls_crt_pem = base64.b64decode(tls_crt_b64)
        # 解析证书过期时间
        cert = x509.load_pem_x509_certificate(tls_crt_pem, default_backend())
        tls_cert_list.append({
            "namespace": secret.metadata.namespace,
            "secret_name": secret.metadata.name,
            "expiry_time": cert.not_valid_after_utc
        })
    return tls_cert_list

4. 批量检测流程 #

串联上述逻辑遍历所有集群，可自定义告警阈值（如剩余30天过期时输出告警）。

if __name__ == "__main__":
    # 配置告警阈值：剩余有效期小于30天时输出告警
    warn_days = 30
    now = datetime.now(timezone.utc)

    for cluster_name, cluster_config in cluster_configs.items():
        print(f"===== 开始检测集群: {cluster_name} =====")
        # 检测API Server证书
        try:
            apiserver_expiry = get_apiserver_cert_expiry(cluster_config)
            days_to_expire = (apiserver_expiry - now).days
            print(f"API Server证书过期时间: {apiserver_expiry}, 剩余有效天数: {days_to_expire}")
            if days_to_expire < warn_days:
                print(f"⚠️  集群{cluster_name}的API Server证书将在{warn_days}天内过期")
        except Exception as e:
            print(f"获取集群{cluster_name}API Server证书失败: {str(e)}")

        # 检测集群内部TLS Secret证书
        try:
            secret_certs = get_all_tls_secret_expiry(cluster_config)
            for cert_info in secret_certs:
                days_to_expire = (cert_info["expiry_time"] - now).days
                print(f"命名空间{cert_info['namespace']}下Secret {cert_info['secret_name']}证书过期时间: {cert_info['expiry_time']}, 剩余有效天数: {days_to_expire}")
                if days_to_expire < warn_days:
                    print(f"⚠️  上述证书将在{warn_days}天内过期")
        except Exception as e:
            print(f"获取集群{cluster_name}内部TLS证书失败: {str(e)}")

注意事项 #

• 权限要求：若要遍历所有命名空间的Secret，对应集群的访问账号需要具备集群级的list secrets权限，否则会返回403错误
• 版本兼容：如果使用的cryptography库版本低于38.0.0，将代码中的not_valid_after_utc替换为not_valid_after.replace(tzinfo=timezone.utc)即可适配
• 多文件kubeconfig：如果你的4个集群配置分散在不同的kubeconfig文件中，可将加载单个文件的逻辑修改为遍历指定目录下所有kubeconfig文件，分别加载对应上下文
• 超时配置：上述代码已配置连接超时参数，避免网络异常时程序长时间阻塞

内容的提问来源于stack exchange，提问作者KeithTt