mokee/nginxserver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

初始化1

Browse Source
This commit is contained in:
zg
2026-07-16 13:03:11 +08:00
parent 0bdfcbc1c8
commit e6b0f287cc
100 changed files with 12184 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
nginx-python-api/services/__init__.py Normal file
View File

263
nginx-python-api/services/cert_notifier.py Normal file
View File

@@ -0,0 +1,263 @@
"""
证书到期邮件通知服务
- 由 cron 每日 10:00 触发
- 检查数据库中的 Let's Encrypt 证书
- 到期前 N 天起，连续每天发送提醒邮件
- 邮件通过 Mokee Gateway 对外开放 API 发送
使用方式:
python cert_notifier.py # 直接运行（cron 调用）
python cert_notifier.py --dry-run # 预览模式，不实际发送
python cert_notifier.py --check # 仅显示证书状态
"""
import os
import sys
import json
import logging
import urllib.request
import urllib.error
from datetime import datetime
# 数据库配置
DB_HOST = os.environ.get('MYSQL_HOST', '10.1.1.122')
DB_PORT = int(os.environ.get('MYSQL_PORT', '3306'))
DB_USER = os.environ.get('MYSQL_USER', 'nginxserver')
DB_PASS = os.environ.get('MYSQL_PASS', 'mokee.')
DB_NAME = os.environ.get('MYSQL_DB', 'nginxserver')
# 网关配置（可通过环境变量覆盖）
GATEWAY_URL = os.environ.get('MOKEE_GATEWAY_URL', 'https://gw.server.mokee.com')
SYSTEM_CODE = os.environ.get('MOKEE_SYSTEM_CODE', 'nginxServer')
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s'
)
logger = logging.getLogger(__name__)
def get_db_connection():
"""获取数据库连接（延迟导入，避免非数据库场景报错）"""
import pymysql
return pymysql.connect(
host=DB_HOST, port=DB_PORT, user=DB_USER,
password=DB_PASS, database=DB_NAME,
charset='utf8mb4', connect_timeout=5
)
def get_token():
"""从 Mokee Gateway 获取一次性 Token"""
url = f"{GATEWAY_URL}/zgapi/v1/open/token"
data = json.dumps({"systemCode": SYSTEM_CODE}).encode('utf-8')
req = urllib.request.Request(url, data=data, method='POST')
req.add_header('Content-Type', 'application/json')
try:
with urllib.request.urlopen(req, timeout=10) as resp:
result = json.loads(resp.read().decode('utf-8'))
if result.get('code') == 200:
token = result['data']['token']
logger.info(f"获取 Token 成功: {SYSTEM_CODE}")
return token
else:
logger.error(f"获取 Token 失败: {result}")
return None
except Exception as e:
logger.error(f"获取 Token 异常: {e}")
return None
def send_email(token, to, subject, content):
"""
通过 Mokee Gateway 发送邮件
成功 = Token 被消费，需要重新获取
"""
url = f"{GATEWAY_URL}/zgapi/v1/open/email/send"
data = json.dumps({
"to": to,
"subject": subject,
"content": content,
"contentType": "html"
}).encode('utf-8')
req = urllib.request.Request(url, data=data, method='POST')
req.add_header('Content-Type', 'application/json')
req.add_header('Authorization', f'Bearer {token}')
try:
with urllib.request.urlopen(req, timeout=15) as resp:
result = json.loads(resp.read().decode('utf-8'))
return result.get('code') == 200, str(result)
except Exception as e:
return False, str(e)
def build_email_html(domain, days_left, expire_date):
"""生成 HTML 邮件内容"""
if days_left <= 5:
level = 'danger'
color = '#F56C6C'
elif days_left <= 10:
level = 'warning'
color = '#E6A23C'
else:
level = 'info'
color = '#409EFF'
return f"""<!DOCTYPE html>
<html><body style="font-family: 'Microsoft YaHei', Arial, sans-serif;">
<div style="max-width:600px;margin:0 auto;padding:20px;border:1px solid #EBEEF5;border-radius:8px;">
<div style="background:{color};padding:16px;border-radius:8px 8px 0 0;text-align:center;">
<h2 style="color:#fff;margin:0;">⚠️ SSL 证书即将到期</h2>
</div>
<div style="padding:20px;background:#fff;">
<table style="width:100%;border-collapse:collapse;">
<tr><td style="padding:8px 0;color:#909399;width:80px;">域名</td>
<td style="padding:8px 0;font-weight:bold;font-size:16px;">{domain}</td></tr>
<tr><td style="padding:8px 0;color:#909399;">到期时间</td>
<td style="padding:8px 0;">{expire_date}</td></tr>
<tr><td style="padding:8px 0;color:#909399;">剩余天数</td>
<td style="padding:8px 0;color:{color};font-weight:bold;font-size:20px;">{days_left} 天</td></tr>
<tr><td style="padding:8px 0;color:#909399;">证书类型</td>
<td style="padding:8px 0;">Let's Encrypt (阿里云 DNS 验证)</td></tr>
</table>
<hr style="border:0;border-top:1px solid #EBEEF5;margin:16px 0;">
<p style="color:#909399;font-size:13px;">
系统已配置 acme.sh 自动续期，正常情况下会在到期前自动更新证书。<br>
如续期失败，请检查：<br>
1. 阿里云 AccessKey 是否有效<br>
2. 服务器能否访问 Let's Encrypt API<br>
3. DNS TXT 记录是否正常清理
</p>
<p style="color:#C0C4CC;font-size:12px;">
此邮件由 Nginx 域名管理平台自动发送，连续提醒至证书续期成功。
</p>
</div>
</div>
</body></html>"""
def check_and_notify(dry_run=False):
"""检查证书到期情况并发送通知"""
conn = get_db_connection()
cursor = conn.cursor()
# 查询需要通知的 Let's Encrypt 证书
cursor.execute('''
SELECT id, domain, notify_email, notify_enabled, notify_days_before, expire_date
FROM certificates
WHERE cert_type = 'letsencrypt'
AND status = 'active'
AND notify_enabled = 1
AND notify_email IS NOT NULL
AND notify_email != ''
''')
certs = cursor.fetchall()
if not certs:
logger.info("无需通知的 Let's Encrypt 证书")
conn.close()
return
today = datetime.now()
token = None # 延迟获取
notified_count = 0
for cert_id, domain, notify_email, enabled, days_before, expire_date_db in certs:
# 从数据库获取过期日期
if expire_date_db:
expire_date = expire_date_db if isinstance(expire_date_db, datetime) else datetime.strptime(str(expire_date_db), '%Y-%m-%d %H:%M:%S')
days_left = (expire_date - today).days
else:
logger.warning(f"{domain}: 无过期日期信息，跳过")
continue
logger.info(f"{domain}: 剩余 {days_left} 天到期 (通知阈值: {days_before}天前开始)")
# 判断是否需要通知: days_before 天前开始，连续通知到到期前1天
if days_left > days_before or days_left <= 0:
logger.info(f"{domain}: 不在通知窗口 (days_left={days_left}, threshold={days_before})")
continue
# 检查今天是否已发送过通知
cursor.execute('''
SELECT id FROM cert_notify_log
WHERE cert_id = %s AND DATE(sent_at) = CURDATE()
''', (cert_id,))
if cursor.fetchone():
logger.info(f"{domain}: 今日已通知，跳过")
continue
# 生成邮件内容
expire_str = expire_date.strftime('%Y-%m-%d %H:%M')
subject = f"⚠️ SSL证书到期提醒 - {domain} (剩余{days_left}天)"
content = build_email_html(domain, days_left, expire_str)
if dry_run:
logger.info(f"[DRY-RUN] 将发送邮件: to={notify_email}, subject={subject}")
continue
# 获取 Token（每次发送前确保有效）
if token is None:
token = get_token()
if token is None:
logger.error(f"无法获取 Token，跳过 {domain}")
continue
# 发送邮件
success, resp = send_email(token, notify_email, subject, content)
if success:
# Token 已消费，下次需要重新获取
token = None
logger.info(f"{domain}: 邮件发送成功 → {notify_email}")
else:
logger.warning(f"{domain}: 邮件发送失败 → {resp}")
# 500 错误不消费 Token，可重试
# 记录日志
cursor.execute('''
INSERT INTO cert_notify_log (cert_id, domain, days_left, email_to, success, response_msg)
VALUES (%s, %s, %s, %s, %s, %s)
''', (cert_id, domain, days_left, notify_email, 1 if success else 0, resp[:1000]))
conn.commit()
if success:
notified_count += 1
conn.close()
logger.info(f"通知完成: 发送 {notified_count} 封邮件")
def check_status():
"""仅显示证书状态（--check 模式）"""
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute('''
SELECT domain, cert_type, status, expire_date,
DATEDIFF(COALESCE(expire_date, NOW()), NOW()) AS days_left,
notify_enabled, notify_email, notify_days_before
FROM certificates
WHERE cert_type = 'letsencrypt'
ORDER BY days_left
''')
print(f"\n{'域名':25s} {'类型':12s} {'状态':8s} {'到期日期':20s} {'剩余天数':>8s} {'通知':>4s} {'收件人'}")
print('-' * 120)
for row in cursor.fetchall():
domain, cert_type, status, expire_date, days_left, enabled, email, days_before = row
dl = str(days_left) if days_left is not None else 'N/A'
print(f"{domain:25s} {cert_type:12s} {status:8s} {str(expire_date):20s} {dl:>8s} {'是' if enabled else '否':>4s} {email or '-'}")
conn.close()
if __name__ == '__main__':
if '--dry-run' in sys.argv:
print("=== 预览模式 (--dry-run) ===\n")
check_and_notify(dry_run=True)
elif '--check' in sys.argv:
check_status()
else:
check_and_notify()

