283 lines
10 KiB
Python
283 lines
10 KiB
Python
"""SSL 监控服务：站点证书探测 + 子域名管理
|
||
|
||
- probe_site_cert: 通过 ssl socket 探测站点证书信息（不校验证书链，仅读取到期信息）
|
||
- create/check/check_all: 站点证书的创建与刷新
|
||
- 子域名 CRUD：list/create/update/delete
|
||
"""
|
||
|
||
import socket
|
||
import ssl
|
||
from datetime import date, datetime
|
||
from typing import Optional
|
||
|
||
from cryptography import x509
|
||
from cryptography.hazmat.primitives import hashes
|
||
from sqlmodel import Session, select
|
||
|
||
from app.core.timeutils import utcnow
|
||
from app.models.asset import Asset
|
||
from app.models.ssl import SiteCert, Subdomain
|
||
|
||
PROBE_TIMEOUT = 8 # 探测连接超时（秒）
|
||
EXPIRING_THRESHOLD = 30 # 到期提醒阈值（天）
|
||
|
||
|
||
def days_until(d: Optional[date]) -> Optional[int]:
|
||
"""计算距今天数（负数为已过期）"""
|
||
if d is None:
|
||
return None
|
||
return (d - date.today()).days
|
||
|
||
|
||
def _judge_status(days: Optional[int]) -> str:
|
||
"""按剩余天数定级：expired / expiring / valid"""
|
||
if days is None:
|
||
return "unknown"
|
||
if days < 0:
|
||
return "expired"
|
||
if days <= EXPIRING_THRESHOLD:
|
||
return "expiring"
|
||
return "valid"
|
||
|
||
|
||
def probe_site_cert(hostname: str, port: int = 443) -> dict:
|
||
"""探测目标站点证书信息（探测失败时抛异常）
|
||
|
||
返回字段：subject_cn / issuer / valid_from / valid_to / fingerprint / san_list
|
||
"""
|
||
ctx = ssl.create_default_context()
|
||
# 不校验证书链：即使证书已过期/自签名也能读到到期信息
|
||
ctx.check_hostname = False
|
||
ctx.verify_mode = ssl.CERT_NONE
|
||
|
||
with socket.create_connection((hostname, port), timeout=PROBE_TIMEOUT) as sock:
|
||
with ctx.wrap_socket(sock, server_hostname=hostname) as ssock:
|
||
der = ssock.getpeercert(binary_form=True)
|
||
cert = x509.load_der_x509_certificate(der)
|
||
try:
|
||
san = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName)
|
||
san_list = san.value.get_values_for_type(x509.DNSName)
|
||
except x509.ExtensionNotFound:
|
||
san_list = []
|
||
return {
|
||
"subject_cn": cert.subject.rfc4514_string() or None,
|
||
"issuer": cert.issuer.rfc4514_string() or None,
|
||
"valid_from": cert.not_valid_before_utc.date(),
|
||
"valid_to": cert.not_valid_after_utc.date(),
|
||
"fingerprint": cert.fingerprint(hashes.SHA256()).hex(),
|
||
"san_list": san_list,
|
||
}
|
||
|
||
|
||
def _to_dict(cert: SiteCert, asset_name: Optional[str] = None) -> dict:
|
||
"""SiteCert 模型转 API 返回结构（含动态计算的剩余天数）"""
|
||
days = days_until(cert.valid_to)
|
||
return {
|
||
"id": cert.id,
|
||
"hostname": cert.hostname,
|
||
"port": cert.port,
|
||
"asset_id": cert.asset_id,
|
||
"asset_name": asset_name,
|
||
"issuer": cert.issuer,
|
||
"subject_cn": cert.subject_cn,
|
||
"valid_from": cert.valid_from.isoformat() if cert.valid_from else None,
|
||
"valid_to": cert.valid_to.isoformat() if cert.valid_to else None,
|
||
"days_to_expiry": days,
|
||
"status": cert.status,
|
||
"error": cert.error,
|
||
"last_checked_at": cert.last_checked_at.isoformat() if cert.last_checked_at else None,
|
||
}
|
||
|
||
|
||
# ---------------- 子域名 CRUD ----------------
|
||
|
||
def list_subdomains(session: Session, asset_id: Optional[int] = None) -> list:
|
||
stmt = select(Subdomain).order_by(Subdomain.host)
|
||
if asset_id is not None:
|
||
stmt = stmt.where(Subdomain.asset_id == asset_id)
|
||
subs = session.exec(stmt).all()
|
||
# 带上所属域名，便于前端展示全名
|
||
assets = session.exec(select(Asset).where(Asset.id.in_({s.asset_id for s in subs}))).all()
|
||
name_map = {a.id: a.name for a in assets}
|
||
return [
|
||
{
|
||
"id": s.id,
|
||
"asset_id": s.asset_id,
|
||
"asset_name": name_map.get(s.asset_id),
|
||
"host": s.host,
|
||
"record_type": s.record_type,
|
||
"record_value": s.record_value,
|
||
"is_active": s.is_active,
|
||
"note": s.note,
|
||
"created_at": s.created_at.isoformat() if s.created_at else None,
|
||
}
|
||
for s in subs
|
||
]
|
||
|
||
|
||
def create_subdomain(session: Session, data: dict) -> Subdomain:
|
||
sub = Subdomain(
|
||
asset_id=data["asset_id"],
|
||
host=data["host"].strip().lower(),
|
||
record_type=data.get("record_type"),
|
||
record_value=data.get("record_value"),
|
||
is_active=data.get("is_active", True),
|
||
note=data.get("note"),
|
||
)
|
||
session.add(sub)
|
||
session.commit()
|
||
session.refresh(sub)
|
||
return sub
|
||
|
||
|
||
def update_subdomain(session: Session, sub_id: int, data: dict) -> Subdomain:
|
||
sub = session.get(Subdomain, sub_id)
|
||
if not sub:
|
||
raise ValueError(f"子域名记录不存在（id={sub_id}）")
|
||
if "host" in data and data["host"]:
|
||
sub.host = data["host"].strip().lower()
|
||
for key in ("record_type", "record_value", "note"):
|
||
if key in data:
|
||
setattr(sub, key, data[key])
|
||
if "is_active" in data:
|
||
sub.is_active = bool(data["is_active"])
|
||
session.add(sub)
|
||
session.commit()
|
||
session.refresh(sub)
|
||
return sub
|
||
|
||
|
||
def delete_subdomain(session: Session, sub_id: int) -> None:
|
||
sub = session.get(Subdomain, sub_id)
|
||
if not sub:
|
||
raise ValueError(f"子域名记录不存在（id={sub_id}）")
|
||
session.delete(sub)
|
||
session.commit()
|
||
|
||
|
||
# ---------------- 站点证书监控 ----------------
|
||
|
||
def _apply_probe(cert: SiteCert, info: dict) -> SiteCert:
|
||
"""把探测结果写入模型并定级"""
|
||
cert.subject_cn = info["subject_cn"]
|
||
cert.issuer = info["issuer"]
|
||
cert.valid_from = info["valid_from"]
|
||
cert.valid_to = info["valid_to"]
|
||
cert.fingerprint = info["fingerprint"]
|
||
cert.error = None
|
||
cert.status = _judge_status(days_until(info["valid_to"]))
|
||
cert.last_checked_at = utcnow()
|
||
return cert
|
||
|
||
|
||
def check_one(session: Session, cert: SiteCert) -> SiteCert:
|
||
"""重新探测单条证书记录（失败则标记 error，保留旧到期信息）"""
|
||
try:
|
||
info = probe_site_cert(cert.hostname, cert.port)
|
||
_apply_probe(cert, info)
|
||
except Exception as e: # noqa: BLE001
|
||
cert.status = "error"
|
||
cert.error = str(e)[:200]
|
||
cert.last_checked_at = utcnow()
|
||
session.add(cert)
|
||
session.commit()
|
||
session.refresh(cert)
|
||
return cert
|
||
|
||
|
||
def create_site_cert(session: Session, hostname: str, port: int = 443, asset_id: Optional[int] = None) -> SiteCert:
|
||
"""创建探测目标并立即探测一次；hostname+port 已存在则复用并刷新"""
|
||
hostname = hostname.strip().lower()
|
||
existing = session.exec(
|
||
select(SiteCert).where(SiteCert.hostname == hostname, SiteCert.port == port)
|
||
).first()
|
||
if existing:
|
||
if asset_id is not None:
|
||
existing.asset_id = asset_id
|
||
return check_one(session, existing)
|
||
cert = SiteCert(hostname=hostname, port=port, asset_id=asset_id)
|
||
session.add(cert)
|
||
session.commit()
|
||
session.refresh(cert)
|
||
return check_one(session, cert)
|
||
|
||
|
||
def list_site_certs(session: Session, status: Optional[str] = None, asset_id: Optional[int] = None) -> list:
|
||
stmt = select(SiteCert)
|
||
if status:
|
||
stmt = stmt.where(SiteCert.status == status)
|
||
if asset_id is not None:
|
||
stmt = stmt.where(SiteCert.asset_id == asset_id)
|
||
certs = session.exec(stmt.order_by(SiteCert.id)).all()
|
||
ids = {c.asset_id for c in certs if c.asset_id}
|
||
assets = session.exec(select(Asset).where(Asset.id.in_(ids))).all() if ids else []
|
||
name_map = {a.id: a.name for a in assets}
|
||
return [_to_dict(c, name_map.get(c.asset_id)) for c in certs]
|
||
|
||
|
||
def delete_site_cert(session: Session, cert_id: int) -> None:
|
||
cert = session.get(SiteCert, cert_id)
|
||
if not cert:
|
||
raise ValueError(f"证书监控记录不存在（id={cert_id}）")
|
||
session.delete(cert)
|
||
session.commit()
|
||
|
||
|
||
def check_all_site_certs(session: Session) -> dict:
|
||
"""全量刷新所有站点证书，返回统计与异常清单
|
||
|
||
探测为纯网络 IO（单次最长 8s），用线程池并发探测后统一写库：
|
||
串行时 N 个站点最坏耗时 N×8s，并发后接近单站点耗时；
|
||
写库集中在主线程一次 commit（原来逐条 commit 产生 N 次事务）。
|
||
"""
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
|
||
certs = session.exec(select(SiteCert)).all()
|
||
stats = {"total": len(certs), "ok": 0, "error": 0, "expiring": 0, "expired": 0}
|
||
problems = []
|
||
if not certs:
|
||
return {"stats": stats, "problems": problems}
|
||
|
||
# 并发探测（不碰数据库，线程安全）
|
||
with ThreadPoolExecutor(max_workers=min(8, len(certs))) as pool:
|
||
futures = {pool.submit(probe_site_cert, c.hostname, c.port): c for c in certs}
|
||
for future in futures:
|
||
cert = futures[future]
|
||
try:
|
||
_apply_probe(cert, future.result())
|
||
except Exception as e: # noqa: BLE001
|
||
cert.status = "error"
|
||
cert.error = str(e)[:200]
|
||
cert.last_checked_at = utcnow()
|
||
session.add(cert)
|
||
session.commit()
|
||
|
||
for cert in certs:
|
||
session.refresh(cert)
|
||
if cert.status == "error":
|
||
stats["error"] += 1
|
||
problems.append({"hostname": cert.hostname, "detail": cert.error})
|
||
elif cert.status == "expired":
|
||
stats["expired"] += 1
|
||
problems.append(
|
||
{"hostname": cert.hostname, "detail": f"证书已过期 {abs(days_until(cert.valid_to))} 天"}
|
||
)
|
||
elif cert.status == "expiring":
|
||
stats["expiring"] += 1
|
||
problems.append(
|
||
{"hostname": cert.hostname, "detail": f"{days_until(cert.valid_to)} 天后到期"}
|
||
)
|
||
else:
|
||
stats["ok"] += 1
|
||
return {"stats": stats, "problems": problems}
|
||
|
||
|
||
def build_cert_message(cert: SiteCert) -> str:
|
||
"""生成单条证书提醒文案"""
|
||
days = days_until(cert.valid_to)
|
||
if days is None:
|
||
return f"- {cert.hostname}：证书信息未知"
|
||
if days < 0:
|
||
return f"- ⚠️ {cert.hostname}：证书已过期 {abs(days)} 天"
|
||
return f"- {cert.hostname}：{days} 天后到期（{cert.valid_to}）"
|