533 lines
18 KiB
Python
533 lines
18 KiB
Python
"""
|
||
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
|