Python实现HTTPS网站证书过期监控及更新
当前HTTP逐渐被大众所抛弃,HTTPS正在成为互联网上的主流。前段时间我们维护的一个HTTPS证书即将过期,由于多云环境比较复杂,团队小伙伴在替换更新证书的过程中出现疏漏,导致有一个域名证书没有及时更新,影响了系统可用性,为了杜绝这种问题再次发生,便写了这么个功能
比较简单,但很实用,再也不会出现证书漏更新的问题,具体流程为:扫描域名列表-->检查是否开启HTTPS-->获取证书过期时间-->记录入库-->更新证书
获取域名列表
我们用了很多的内部私有云SAAS服务,这些SAAS服务都提供有完善的API支持,DNS服务便是其中之一,可以根据DNS系统提供的API拿到所有的域名和记录。公有云也提供有完善的API文档,这里以阿里云为例,获取域名记录的代码如下
import json
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.acs_exception.exceptions import ClientException
from aliyunsdkcore.acs_exception.exceptions import ServerException
from aliyunsdkdomain.request.v20180129.QueryDomainListRequest import QueryDomainListRequest
from aliyunsdkalidns.request.v20150109.DescribeDomainRecordsRequest import DescribeDomainRecordsRequest
class DomainApi:
def __init__(self):
self.client = client = AcsClient('<accessKeyId>', '<accessSecret>', 'cn-hangzhou')
def get_domains(self, pagenum=1, pagesize=10):
try:
request = QueryDomainListRequest()
request.set_accept_format('json')
request.set_PageNum(pagenum)
request.set_PageSize(pagesize)
response = self.client.do_action_with_exception(request)
jsondata = json.loads(str(response, encoding='utf-8'))
return True, jsondata
except Exception as e:
return False, str(e)
def get_records(self, domain):
try:
request = DescribeDomainRecordsRequest()
request.set_accept_format('json')
request.set_DomainName(domain)
response = self.client.do_action_with_exception(request)
jsondata = json.loads(str(response, encoding='utf-8'))
return True, jsondata
except Exception as e:
return False, str(e)
if __name__ == '__main__':
print(DomainApi().get_records('ops-coffee.cn'))
以上代码使用了阿里云提供的SDK,调用简单方便,最终返回请求状态及数据。get_domains
方法可以获取到账号下的所有域名,get_records
方法可以获取到域名下的所有解析记录,需要注意数据量大小,以确定是否需要分页查询
检查是否开启HTTPS
检查是否开启HTTPS也是简单粗暴,直接通过requests
模块请求HTTPS地址,没有报错则表示开启了https支持,大概代码如下
session = requests.session()
try:
session.get('https://' + domain)
except Exception as e:
print(e)
需要注意的是,如果批量扫描域名的话需要使用session
模式,否则可能会因为链接过多而报错
获取证书过期时间
之后再通过pyopenssl
模块来拿到域名的HTTPS证书过期时间,代码如下
from _datetime import datetime
from urllib3.contrib import pyopenssl
def get_expire(domain):
try:
certificate = pyopenssl.ssl.get_server_certificate((domain, 443))
data = pyopenssl.OpenSSL.crypto.load_certificate(pyopenssl.OpenSSL.crypto.FILETYPE_PEM, certificate)
expire_time = datetime.strptime(data.get_notAfter().decode()[0:-1], '%Y%m%d%H%M%S')
expire_days = (expire_time - datetime.now()).days
return True, 200, {'expire_time': str(expire_time), 'expire_days': expire_days}
except Exception as e:
return False, 500, str(e)
if __name__ == '__main__':
print(get_expire('blog.ops-coffee.cn'))
使用之前需要先安装pyopenssl
模块,这里建议使用python3.6及以上版本,除了get_notAfter
可以拿到证书过期时间外,还有以下方法能够获取到更多证书相关的信息:get_notAfter
,get_notBefore
,get_pubkey
,get_serial_number
,get_signature_algorithm
,get_subject
,get_version
,gmtime_adj_notAfter
,gmtime_adj_notBefore
,has_expired
定时执行入库
以上步骤会定时执行监控,当发现证书过期时间小于30天时发报警,执行日志就写入了数据库方便前端页面展示
class Domain(models.Model):
create_time = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
domain = models.CharField(max_length=64, verbose_name='一级域名')
jsondata = models.TextField(verbose_name='域名详情')
数据库就三个字段,在每次执行完一轮扫码后都会将详细信息计入上表中
jsondata = {
'items': [{
'record': 'blog.ops-coffee.cn',
'rdtype': 'A',
'dept': '天玑'
'enable_https': 1,
'expire_time': '2021-10-22 12:00:00',
'expire_days': '387',
'notes': ''
}],
'total_count': 26,
'record_a_count': 23,
'record_a_https_count': 19}
Domain.objects.create(domain='ops-coffee.cn', jsondata=jsondata)
前端获取最新一条数据展示
Domain.objects.filter(domain=domain).order_by('-create_time').first()
更新证书
证书更新是最繁琐的事情,因为涉及到多平台不同环境,每家公司情况可能都有不同,我们因为大量使用了SAAS服务,SAAS服务又提供有API,所以更新起来比较简单,调用API即
class LBCApi:
def __init__(self):
self.domain = 'https://lbc.ops-coffee.cn'
self.headers = {
'content-type': 'application/json',
'Auth-Token': get_auth_token()[1]
}
def update_cert(self, id, cert):
try:
data = json.dumps({"cert": cert})
r = requests.put(self.domain + '/port/%d' % (id), data=data, headers=self.headers)
if r.status_code != 200:
return False, r.status_code, r.json()
return True, 200, port['id']
except Exception as e:
return False, 500, 'PortId:%s,更新证书失败' % str(e)
if __name__ == '__main__':
state, code, data = LBCApi().update_cert(37, 'cert-ops-coffee-cn')
print(state, code, json.dumps(data))
搞定收官!KPI++