228
nginx-python-api/services/dns_service.py Normal file
View File

@@ -0,0 +1,228 @@
"""
阿里云 DNS 解析管理服务
通过阿里云 OpenAPI 管理 zgitm.com 的 DNS 解析记录
使用 HMAC-SHA1 签名，兼容 Python 3.6+
"""
import os
import hmac
import hashlib
import base64
import logging
import urllib.request
import urllib.parse
import urllib.error
import json
import time
import uuid
logger = logging.getLogger(__name__)
# 阿里云 API 配置
ALIDNS_ENDPOINT = 'https://alidns.aliyuncs.com'
API_VERSION = '2015-01-09'
SIGNATURE_METHOD = 'HMAC-SHA1'
SIGNATURE_VERSION = '1.0'
class DnsService:
"""阿里云 DNS 解析管理"""
def __init__(self):
self.access_key_id = os.environ.get('Ali_Key', '')
self.access_key_secret = os.environ.get('Ali_Secret', '')
def _sign(self, method, params):
"""
HMAC-SHA1 签名（阿里云 API 签名 v1.0）
"""
# 1. 按参数名排序
sorted_keys = sorted(params.keys())
sorted_params = [(k, params[k]) for k in sorted_keys]
# 2. 构造规范化查询字符串
canonicalized = urllib.parse.urlencode(sorted_params)
# 3. 构造签名源字符串
string_to_sign = method + '&' + urllib.parse.quote('/', safe='') + '&' + urllib.parse.quote(canonicalized, safe='')
# 4. HMAC-SHA1 签名
key = (self.access_key_secret + '&').encode('utf-8')
h = hmac.new(key, string_to_sign.encode('utf-8'), hashlib.sha1)
signature = base64.b64encode(h.digest()).decode('utf-8')
return signature
def _call(self, action, extra_params=None):
"""
调用阿里云 OpenAPI
Args:
action: API 名称 (如 DescribeDomainRecords)
extra_params: 额外参数
Returns:
dict: API 响应
"""
if not self.access_key_id or not self.access_key_secret:
raise RuntimeError('阿里云 AccessKey 未配置')
# 公共参数
params = {
'AccessKeyId': self.access_key_id,
'Action': action,
'Format': 'JSON',
'Version': API_VERSION,
'SignatureMethod': SIGNATURE_METHOD,
'SignatureVersion': SIGNATURE_VERSION,
'SignatureNonce': str(uuid.uuid4()),
'Timestamp': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()),
}
if extra_params:
params.update(extra_params)
# 签名
params['Signature'] = self._sign('GET', params)
# 构造 URL
url = ALIDNS_ENDPOINT + '/?' + urllib.parse.urlencode(params)
# 发送请求
try:
req = urllib.request.Request(url, method='GET')
with urllib.request.urlopen(req, timeout=15) as resp:
result = json.loads(resp.read().decode('utf-8'))
return result
except urllib.error.HTTPError as e:
body = e.read().decode('utf-8') if e.fp else ''
logger.error(f'DNS API HTTP {e.code}: {body}')
raise RuntimeError(f'DNS API 请求失败: {e.code} {body}')
except Exception as e:
logger.error(f'DNS API 异常: {e}')
raise RuntimeError(f'DNS API 异常: {e}')
def list_records(self, domain_zone, rr_keyword=None, record_type=None):
"""
获取域名解析记录列表
Args:
domain_zone: 主域名 (如 zgitm.com)
rr_keyword: 主机记录筛选 (如 www)
record_type: 记录类型筛选 (如 A, CNAME)
Returns:
list[dict]: 解析记录列表
"""
params = {'DomainName': domain_zone}
if rr_keyword:
params['RRKeyWord'] = rr_keyword
if record_type:
params['Type'] = record_type
resp = self._call('DescribeDomainRecords', params)
records = resp.get('DomainRecords', {}).get('Record', [])
# 统一转列表
if isinstance(records, dict):
records = [records]
result = []
for r in records:
result.append({
'record_id': r.get('RecordId', ''),
'rr': r.get('RR', ''),
'type': r.get('Type', 'A'),
'value': r.get('Value', ''),
'ttl': r.get('TTL', 600),
'line': r.get('Line', 'default'),
'status': r.get('Status', 'ENABLE'),
'full_domain': (r.get('RR', '') + '.' + domain_zone).strip('.'),
})
return result
def add_record(self, domain_zone, rr, record_type, value, ttl=600):
"""
添加解析记录
Args:
domain_zone: 主域名
rr: 主机记录 (如 webssd, @, www)
record_type: 记录类型 (A, CNAME, TXT 等)
value: 记录值 (IP地址或域名)
ttl: TTL 秒数
Returns:
dict: {'record_id': ..., 'full_domain': ...}
"""
params = {
'DomainName': domain_zone,
'RR': rr,
'Type': record_type.upper(),
'Value': value,
'TTL': str(ttl),
}
resp = self._call('AddDomainRecord', params)
record_id = resp.get('RecordId', '')
full_domain = (rr + '.' + domain_zone).strip('.').replace('@.', '')
logger.info(f'DNS 记录已添加: {full_domain} {record_type} {value} (ID: {record_id})')
return {'record_id': record_id, 'full_domain': full_domain}
def delete_record(self, record_id):
"""
删除解析记录
Args:
record_id: 记录ID
Returns:
bool: 是否成功
"""
resp = self._call('DeleteDomainRecord', {'RecordId': str(record_id)})
logger.info(f'DNS 记录已删除: {record_id}')
return True
def update_record(self, record_id, rr, record_type, value, ttl=600):
"""
修改解析记录
Args:
record_id: 记录ID
rr: 主机记录
record_type: 记录类型
value: 记录值
ttl: TTL
Returns:
bool: 是否成功
"""
params = {
'RecordId': str(record_id),
'RR': rr,
'Type': record_type.upper(),
'Value': value,
'TTL': str(ttl),
}
try:
self._call('UpdateDomainRecord', params)
logger.info(f'DNS 记录已修改: {record_id} → {rr} {record_type} {value}')
except RuntimeError as e:
msg = str(e)
# 未做任何改动直接保存，阿里云返回 DomainRecordDuplicate，视为成功
if 'DomainRecordDuplicate' in msg:
logger.info(f'DNS 记录未变更，跳过: {record_id} {rr} {record_type} {value}')
else:
raise
return True
def set_record_status(self, record_id, status):
"""
启用/停用解析记录
Args:
record_id: 记录ID
status: Enable / Disable
"""
self._call('SetDomainRecordStatus', {
'RecordId': str(record_id),
'Status': status.upper(),
})
logger.info(f'DNS 记录状态已更新: {record_id} → {status}')
return True

