首页实践项目 Python 通过python脚本实现自动更新腾讯云CDN数字证书功能

通过python脚本实现自动更新腾讯云CDN数字证书功能

2018年10月5日

4318

0x01 前言

我的博客和腾讯云CDN都是用let’s encrypt的数字证书，每次签发的有效期为90天，也就是说每年我至少需要手动更换4次数字证书。

这里有个大问题，我都懒得出汁了，监控数字证书的有效期都要依靠腾讯云的服务，每次需要更新我都是等到最后几天才进行操作。为此，我决定写个简单的脚本，实现自动更换腾讯云CDN的数字证书。

0x02 准备

在使用该脚本前需要先在腾讯云CDN服务中完成域名的初步配置，也就是完成域名的添加并处于正常服务状态。完成添加的域名是否已启用HTTPS并不重要，在本脚本中的配置文件有相关设置内容，请注意本文的相关内容。

确认腾讯云CDN的服务处于正常状态后，需要准备调用API时用于验证的SecretId与SecretKey，该部分请根据腾讯云的文档进行安全配置，在这里我给出最快捷但不安全的获取方式：

登入腾讯云的管理界面后，将鼠标放置在用户名上，在弹出的菜单中选择“访问管理”打开管理页面，然后单击“云API密钥”中的“API密钥管理”，在打开的页面中新建或查看相关API密钥对。

注意！在此处建立API密钥对将有可能让你的腾讯云账户处于危险的处境，请根据腾讯云的相关手册进行安全配置！

将此密钥对记录到本地，稍后需要填写到脚本中。

第三步需要准备有效的数字证书，数字证书必须处于有效期内，必须是由可信的颁发机构颁发的且格式为pem。

当数字证书未生效、已过期或不受信任将使脚本运行错误并使腾讯云返还错误信息。有误的数字证书并不会对生产环境造成影响，因为它无法通过脚本和腾讯云的验证。

0x03 脚本

0x03.1 逻辑

我这个脚本的逻辑很简单，主要是校验证书的有效期与相关域名是否在该证书的授权范围内。如果一切正常、有效，则调用腾讯云的API。

脚本内并无校验证书有效性的模块，如果证书是由不受信任的颁发机构颁发的，调用腾讯云API后将返还错误信息。

0x03.2 配置文件

以下是配置文件的内容，json格式，可添加多个域名：

{
  "domain1.com": {
    "https_type": 2,
    "https_force_switch": 2,
    "http2": "on",
    "secret_id": "",
    "secret_key": "",
    "cert_filename": "",
    "key_filename": "",
    "validity": "10"
  },
  "www.domain1.com": {
    "https_type": 2,
    "https_force_switch": 2,
    "http2": "on",
    "secret_id": "",
    "secret_key": "",
    "cert_filename": "",
    "key_filename": "",
    "validity": "10"
  }
}

0x03.3 证书SAN字段

Subject Alternative Name简称为：SAN，该字段中记录着数字证书授权的所有域名：

在脚本中将读取该字段的内容，与配置文件中的key，也就是域名相比对。如果配置文件中的域名在该数字证书的授权范围内，则通过验证，否则将抛出错误并跳过。

在提取该字段的模块中，我是用cryptography这个函数：

# 查找证书的有效期、序列号
# 查找证书中包含的所有域名
def check_cert_info(cert_file_path):
    crt_open = open(cert_file_path, 'r')
    crt_content = crt_open.read()
    crt_open.close()

    crt = x509.load_pem_x509_certificate(crt_content.encode(), default_backend())

    # 获取证书序列号
    crt_serial_number = crt.serial_number

    # 获取alt name
    crt_altname = crt.extensions.get_extension_for_oid(x509.oid.ExtensionOID.SUBJECT_ALTERNATIVE_NAME)
    crt_altname = crt_altname.value.get_values_for_type(x509.DNSName)

    # 获取生效时间
    crt_not_valid_before = crt.not_valid_before
    # 获取过期时间
    crt_not_valid_after = crt.not_valid_after

    crt_output = dict()
    crt_output['crt_altname'] = crt_altname
    crt_output['crt_not_valid_before'] = crt_not_valid_before
    crt_output['crt_not_valid_after'] = crt_not_valid_after
    crt_output['crt_serial_number'] = crt_serial_number

    return crt_output

该自定义函数输出字典格式的内容，内容如下：

{'crt_serial_number': 311947205282921440934715912729475609824391, 'crt_not_valid_after': datetime.datetime(2018, 12, 31, 15, 59, 26), 'crt_not_valid_before': datetime.datetime(2018, 10, 2, 15, 59, 26), 'crt_altname': ['*.ngx.hk', 'ngx.hk']}

返还的内容中除了SAN字段的域名以外，还包含该证书的序列号与有效期。其中有效期为datetime类型，方便传递到下一个函数进行运算。

0x03.4 证书有效期

