219 lines
9.2 KiB
Python
219 lines
9.2 KiB
Python
#!/usr/bin/python3
|
||
"""校验证书并同步至 Smail，私钥与回退副本仅在服务器保存。"""
|
||
|
||
from datetime import datetime, timezone
|
||
from contextlib import contextmanager
|
||
import fcntl
|
||
import hashlib
|
||
import os
|
||
from pathlib import Path
|
||
import pwd
|
||
import re
|
||
import shlex
|
||
import socket
|
||
import ssl
|
||
import stat
|
||
import subprocess
|
||
import time
|
||
import uuid
|
||
|
||
|
||
def configured_domain(name):
|
||
value = os.environ[name].strip().lower()
|
||
labels = value.split('.')
|
||
if (len(value) > 253 or len(labels) < 2
|
||
or any(not re.fullmatch(r'[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?', label)
|
||
for label in labels)):
|
||
raise ValueError(name)
|
||
return value
|
||
|
||
|
||
MAIL_DOMAIN = configured_domain('SMAIL_MAIL_DOMAIN')
|
||
CERTIFICATE_SNI = configured_domain('SMAIL_WEB_DOMAIN')
|
||
if (CERTIFICATE_SNI != MAIL_DOMAIN
|
||
and (not CERTIFICATE_SNI.endswith('.' + MAIL_DOMAIN)
|
||
or CERTIFICATE_SNI.count('.') != MAIL_DOMAIN.count('.') + 1)):
|
||
raise ValueError('SMAIL_WEB_DOMAIN')
|
||
DOMAINS = (MAIL_DOMAIN, '*.' + MAIL_DOMAIN)
|
||
VERIFY_HOSTNAMES = (MAIL_DOMAIN, CERTIFICATE_SNI,
|
||
*(prefix + '.' + MAIL_DOMAIN for prefix in ('mail', 'smtp', 'imap', 'pop')))
|
||
LINEAGE = Path('/var/lib/smail-acme-tls/certificates')
|
||
TARGET = Path('/opt/smail/ssl')
|
||
BACKUPS = Path('/var/backups')
|
||
LOCK = Path('/run/smail-certificate-renew.lock')
|
||
CA_FILE = '/etc/ssl/certs/ca-certificates.crt'
|
||
LIVE_PORTS = (443, 8002)
|
||
|
||
|
||
def run(*args, **kwargs):
|
||
kwargs.setdefault('timeout', 15)
|
||
return subprocess.run(args, check=True, capture_output=True, **kwargs).stdout
|
||
|
||
|
||
@contextmanager
|
||
def certificate_lock():
|
||
fd = os.open(LOCK, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600)
|
||
with os.fdopen(fd, 'a') as stream:
|
||
info = os.fstat(stream.fileno())
|
||
if (not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_nlink != 1
|
||
or info.st_mode & 0o077):
|
||
raise RuntimeError('证书锁文件不安全')
|
||
fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||
yield
|
||
|
||
|
||
def read_file(path, uid):
|
||
if path.parent.resolve(strict=True) != path.parent:
|
||
raise RuntimeError('证书目录不能包含符号链接')
|
||
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC | os.O_NONBLOCK)
|
||
with os.fdopen(fd, 'rb') as stream:
|
||
info = os.fstat(stream.fileno())
|
||
if (not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or info.st_uid != uid
|
||
or info.st_mode & 0o077 or info.st_size > 1024 * 1024):
|
||
raise RuntimeError('证书文件类型、属主、权限或大小不符合要求')
|
||
data = stream.read(1024 * 1024 + 1)
|
||
if len(data) != info.st_size:
|
||
raise RuntimeError('读取期间证书发生变化')
|
||
return data
|
||
|
||
|
||
def check_nginx_paths():
|
||
configuration = run('nginx', '-T').decode('utf-8')
|
||
found = set()
|
||
expected = {'ssl_certificate': TARGET / 'custom.crt',
|
||
'ssl_certificate_key': TARGET / 'custom.key'}
|
||
for line in configuration.splitlines():
|
||
if not line.lstrip().startswith(('ssl_certificate ', 'ssl_certificate\t',
|
||
'ssl_certificate_key ', 'ssl_certificate_key\t')):
|
||
continue
|
||
tokens = shlex.split(line, comments=True)
|
||
if len(tokens) < 2 or tokens[0] not in expected:
|
||
continue
|
||
value = tokens[1].rstrip(';')
|
||
if value.startswith('/var/lib/smail-acme') or value.startswith(str(LINEAGE) + '/'):
|
||
raise RuntimeError('Nginx 仍引用签发目录，必须先迁移到已校验的线上证书路径')
|
||
if value == str(expected[tokens[0]]):
|
||
found.add(tokens[0])
|
||
if found != set(expected):
|
||
raise RuntimeError('未找到 Nginx 的 Smail 线上证书路径，拒绝发布')
|
||
|
||
|
||
def replace_file(path, data, uid, gid):
|
||
staged = path.with_name(path.name + '.new-' + uuid.uuid4().hex)
|
||
fd = os.open(staged, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
||
with os.fdopen(fd, 'wb') as stream:
|
||
os.fchown(stream.fileno(), uid, gid)
|
||
stream.write(data)
|
||
stream.flush()
|
||
os.fsync(stream.fileno())
|
||
os.replace(staged, path)
|
||
fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
|
||
try:
|
||
os.fsync(fd)
|
||
finally:
|
||
os.close(fd)
|
||
|
||
|
||
def validate(directory):
|
||
names = run('openssl', 'x509', '-in', str(directory / 'candidate.crt'), '-noout',
|
||
'-ext', 'subjectAltName').decode('utf-8')
|
||
certificate_domains = {item.strip()[4:] for line in names.splitlines()
|
||
for item in line.split(',') if item.strip().startswith('DNS:')}
|
||
if not set(DOMAINS).issubset(certificate_domains):
|
||
raise RuntimeError('证书必须同时覆盖根域名和泛域名')
|
||
for domain in VERIFY_HOSTNAMES:
|
||
run('openssl', 'verify', '-CAfile', CA_FILE, '-purpose', 'sslserver',
|
||
'-untrusted', str(directory / 'candidate.issuer.crt'), '-verify_hostname', domain,
|
||
str(directory / 'candidate.crt'))
|
||
run('openssl', 'x509', '-in', str(directory / 'candidate.crt'), '-checkend', '604800', '-noout')
|
||
public_cert = run('openssl', 'x509', '-in', str(directory / 'candidate.crt'), '-pubkey', '-noout')
|
||
public_key = run('openssl', 'pkey', '-in', str(directory / 'candidate.key'), '-pubout')
|
||
if public_cert != public_key:
|
||
raise RuntimeError('证书与私钥不匹配')
|
||
# 普通续期只允许复用密钥，避免两个文件替换之间出现不匹配；密钥轮换另行维护。
|
||
previous_key = run('openssl', 'pkey', '-in', str(directory / 'previous.key'), '-pubout')
|
||
previous_cert = run('openssl', 'x509', '-in', str(directory / 'previous.crt'), '-pubkey', '-noout')
|
||
if public_key != previous_key or previous_key != previous_cert:
|
||
raise RuntimeError('无停机续期必须复用现有密钥，密钥轮换需要单独流程')
|
||
|
||
|
||
def verify_live_certificate(fingerprint):
|
||
context = ssl.create_default_context(cafile=CA_FILE)
|
||
for port in LIVE_PORTS:
|
||
deadline = time.monotonic() + 12
|
||
while True:
|
||
try:
|
||
with socket.create_connection(('127.0.0.1', port), timeout=2) as connection:
|
||
with context.wrap_socket(connection, server_hostname=CERTIFICATE_SNI) as secure:
|
||
actual = hashlib.sha256(secure.getpeercert(binary_form=True)).hexdigest()
|
||
if actual == fingerprint:
|
||
break
|
||
raise RuntimeError('线上入口未返回预期证书')
|
||
except (OSError, RuntimeError):
|
||
if time.monotonic() >= deadline:
|
||
raise RuntimeError('线上证书验证失败，端口=' + str(port)) from None
|
||
time.sleep(0.25)
|
||
|
||
|
||
def fingerprint(path):
|
||
return hashlib.sha256(run('openssl', 'x509', '-in', str(path), '-outform', 'DER')).hexdigest()
|
||
|
||
|
||
def publish():
|
||
check_nginx_paths()
|
||
run('systemctl', 'is-active', '--quiet', 'nginx')
|
||
identity = pwd.getpwnam('smail')
|
||
sources = {'candidate.crt': read_file(LINEAGE / (MAIL_DOMAIN + '.crt'), 0),
|
||
'candidate.key': read_file(LINEAGE / (MAIL_DOMAIN + '.key'), 0),
|
||
'candidate.issuer.crt': read_file(LINEAGE / (MAIL_DOMAIN + '.issuer.crt'), 0),
|
||
'previous.crt': read_file(TARGET / 'custom.crt', identity.pw_uid),
|
||
'previous.key': read_file(TARGET / 'custom.key', identity.pw_uid)}
|
||
stamp = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S') + '-' + uuid.uuid4().hex[:8]
|
||
backup = BACKUPS / ('smail-certificate-' + stamp)
|
||
backup.mkdir(mode=0o700)
|
||
print('证书校验及回退快照=' + str(backup), flush=True)
|
||
for name, data in sources.items():
|
||
replace_file(backup / name, data, 0, 0)
|
||
# 校验与发布使用同一份受保护快照，不重新读取可能变化的签发目录。
|
||
validate(backup)
|
||
expected = fingerprint(backup / 'candidate.crt')
|
||
changed = [suffix for suffix in ('key', 'crt')
|
||
if sources['candidate.' + suffix] != sources['previous.' + suffix]]
|
||
if not changed:
|
||
try:
|
||
verify_live_certificate(expected)
|
||
except RuntimeError:
|
||
# 恢复上次在文件发布后、重载前中断的任务；不能仅因文件相同就宣告成功。
|
||
run('nginx', '-t')
|
||
run('systemctl', 'reload', 'nginx')
|
||
verify_live_certificate(expected)
|
||
print('证书校验通过，线上入口与运行文件已同步')
|
||
return
|
||
try:
|
||
for suffix in changed:
|
||
replace_file(TARGET / ('custom.' + suffix), sources['candidate.' + suffix],
|
||
identity.pw_uid, identity.pw_gid)
|
||
run('nginx', '-t')
|
||
run('systemctl', 'reload', 'nginx')
|
||
verify_live_certificate(expected)
|
||
except Exception:
|
||
for suffix in changed:
|
||
replace_file(TARGET / ('custom.' + suffix), sources['previous.' + suffix],
|
||
identity.pw_uid, identity.pw_gid)
|
||
run('nginx', '-t')
|
||
run('systemctl', 'reload', 'nginx')
|
||
verify_live_certificate(fingerprint(backup / 'previous.crt'))
|
||
raise
|
||
print('证书已同步，SHA256=' + expected)
|
||
|
||
|
||
def main():
|
||
if os.geteuid() != 0:
|
||
raise RuntimeError('必须使用 root 执行证书同步')
|
||
with certificate_lock():
|
||
publish()
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|