567
nginx-python-api/services/nginx_service.py Normal file
View File

@@ -0,0 +1,567 @@
"""
Nginx 配置管理服务
负责读写 Nginx 配置文件、重载 Nginx
全局优化（在 /etc/nginx/nginx.conf http 块中）:
- gzip + gzip_types（14种MIME）— 压缩传输
- SSL session cache (10m) + timeout (60m) — 减少重复TLS握手
- ssl_session_tickets off + ssl_buffer_size 4k
每个域名配置的优化:
- listen 443 ssl http2 — HTTP/2 多路复用
- proxy_buffering off — 边收边发，不阻塞
- 静态资源 30天强缓存（public, immutable）
"""
import os
import re
import subprocess
import logging
logger = logging.getLogger(__name__)
# Nginx HTTP 配置模板
HTTP_TEMPLATE = """# Nginx转发规则 - {domain}
# 协议: HTTP
# 自动生成，请勿手动修改
server {{
listen 80;
server_name {domain};
client_max_body_size 2048m;
# 静态资源缓存（JS/CSS 带 hash，30天）
location ~* \\.(js|css|svg|woff2|png|jpg|ico)$ {{
expires 30d;
add_header Access-Control-Allow-Origin * always;
add_header Access-Control-Allow-Methods \"GET, POST, PUT, DELETE, OPTIONS\" always;
add_header Access-Control-Allow-Headers \"Content-Type, Authorization\" always;
add_header Cache-Control \"public, immutable\";
proxy_pass {target};
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_http_version 1.1;
proxy_set_header Connection \"\";
proxy_set_header Accept-Encoding \"gzip\";
proxy_connect_timeout 43200s;
proxy_send_timeout 43200s;
proxy_read_timeout 43200s;
proxy_buffering off;
}}
location / {{
add_header Access-Control-Allow-Origin * always;
add_header Access-Control-Allow-Methods \"GET, POST, PUT, DELETE, OPTIONS\" always;
add_header Access-Control-Allow-Headers \"Content-Type, Authorization\" always;
proxy_pass {target};
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_http_version 1.1;
proxy_set_header Connection \"\";
proxy_set_header Accept-Encoding \"gzip\";
proxy_connect_timeout 43200s;
proxy_send_timeout 43200s;
proxy_read_timeout 43200s;
proxy_buffering off;
}}
}}
"""
# Nginx HTTPS 配置模板
HTTPS_TEMPLATE = """# Nginx转发规则 - {domain}
# 协议: HTTPS
# 自动生成，请勿手动修改
# HTTP → HTTPS 重定向
server {{
listen 80;
server_name {domain};
return 301 https://$host$request_uri;
}}
# HTTPS 服务
server {{
listen 443 ssl http2;
server_name {domain};
# CORS 跨域响应头（server 级别，确保 OPTIONS 204 也带 CORS 头）
add_header Access-Control-Allow-Origin * always;
add_header Access-Control-Allow-Methods \"GET, POST, PUT, DELETE, OPTIONS\" always;
add_header Access-Control-Allow-Headers \"Content-Type, Authorization\" always;
# CORS 预检请求快速返回（避免穿透到后端）
if ($request_method = OPTIONS) {{ return 204; }}
ssl_certificate {ssl_cert};
ssl_certificate_key {ssl_key};
ssl_protocols TLSv1.2 TLSv1.3;
ssl_ciphers HIGH:!aNULL:!MD5;
ssl_prefer_server_ciphers on;
# 静态资源缓存（JS/CSS 带 hash，30天）
location ~* \\.(js|css|svg|woff2|png|jpg|ico)$ {{
expires 30d;
add_header Access-Control-Allow-Origin * always;
add_header Access-Control-Allow-Methods \"GET, POST, PUT, DELETE, OPTIONS\" always;
add_header Access-Control-Allow-Headers \"Content-Type, Authorization\" always;
add_header Cache-Control \"public, immutable\";
proxy_pass {target};
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_http_version 1.1;
proxy_set_header Connection \"\";
proxy_set_header Accept-Encoding \"gzip\";
proxy_connect_timeout 43200s;
proxy_send_timeout 43200s;
proxy_read_timeout 43200s;
proxy_buffering off;
}}
location / {{
add_header Access-Control-Allow-Origin * always;
add_header Access-Control-Allow-Methods \"GET, POST, PUT, DELETE, OPTIONS\" always;
add_header Access-Control-Allow-Headers \"Content-Type, Authorization\" always;
proxy_pass {target};
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_http_version 1.1;
proxy_set_header Connection \"\";
proxy_set_header Accept-Encoding \"gzip\";
proxy_connect_timeout 43200s;
proxy_send_timeout 43200s;
proxy_read_timeout 43200s;
proxy_buffering off;
}}
}}
"""
class NginxService:
"""Nginx 配置管理服务"""
def __init__(self, conf_dir='/etc/nginx/conf.d', ssl_dir='/etc/nginx/ssl'):
self.conf_dir = conf_dir
self.ssl_dir = ssl_dir
# 确保目录存在
os.makedirs(self.conf_dir, exist_ok=True)
def create_config(self, domain, protocol, target, ssl_cert=None, ssl_key=None):
"""
创建域名转发配置文件
Args:
domain: 域名
protocol: 协议 (http/https)
target: 转发目标地址
ssl_cert: SSL证书路径（仅在HTTPS时需要）
ssl_key: SSL私钥路径（仅在HTTPS时需要）
Returns:
config_file: 生成的配置文件路径
"""
config_path = os.path.join(self.conf_dir, f'{domain}.conf')
if protocol == 'https':
content = HTTPS_TEMPLATE.format(
domain=domain,
target=target,
ssl_cert=ssl_cert or f'{self.ssl_dir}/{domain}.pem',
ssl_key=ssl_key or f'{self.ssl_dir}/{domain}.key'
)
else:
content = HTTP_TEMPLATE.format(
domain=domain,
target=target
)
with open(config_path, 'w', encoding='utf-8') as f:
f.write(content)
logger.info(f"已创建配置文件: {config_path}")
return config_path
def delete_config(self, domain):
"""
删除域名配置文件
Args:
domain: 域名
Returns:
bool: 是否成功删除
"""
config_path = os.path.join(self.conf_dir, f'{domain}.conf')
if os.path.exists(config_path):
os.remove(config_path)
logger.info(f"已删除配置文件: {config_path}")
return True
logger.warning(f"配置文件不存在: {config_path}")
return False
def list_rules(self):
"""
解析所有 Nginx 配置文件，提取转发规则列表
Returns:
list[dict]: 规则列表
"""
rules = []
if not os.path.exists(self.conf_dir):
return rules
for filename in sorted(os.listdir(self.conf_dir)):
if not filename.endswith('.conf'):
continue
filepath = os.path.join(self.conf_dir, filename)
rule = self._parse_config(filepath)
if rule:
rules.append(rule)
return rules
def find_rule(self, domain):
"""
查找指定域名的规则
Args:
domain: 域名
Returns:
dict | None: 规则信息
"""
config_path = os.path.join(self.conf_dir, f'{domain}.conf')
if not os.path.exists(config_path):
return None
return self._parse_config(config_path)
def _parse_config(self, filepath):
"""
解析单个 Nginx 配置文件
Args:
filepath: 配置文件路径
Returns:
dict | None: 解析出的规则信息
"""
try:
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
rule = {
'config_file': filepath
}
# 提取域名 (server_name)
server_name_match = re.search(r'server_name\s+([^;]+);', content)
if server_name_match:
rule['domain'] = server_name_match.group(1).strip()
# 提取转发目标 (proxy_pass)
proxy_pass_match = re.search(r'proxy_pass\s+([^;]+);', content)
if proxy_pass_match:
rule['target'] = proxy_pass_match.group(1).strip()
# 判断协议类型
if 'listen 443 ssl' in content:
rule['protocol'] = 'https'
# 提取 SSL 证书路径
cert_match = re.search(r'ssl_certificate\s+([^;]+);', content)
if cert_match:
rule['ssl_cert'] = cert_match.group(1).strip()
key_match = re.search(r'ssl_certificate_key\s+([^;]+);', content)
if key_match:
rule['ssl_key'] = key_match.group(1).strip()
else:
rule['protocol'] = 'http'
return rule
except Exception as e:
logger.error(f"解析配置文件失败 {filepath}: {e}")
return None
def check_config(self):
"""
检查 Nginx 配置语法
Returns:
tuple: (是否通过, 错误信息)
"""
try:
result = subprocess.run(
['nginx', '-t'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=True,
timeout=10
)
if result.returncode == 0:
return True, 'ok'
else:
error_msg = result.stderr.strip() or '配置语法错误'
return False, error_msg
except FileNotFoundError:
return False, 'nginx 命令未找到，请确认 Nginx 已安装'
except subprocess.TimeoutExpired:
return False, 'nginx -t 执行超时'
except Exception as e:
return False, str(e)
def reload(self):
"""
重载 Nginx 配置
Returns:
tuple: (是否成功, 消息)
"""
# 先检查配置语法
valid, msg = self.check_config()
if not valid:
return False, f'配置语法检查失败: {msg}'
# 执行 reload
try:
# 尝试 systemctl reload
result = subprocess.run(
['systemctl', 'reload', 'nginx'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=True,
timeout=30
)
if result.returncode == 0:
logger.info("Nginx 重载成功 (systemctl)")
return True, 'Nginx 重载成功'
# 回退到 nginx -s reload
result = subprocess.run(
['nginx', '-s', 'reload'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=True,
timeout=30
)
if result.returncode == 0:
logger.info("Nginx 重载成功 (nginx -s reload)")
return True, 'Nginx 重载成功'
error_msg = result.stderr.strip() or '未知错误'
return False, f'重载失败: {error_msg}'
except FileNotFoundError:
return False, 'nginx 命令未找到'
except subprocess.TimeoutExpired:
return False, 'reload 执行超时'
except Exception as e:
return False, str(e)
def parse_gzip_config(self, domain):
"""
从配置文件中解析 gzip 压缩配置
Args:
domain: 域名
Returns:
dict: gzip 配置信息
"""
config_path = os.path.join(self.conf_dir, f'{domain}.conf')
if not os.path.exists(config_path):
return {
'enabled': False,
'types': [],
'comp_level': 6,
'min_length': '1000'
}
with open(config_path, 'r', encoding='utf-8') as f:
content = f.read()
result = {
'enabled': False,
'types': [],
'comp_level': 6,
'min_length': '1000'
}
# 解析 gzip on/off
gzip_match = re.search(r'^\s*gzip\s+(on|off)\s*;', content, re.MULTILINE)
if gzip_match:
result['enabled'] = gzip_match.group(1) == 'on'
# 解析 gzip_types
types_match = re.search(r'^\s*gzip_types\s+([^;]+);', content, re.MULTILINE)
if types_match:
types_str = types_match.group(1).strip()
result['types'] = [t.strip() for t in types_str.split() if t.strip()]
# 解析 gzip_comp_level
level_match = re.search(r'^\s*gzip_comp_level\s+(\d+)\s*;', content, re.MULTILINE)
if level_match:
result['comp_level'] = int(level_match.group(1))
# 解析 gzip_min_length
length_match = re.search(r'^\s*gzip_min_length\s+(\S+)\s*;', content, re.MULTILINE)
if length_match:
result['min_length'] = length_match.group(1).strip()
return result
def set_gzip_config(self, domain, gzip_config):
"""
修改配置文件中的 gzip 压缩配置（在 server 块内添加/更新 gzip 指令）
Args:
domain: 域名
gzip_config: dict with keys:
- enabled (bool): 是否开启
- types (list[str]): MIME 类型列表
- comp_level (int): 压缩级别 1-9
- min_length (str): 最小压缩长度
Returns:
tuple: (成功, 消息)
"""
config_path = os.path.join(self.conf_dir, f'{domain}.conf')
if not os.path.exists(config_path):
return False, f'域名 {domain} 的配置文件不存在'
# 读取原内容
with open(config_path, 'r', encoding='utf-8') as f:
original_content = f.read()
content = original_content
# 1. 删除 server 块内所有已有的 gzip 指令（包括注释行）
content = re.sub(r'^\s*#\s*内容压缩[^\n]*\n?', '', content, flags=re.MULTILINE)
content = re.sub(r'^\s*gzip\s+(on|off)\s*;\s*\n?', '', content, flags=re.MULTILINE)
content = re.sub(r'^\s*gzip_types\s+[^;]+;\s*\n?', '', content, flags=re.MULTILINE)
content = re.sub(r'^\s*gzip_comp_level\s+\d+\s*;\s*\n?', '', content, flags=re.MULTILINE)
content = re.sub(r'^\s*gzip_min_length\s+\S+\s*;\s*\n?', '', content, flags=re.MULTILINE)
# 2. 构建新的 gzip 指令块
gzip_lines = []
if gzip_config.get('enabled', False):
gzip_lines.append(' # 内容压缩（gzip）')
gzip_lines.append(' gzip on;')
types = gzip_config.get('types', [])
if types:
types_str = ' '.join(types)
gzip_lines.append(f' gzip_types {types_str};')
comp_level = gzip_config.get('comp_level', 6)
gzip_lines.append(f' gzip_comp_level {comp_level};')
min_length = gzip_config.get('min_length', '1000')
gzip_lines.append(f' gzip_min_length {min_length};')
else:
gzip_lines.append(' # 内容压缩（gzip）')
gzip_lines.append(' gzip off;')
gzip_block = '\n'.join(gzip_lines) + '\n'
# 3. 在 server_name 行之后插入 gzip 指令块
# 匹配 server_name xxx; 并在其后插入
def insert_gzip(match):
return match.group(0) + '\n' + gzip_block
content = re.sub(
r'^\s*server_name\s+[^;]+;\s*$',
insert_gzip,
content,
flags=re.MULTILINE
)
# 4. 写入文件
with open(config_path, 'w', encoding='utf-8') as f:
f.write(content)
logger.info(f"已更新 {domain} 的 gzip 压缩配置: enabled={gzip_config.get('enabled')}")
# 5. 语法检查
valid, msg = self.check_config()
if not valid:
# 回滚
with open(config_path, 'w', encoding='utf-8') as f:
f.write(original_content)
logger.error(f"gzip 配置语法错误，已回滚: {msg}")
return False, f'配置语法检查失败，已自动回滚: {msg}'
# 6. 重载 Nginx
reload_success, reload_msg = self.reload()
if not reload_success:
logger.warning(f"Nginx 重载失败: {reload_msg}")
return False, f'gzip 配置已写入但重载失败: {reload_msg}'
return True, 'gzip 压缩配置已保存并生效'
def get_domain_logs(self, domain, lines=200):
"""
获取指定域名的 Nginx 访问日志和错误日志
Args:
domain: 域名
lines: 返回最近多少行日志（默认200）
Returns:
dict: access_log 和 error_log 内容
"""
result = {
'domain': domain,
'access_log': '',
'error_log': '',
'access_log_path': '/var/log/nginx/access.log',
'error_log_path': '/var/log/nginx/error.log'
}
# 访问日志 — 过滤包含域名的行
access_log_path = '/var/log/nginx/access.log'
if os.path.exists(access_log_path):
try:
# 使用 tail + grep 获取最近包含域名的日志行
import subprocess as _sp
proc = _sp.run(
['tail', '-n', str(lines * 5), access_log_path],
stdout=_sp.PIPE, stderr=_sp.PIPE,
universal_newlines=True, timeout=10
)
if proc.returncode == 0:
all_lines = proc.stdout.splitlines()
matched = [l for l in all_lines if domain in l]
result['access_log'] = '\n'.join(matched[-lines:])
except Exception as e:
logger.warning(f"读取访问日志失败: {e}")
result['access_log'] = f'[读取失败] {e}'
# 错误日志 — 过滤包含域名的行
error_log_path = '/var/log/nginx/error.log'
if os.path.exists(error_log_path):
try:
import subprocess as _sp
proc = _sp.run(
['tail', '-n', str(lines * 5), error_log_path],
stdout=_sp.PIPE, stderr=_sp.PIPE,
universal_newlines=True, timeout=10
)
if proc.returncode == 0:
all_lines = proc.stdout.splitlines()
matched = [l for l in all_lines if domain in l]
result['error_log'] = '\n'.join(matched[-lines:])
except Exception as e:
logger.warning(f"读取错误日志失败: {e}")
result['error_log'] = f'[读取失败] {e}'
return result
def _escape_nginx(content):
"""转义 Nginx 配置中的特殊字符（模板用）"""
return content

532
nginx-python-api/services/ssl_service.py Normal file
View File

@@ -0,0 +1,532 @@
"""
SSL 证书管理服务
支持两种证书签发方式：
- 内部自建 CA → 手动创建的内网域名
- Let's Encrypt DNS-01 → 系统内置域名 (zgitm.com)
"""
import os
import subprocess
import json
import logging
from datetime import datetime, timedelta
logger = logging.getLogger(__name__)
# 证书有效期（天）
CERT_VALIDITY_DAYS = 365
# ─── Let's Encrypt 系统域名配置 ───────────────────
# zgitm.com 及其所有子域名使用 Let's Encrypt 证书
LE_ROOT_DOMAIN = 'zgitm.com'
ACME_SH = '/root/.acme.sh/acme.sh'
LE_STAGING = os.environ.get('LE_STAGING', '0') == '1' # 1=测试环境
# OpenSSL 配置文件模板（用于签发域名证书）
OPENSSL_CNF_TEMPLATE = """[req]
default_bits = 2048
prompt = no
default_md = sha256
distinguished_name = dn
req_extensions = req_ext
[dn]
CN = {domain}
O = Mokee Internal
OU = Nginx Manager
[req_ext]
subjectAltName = @alt_names
[alt_names]
DNS.1 = {domain}
"""
class SSLService:
"""SSL 证书管理服务"""
def __init__(self, ca_dir='/etc/nginx/ssl/ca', ssl_dir='/etc/nginx/ssl'):
self.ca_dir = ca_dir
self.ssl_dir = ssl_dir
self.ca_key = os.path.join(ca_dir, 'ca.key')
self.ca_crt = os.path.join(ca_dir, 'ca.crt')
self.ca_srl = os.path.join(ca_dir, 'ca.srl')
def init_ca(self):
"""
初始化内部 CA
首次运行时自动生成根证书，之后重复使用
"""
os.makedirs(self.ca_dir, exist_ok=True)
if os.path.exists(self.ca_key) and os.path.exists(self.ca_crt):
logger.info("内部 CA 已存在，跳过初始化")
return
logger.info("正在生成内部根 CA...")
# 生成 CA 私钥
subprocess.run([
'openssl', 'genrsa',
'-out', self.ca_key,
'2048'
], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
# 生成自签名 CA 证书
subprocess.run([
'openssl', 'req',
'-x509',
'-new',
'-nodes',
'-key', self.ca_key,
'-sha256',
'-days', str(CERT_VALIDITY_DAYS * 10), # CA 有效期 10 年
'-out', self.ca_crt,
'-subj', '/CN=Mokee Internal CA/O=Mokee/OU=Nginx Manager'
], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
# 设置适当的权限
os.chmod(self.ca_key, 0o600)
os.chmod(self.ca_crt, 0o644)
logger.info(f"内部 CA 已生成: {self.ca_crt}")
logger.info("客户端需要信任此 CA 证书才能避免浏览器警告")
def ensure_certificate(self, domain):
"""
确保域名拥有有效的 SSL 证书
如果不存在则自动签发
Args:
domain: 域名
Returns:
tuple: (证书路径, 私钥路径)
"""
cert_path = os.path.join(self.ssl_dir, f'{domain}.pem')
key_path = os.path.join(self.ssl_dir, f'{domain}.key')
# 确保证书目录存在
os.makedirs(self.ssl_dir, exist_ok=True)
# 如果证书已存在且未过期，直接复用
if os.path.exists(cert_path) and os.path.exists(key_path):
if self._is_cert_valid(cert_path):
logger.info(f"域名 {domain} 的证书已存在且有效，复用")
return cert_path, key_path
else:
logger.info(f"域名 {domain} 的证书已过期，重新签发")
# 签发新证书
self._issue_certificate(domain, cert_path, key_path)
return cert_path, key_path
def _issue_certificate(self, domain, cert_path, key_path):
"""
使用内部 CA 签发域名证书
Args:
domain: 域名
cert_path: 证书输出路径
key_path: 私钥输出路径
"""
# 确保证书目录已创建
os.makedirs(self.ssl_dir, exist_ok=True)
# 创建临时 OpenSSL 配置文件
cnf_path = os.path.join(self.ssl_dir, f'{domain}.cnf')
with open(cnf_path, 'w') as f:
f.write(OPENSSL_CNF_TEMPLATE.format(domain=domain))
try:
# 生成域名私钥
subprocess.run([
'openssl', 'genrsa',
'-out', key_path,
'2048'
], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
# 生成 CSR 并使用 CA 签发
subprocess.run([
'openssl', 'req',
'-new',
'-key', key_path,
'-out', os.path.join(self.ssl_dir, f'{domain}.csr'),
'-config', cnf_path
], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
# 使用 CA 签发证书
subprocess.run([
'openssl', 'x509',
'-req',
'-in', os.path.join(self.ssl_dir, f'{domain}.csr'),
'-CA', self.ca_crt,
'-CAkey', self.ca_key,
'-CAcreateserial',
'-out', cert_path,
'-days', str(CERT_VALIDITY_DAYS),
'-sha256',
'-extensions', 'req_ext',
'-extfile', cnf_path
], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
# 设置权限
os.chmod(key_path, 0o600)
os.chmod(cert_path, 0o644)
# 清理临时文件
csr_path = os.path.join(self.ssl_dir, f'{domain}.csr')
if os.path.exists(csr_path):
os.remove(csr_path)
if os.path.exists(cnf_path):
os.remove(cnf_path)
logger.info(f"已为域名 {domain} 签发 SSL 证书: {cert_path}")
except subprocess.CalledProcessError as e:
logger.error(f"签发证书失败: {e.stderr.decode() if e.stderr else str(e)}")
raise RuntimeError(f"签发SSL证书失败: {e}")
except Exception as e:
# 清理可能残留的临时文件
for tmp in [cnf_path, os.path.join(self.ssl_dir, f'{domain}.csr')]:
if os.path.exists(tmp):
os.remove(tmp)
raise
def _is_cert_valid(self, cert_path):
"""
检查证书是否有效且未过期
Args:
cert_path: 证书文件路径
Returns:
bool: 是否有效
"""
try:
result = subprocess.run([
'openssl', 'x509',
'-in', cert_path,
'-noout',
'-enddate'
], stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, timeout=5)
if result.returncode != 0:
return False
# 解析过期日期
# 输出格式: notAfter=Dec 31 23:59:59 2025 GMT
output = result.stdout.strip()
if 'notAfter=' in output:
date_str = output.split('notAfter=')[1].strip()
expire_date = datetime.strptime(date_str, '%b %d %H:%M:%S %Y %Z')
return datetime.now() < expire_date
return False
except Exception as e:
logger.warning(f"检查证书有效性失败: {e}")
return False
def get_cert_info(self, domain):
"""
获取域名证书信息
Args:
domain: 域名
Returns:
dict | None: 证书信息
"""
cert_path = os.path.join(self.ssl_dir, f'{domain}.pem')
key_path = os.path.join(self.ssl_dir, f'{domain}.key')
if not os.path.exists(cert_path):
return None
info = {
'domain': domain,
'cert_path': cert_path,
'key_path': key_path,
'valid': self._is_cert_valid(cert_path)
}
# 获取过期日期
try:
result = subprocess.run([
'openssl', 'x509',
'-in', cert_path,
'-noout',
'-enddate'
], stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, timeout=5)
if result.returncode == 0 and 'notAfter=' in result.stdout:
info['expire_date'] = result.stdout.split('notAfter=')[1].strip()
except Exception:
pass
return info
def revoke_certificate(self, domain):
"""
删除域名证书（软删除，实际保留文件但标记无效）
Args:
domain: 域名
Returns:
bool: 是否成功
"""
cert_path = os.path.join(self.ssl_dir, f'{domain}.pem')
key_path = os.path.join(self.ssl_dir, f'{domain}.key')
removed = False
for path in [cert_path, key_path]:
if os.path.exists(path):
os.remove(path)
logger.info(f"已删除: {path}")
removed = True
return removed
# ═══════════════════════════════════════════════════════════════
# Let's Encrypt 证书管理（acme.sh + 阿里云 DNS）
# ═══════════════════════════════════════════════════════════════
@staticmethod
def is_zgitm_domain(domain):
"""判断域名是否属于 zgitm.com（含所有子域名），应使用 Let's Encrypt 证书"""
return domain == LE_ROOT_DOMAIN or domain.endswith('.' + LE_ROOT_DOMAIN)
@staticmethod
def is_system_domain(domain):
"""判断是否为系统保护的根域名（zgitm.com / www.zgitm.com），不可在页面创建"""
return domain == LE_ROOT_DOMAIN or domain == 'www.' + LE_ROOT_DOMAIN
@staticmethod
def get_le_root_domain():
"""获取 Let's Encrypt 根域名"""
return LE_ROOT_DOMAIN
def ensure_le_certificate(self, domain):
"""
确保 Let's Encrypt 证书存在且有效
- 已有有效证书 → 直接返回路径
- 证书不存在或过期 → 调用 acme.sh 签发
Args:
domain: 域名
Returns:
tuple: (证书路径, 私钥路径) 或 (None, None) 失败时
"""
cert_path = os.path.join(self.ssl_dir, f'{domain}.pem')
key_path = os.path.join(self.ssl_dir, f'{domain}.key')
os.makedirs(self.ssl_dir, exist_ok=True)
# 已有有效证书，直接复用
if os.path.exists(cert_path) and os.path.exists(key_path):
if self._is_cert_valid(cert_path):
logger.info(f"LE 证书已存在且有效: {domain}")
return cert_path, key_path
else:
logger.info(f"LE 证书已过期，重新签发: {domain}")
# 签发新证书
success = self._issue_le_certificate(domain)
if success:
return cert_path, key_path
return None, None
def _issue_le_certificate(self, domain):
"""
使用 acme.sh DNS-01 方式签发 Let's Encrypt 证书
通过阿里云 DNS API 自动添加 TXT 验证记录
Args:
domain: 域名
Returns:
bool: 是否成功
"""
# 读取阿里云凭据（从环境变量，安全敏感信息不能硬编码）
ali_key = os.environ.get('Ali_Key', '')
ali_secret = os.environ.get('Ali_Secret', '')
if not ali_key or not ali_secret:
logger.error("缺少阿里云 AccessKey 环境变量 Ali_Key / Ali_Secret")
return False
logger.info(f"开始签发 LE 证书: {domain}")
try:
# Step 1: acme.sh 签发（DNS-01 挑战）
# acme.sh DNS-01 签发（env vars 从 systemd 继承）
log_file = f'/tmp/acme_{domain}.log'
args = [ACME_SH, '--issue', '--dns', 'dns_ali', '-d', domain,
'-k', 'ec-256', '--home', '/root/.acme.sh',
'--server', 'letsencrypt', '--log', log_file]
if LE_STAGING:
args.append('--staging')
result = subprocess.run(
args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
timeout=180
)
combined_output = result.stdout + result.stderr
if 'Cert success' not in combined_output:
# 读取 acme.sh 详细日志
try:
with open(log_file, 'r') as f:
debug_log = f.read()
except Exception:
debug_log = '(log file not found)'
logger.error(f"acme.sh 签发失败 ({domain}): "
f"rc={result.returncode} stdout={result.stdout[-200:]} "
f"stderr={result.stderr[-200:]} log={debug_log[-500:]}")
return False
logger.info(f"acme.sh 签发成功: {domain}")
# Step 2: 安装证书到 Nginx SSL 目录
cert_file = os.path.join(self.ssl_dir, f'{domain}.pem')
key_file = os.path.join(self.ssl_dir, f'{domain}.key')
install_result = subprocess.run(
[ACME_SH, '--install-cert', '-d', domain,
'--key-file', key_file, '--fullchain-file', cert_file,
'--reloadcmd', 'systemctl reload nginx',
'--home', '/root/.acme.sh'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=True, timeout=30)
if install_result.returncode != 0:
logger.error(f"安装证书失败: {install_result.stderr[-300:]}")
return False
# 设置权限
os.chmod(key_file, 0o600)
os.chmod(cert_file, 0o644)
logger.info(f"LE 证书已安装: {cert_file}")
return True
except subprocess.TimeoutExpired:
logger.error(f"acme.sh 签发超时 (180s): {domain}")
return False
except Exception as e:
logger.error(f"签发 LE 证书异常: {e}")
return False
def renew_le_certificate(self, domain):
"""
手动续期 Let's Encrypt 证书
acme.sh 有内置 cron 自动续期，此方法用于手动触发
Args:
domain: 域名
Returns:
dict: {success: bool, message: str, log: str}
"""
logger.info(f"手动续期 LE 证书: {domain}")
try:
ali_key = os.environ.get('Ali_Key', '')
ali_secret = os.environ.get('Ali_Secret', '')
args = [ACME_SH, '--renew', '-d', domain, '--force',
'--home', '/root/.acme.sh', '--server', 'letsencrypt']
if LE_STAGING:
args.append('--staging')
result = subprocess.run(
args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
timeout=180
)
output = (result.stdout + result.stderr)[-2000:]
success = 'Cert success' in (result.stdout + result.stderr)
if success:
logger.info(f"续期成功: {domain}")
# 更新证书文件权限
cert_file = os.path.join(self.ssl_dir, f'{domain}.pem')
key_file = os.path.join(self.ssl_dir, f'{domain}.key')
if os.path.exists(cert_file):
os.chmod(cert_file, 0o644)
if os.path.exists(key_file):
os.chmod(key_file, 0o600)
return {'success': True, 'message': f'{domain} 续期成功', 'log': output}
else:
logger.error(f"续期失败: {domain}\n{output}")
return {'success': False, 'message': f'{domain} 续期失败', 'log': output}
except subprocess.TimeoutExpired:
return {'success': False, 'message': '续期超时 (180s)', 'log': ''}
except Exception as e:
return {'success': False, 'message': str(e), 'log': ''}
def get_le_cert_info(self, domain):
"""
获取 Let's Encrypt 证书详细信息
Args:
domain: 域名
Returns:
dict: 证书信息，包含过期日期、剩余天数等
"""
cert_path = os.path.join(self.ssl_dir, f'{domain}.pem')
if not os.path.exists(cert_path):
return {'domain': domain, 'exists': False, 'error': '证书文件不存在'}
info = {
'domain': domain,
'exists': True,
'cert_path': cert_path,
'valid': self._is_cert_valid(cert_path)
}
# 获取过期日期
try:
result = subprocess.run([
'openssl', 'x509', '-in', cert_path, '-noout', '-enddate'
], stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=True, timeout=10)
if 'notAfter=' in result.stdout:
info['expire_date_str'] = result.stdout.split('notAfter=')[1].strip()
expire_date = datetime.strptime(info['expire_date_str'], '%b %d %H:%M:%S %Y %Z')
info['expire_date'] = expire_date.isoformat()
info['days_left'] = (expire_date - datetime.now()).days
info['expiring_soon'] = 0 <= info['days_left'] <= 30
except Exception as e:
info['parse_error'] = str(e)
# 获取签发者
try:
result = subprocess.run([
'openssl', 'x509', '-in', cert_path, '-noout', '-issuer'
], stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=True, timeout=10)
info['issuer'] = result.stdout.strip().replace('issuer=', '')
except Exception:
pass
# 获取主题
try:
result = subprocess.run([
'openssl', 'x509', '-in', cert_path, '-noout', '-subject'
], stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=True, timeout=10)
info['subject'] = result.stdout.strip().replace('subject=', '')
except Exception:
pass
return info