首先需要确认证书已生效，然后判断证书的失效期大于配置文件中“validity”的值，确认无误后分别将证书与私钥的内容进行base64编码：

def format_cert_key(domain_name, crt_file_name, key_file_name, crt_not_valid_before, crt_not_valid_after):
    crt_file_path = cert_file_folder + crt_file_name
    key_file_path = cert_file_folder + key_file_name

    datetime_now = datetime.utcnow()

    # 如果现在的时间大于证书生效时间，则为True
    if datetime_now >= crt_not_valid_before:
        # 如果证书失效时间-现在的时间大于配置文件中validity（单位：天）的值，则为True
        if (crt_not_valid_after - datetime_now).days >= int(config_json[domain_name]['validity']):
            crt_content = try_to_open_file(crt_file_path, 1)
            crt_base64 = base64.encodebytes(crt_content[1].encode()).decode()

            key_content = try_to_open_file(key_file_path, 1)
            key_base64 = base64.encodebytes(key_content[1].encode()).decode()

            output_dict = dict()
            output_dict['crt'] = crt_base64
            output_dict['key'] = key_base64
            return output_dict
        else:
            return False
    else:
        return False

如果相关的时间有误，则返还False。

0x03.5 检查CDN域名

在调试中我觉得需要增加一个检查腾讯CDN域名的模块，以此进一步提升可靠性：

def get_cdn_domain(config):
    # 腾讯云基础设定
    action = 'DescribeCdnHosts'
    module = 'cdn'
    params = {
        'detail': 0
    }
    # 调用API
    service = QcloudApi(module, config)
    # 可输出编码后的URL，主要用于日志，也可以生成URL后手动执行
    # 自动化应用一般不需要
    # print(service.generateUrl(action, params))
    # 执行API
    qcloud_output = service.call(action, params).decode()
    qcloud_output_json = json.loads(qcloud_output)

    cdn_host_list = []
    for i in qcloud_output_json['data']['hosts']:
        cdn_host_list.append(i['host'])

    return cdn_host_list

该模块会调用腾讯云CDN的API，获取CDN服务中所有的域名并以字典的形式输出，方便后续的模块检查配置文件中的域名是否已完成配置。

0x03.6 SAN校验

既然获得了证书中SAN字段的内容，接下来就需要进行校验，以下是校验模块的代码：

def crt_chk_alt_name(domain, alt_name):
    for i in alt_name:
        if i[0] == "*":
            i = i[2:]
            domain = domain.replace(i, '')
            if domain.count('.') == 1:
                return True
            else:
                pass
        else:
            if i == domain:
                return True
            else:
                pass
    return False

一开始我是使用正则来进行校验的，但实现起来过于复杂，最终形成了以上的形式。首先分为两种情况，一种是适配通配域名，另一种是普通域名。

首先函数接收两个传入参数，使用遍历功能逐一匹配alt_name变量中的域名。

如果第1个字符为“*”，则为通配域名，选取第3个到最后一个字符的内容传递给变量“i”；
将该变量中与变量“domain”相同的部分删除并重新赋值给变量“domain”；
统计变量“domain”中“.”的数量，如果为1则返还True，否则跳过此次循环。

第二种情况比较简单，如果变量“domain”匹配上列表“alt_name”中的某个值，则返还True否则跳过此次循环。

如果循环结束还没返还True，则返还False。

在这里演示下比较复杂的部分，举个例子，通配域名是这样的：*.ngx.hk，而我的CDN域名是这样的：cdn.ngx.hk，我需要怎样才能判断CDN域名在通配域名的子集里？

首先通配域名有一个特征，第一个字符肯定是星号“*”，后面跟着一个半个字符“.”，这样从第3个字符开始到最后一个字符为域名，这里成为域名i。

此时只需要将CDN域名中与域名i相同的部分删除，就会剩下“cdn.”。如果不是，则说明该CDN域名不在通配域名的子集里！

最后只需要计算删除后剩下的部分中的半个字符“.”的数量，如果为1则通过，否则为False。

0x03.7 临时文件

最后还需要一个临时文件的帮助，减少不必要的API调用。因为API调用次数在某一时段内是有限制的，所需要记录下域名与数字证书的序列号。

当脚本执行时或检索该临时文件，找到相关的序列号，如果临时文件中的序列号与证书文件中的一致，则不调用腾讯云CDN API，因为正是并未发生改动。

def write_temp_file(qcloud_output, domain, crt_serial_number):
    qcloud_output_json = json.loads(qcloud_output)
    if qcloud_output_json['code'] == 0:
        o_file = try_to_open_file(tmp_file_path, 0)
        if o_file[0]:
            content_dict = json.loads(o_file[1])
        else:
            content_dict = dict()

        o_file = open(tmp_file_path, 'w')
        content_dict[domain] = crt_serial_number
        content_dict = json.dumps(content_dict)
        o_file.write(str(content_dict))
        o_file.close()
    else:
        pass