"""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})"