监控SSL证书可用性并发送Telegram通知
实现功能
检查时间策略
告警时间策略
通知时间策略
采集器公网地址汇报
websites.yml
websites.ymlwebsites:
group_1:
api:
- xxxx.com
- xxxx.com
app:
- xxxx.com
- xxxx.com
group_2:
api:
- xxxx.com
- xxxx.com
app:
- xxxx.com
- xxxx.comcheck_ssl.py
#!/usr/bin/python3
# pip install python-whois requests psutil pyyaml
import configparser
import threading
import time
import os
import datetime
import sys
import ssl
import socket
import yaml
import psutil
import requests
PUBLIC_NETWORK = os.environ.get("PUBLIC_NETWORK", '8.8.8.8') # 公网IP地址
TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", '')
TELEGRAM_CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", '')
SSL_ALTER_INTERVAL = os.environ.get("SSL_ALTER_INTERVAL", 86400) # 告警间隔，单位秒
SSL_CHECK_INTERVAL = os.environ.get("SSL_CHECK_INTERVAL", 3600) # 证书检测间隔时间，单位秒
last_alert_times = {}
if TELEGRAM_BOT_TOKEN is None or TELEGRAM_CHAT_ID is None:
print(f"TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID can't be empty")
sys.exit(1)
def get_cpu_count():
"""获取 CPU 数量"""
return os.cpu_count()
def allocate_threads():
"""根据 CPU 数量分配线程数"""
cpu_count = get_cpu_count()
if cpu_count is None:
return 1 # 如果无法获取 CPU 数量，则默认为一个线程
elif cpu_count == 1:
return 1 # 如果只有一个 CPU，则使用一个线程
else:
return min(cpu_count * 3, 32) # 根据 CPU 数量动态分配线程数；min() 函数用于在多个参数中找到最小值, 例如，最多使用 16 个线程
def get_cpu_use():
'''获取 CPU 的负载率'''
cpu_load = psutil.cpu_percent(interval=1)
return cpu_load
def read_config_ini(filename):
'''读取配置文件'''
config = configparser.ConfigParser()
config.read(filename)
return config
# 读取配置文件
def read_config_yaml(filename):
with open(filename, 'r') as file:
config = yaml.safe_load(file)
return config
def send_telegram_alert(message):
'''发送 Telegram 告警'''
current_time = get_current_time()
print(f"{current_time} Sending Telegram alert for: {message}")
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
data = {"chat_id": TELEGRAM_CHAT_ID, "text": f"Collector: {PUBLIC_NETWORK} \n{message}"}
response = requests.post(url, data=data)
if response.status_code == 200:
print(f"{current_time} Message sent successfully!")
else:
print(f"{current_time} Failed to send message. Status code: {response.status_code}")
def get_current_time():
'''获取当前时间'''
current_datetime = datetime.datetime.now()
formatted_datetime = current_datetime.strftime("%Y-%m-%d %H:%M:%S")
return formatted_datetime
def get_cert_expiry(domain):
'''获取证书时间'''
# try:
context = ssl.create_default_context()
with context.wrap_socket(socket.socket(), server_hostname=domain) as ssock:
ssock.connect((domain, 443))
cert = ssock.getpeercert()
# 获取证书到期日期
expiry_date_str = cert['notAfter']
expiry_date = datetime.datetime.strptime(expiry_date_str, "%b %d %H:%M:%S %Y %Z")
return expiry_date
# except Exception as e:
# send_telegram_alert(f"Content: {domain} {e}")
# print("An error occurred:", e)
def check_certificate_expiry(url, remark):
'''检查是否使用 https 协议'''
try:
if url.startswith("https://"):
domain = url.split("//")[1]
expiry_date = get_cert_expiry(domain)
if expiry_date:
current_date = datetime.datetime.now()
days_remaining = (expiry_date - current_date).days
# # 获取上一次告警时间
last_alert_time = last_alert_times.get(f"{remark}_{domain}", 0)
c_time = time.time()
if c_time - last_alert_time > SSL_ALTER_INTERVAL:
# 在到期前特定天数发送 Telegram 通知
if days_remaining < 7:
send_telegram_alert(f"Content: 🤦Certificate will expire in ${days_remaining} days \n{remark}: {domain}")
last_alert_times[f"{remark}_{domain}"] = c_time
# print('正在检查:', domain)
current_time = get_current_time()
print(f"{current_time} {domain} Certificate expiration date：{expiry_date}, left until expiration {days_remaining} day")
except Exception as e:
# # 获取上一次告警时间
last_alert_time = last_alert_times.get(f"{remark}_{domain}", 0)
c_time = time.time()
if c_time - last_alert_time > SSL_ALTER_INTERVAL:
send_telegram_alert(f"Content: {url} {e}")
print("An error occurred:", e)
def main():
'''主函数'''
# config = read_config_yaml('websites.yml') # 读取配置文件
while True: # 持续循环检查
config = read_config_yaml('websites.yml') # 读取配置文件
must_schems = "https://"
ssl_threads = []
THREADS = allocate_threads()
# 根据环境判断使用进程，还是线程
current_time = get_current_time()
cpu_use = get_cpu_use()
print('='*70)
if cpu_use > 100: # cpu load 大于 100
msg = f"{current_time} cpu load is high ({cpu_use}%), continue, sleep {SSL_CHECK_INTERVAL} s......."
print(msg)
send_telegram_alert(f"Content: {msg}")
time.sleep(SSL_CHECK_INTERVAL)
continue
# print(config)
for website in config['websites']:
for type_name in config['websites'][website]:
for index, domain in enumerate(config['websites'][website][type_name]):
t = threading.Thread(target=check_certificate_expiry, args=(f"{must_schems}{domain}", f"{website}_{type_name}")) # 使用多线程检查网站状态
t.start()
ssl_threads.append(t)
if len(ssl_threads) >= THREADS:
for t in ssl_threads:
t.join()
ssl_threads = []
time.sleep(SSL_CHECK_INTERVAL) # 休眠后再次检查
if __name__ == "__main__":
main()效果
Docker 容器方式
root@monitor-1:/data/monitor/check# cat ssl_Dockerfile
FROM python:3.9-slim
MAINTAINER ck
WORKDIR /app
COPY check_ssl.py main.py
RUN pip install requests psutil pyyaml python-whois --no-cache-dir
ENV TZ=Asia/Shanghai
RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
CMD ["python", "main.py"]docker build --no-cache -t check_ssl -f ssl_Dockerfile .
docker-compose up -ddocker-compose.yaml
root@monitor-1:/data/monitor/check# cat docker-compose.yml
version: "3"
services:
check_ssl:
image: check_ssl:latest
container_name: check_ssl
network_mode: host
dns:
- 8.8.8.8
environment:
- PUBLIC_NETWORK=47.57.142.166
- TELEGRAM_BOT_TOKEN=6685540055:AAHTc5hXy77YV9TPck72HYhQcU53FH82YhM
- TELEGRAM_CHAT_ID=-1002013511378
volumes:
- ${PWD}/websites.yml:/app/websites.ymlLast updated