diff --git a/app/core/security.py b/app/core/security.py index 8ae618f..30ce499 100644 --- a/app/core/security.py +++ b/app/core/security.py @@ -5,6 +5,7 @@ - 若已配置，则要求请求头携带正确的 X-API-Key，否则返回 401。 """ +import hmac from typing import Optional from fastapi import Header, HTTPException, status @@ -12,13 +13,20 @@ from fastapi import Header, HTTPException, status from app.core.config import settings +def _key_matches(provided: Optional[str], expected: str) -> bool: + """常量时间比较密钥，避免时序旁路泄露密钥长度/前缀信息""" + if not provided: + return False + return hmac.compare_digest(provided.encode(), expected.encode()) + + async def require_api_key( x_api_key: Optional[str] = Header(default=None, alias="X-API-Key"), ) -> None: """校验 API Key（可选启用）""" if not settings.API_KEY: return - if x_api_key != settings.API_KEY: + if not _key_matches(x_api_key, settings.API_KEY): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="API Key 无效或缺失", @@ -31,7 +39,7 @@ async def require_agent_key( """校验 Agent 上报 Key（可选启用）""" if not settings.AGENT_KEY: return - if x_agent_key != settings.AGENT_KEY: + if not _key_matches(x_agent_key, settings.AGENT_KEY): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Agent Key 无效或缺失", diff --git a/app/database.py b/app/database.py index b12eddf..540499a 100644 --- a/app/database.py +++ b/app/database.py @@ -16,26 +16,26 @@ DATA_DIR.mkdir(parents=True, exist_ok=True) ASSETS_DB_URL = f"sqlite:///{DATA_DIR / 'assets.db'}" METRICS_DB_URL = f"sqlite:///{DATA_DIR / 'metrics.db'}" +# timeout：sqlite3 内置 busy 等待（秒），随连接创建生效。 +# SQLite 默认使用 NullPool（每请求新建连接），PRAGMA busy_timeout 只在单连接有效， +# 必须通过 connect_args 传递，否则并发写入（Agent 上报 + 清理任务）会报 database is locked。 assets_engine = create_engine( ASSETS_DB_URL, echo=False, - connect_args={"check_same_thread": False}, - pool_pre_ping=True, + connect_args={"check_same_thread": False, "timeout": 30}, ) metrics_engine = create_engine( METRICS_DB_URL, echo=False, - connect_args={"check_same_thread": False}, - pool_pre_ping=True, + connect_args={"check_same_thread": False, "timeout": 30}, ) def _enable_wal(engine) -> None: - """启用 WAL 模式，提升 SQLite 并发读写能力""" + """启用 WAL 模式，提升 SQLite 并发读写能力（journal_mode 持久化到库文件，设置一次即可）""" import sqlalchemy as sa with engine.connect() as conn: conn.execute(sa.text("PRAGMA journal_mode=WAL")) - conn.execute(sa.text("PRAGMA busy_timeout=5000")) _enable_wal(assets_engine) @@ -152,12 +152,16 @@ def _backfill_provider_services(conn) -> None: def get_session() -> Generator[Session, None, None]: - """资产库会话（默认）""" - with Session(assets_engine) as session: + """资产库会话（默认） + + expire_on_commit=False：commit 后不失效对象属性，避免后续访问触发隐式 + 重新加载查询；需要最新值的场景由调用方显式 session.refresh()。 + """ + with Session(assets_engine, expire_on_commit=False) as session: yield session def get_metrics_session() -> Generator[Session, None, None]: """监控 / 日志库会话""" - with Session(metrics_engine) as session: + with Session(metrics_engine, expire_on_commit=False) as session: yield session diff --git a/app/main.py b/app/main.py index e4ed1ee..dcd6c62 100644 --- a/app/main.py +++ b/app/main.py @@ -35,12 +35,20 @@ def _asset_version() -> str: 用于前端引用 ?v= 参数，绕开浏览器启发式缓存（旧响应无 Cache-Control 时存下的条目会被视为新鲜而不再回源验证）。 + 启动时计算一次并缓存：静态资源只在代码更新时变化，而更新后服务会重启， + 避免每次页面请求都遍历 stat 整个 static 目录。 """ - latest = 0 - for p in STATIC_DIR.rglob("*"): - if p.is_file(): - latest = max(latest, int(p.stat().st_mtime)) - return str(latest) + global _ASSET_VERSION_CACHE + if _ASSET_VERSION_CACHE is None: + latest = 0 + for p in STATIC_DIR.rglob("*"): + if p.is_file(): + latest = max(latest, int(p.stat().st_mtime)) + _ASSET_VERSION_CACHE = str(latest) + return _ASSET_VERSION_CACHE + + +_ASSET_VERSION_CACHE: str | None = None # 全局日志配置：统一格式，便于生产环境排查 logging.basicConfig( diff --git a/app/routers/monitor.py b/app/routers/monitor.py index 87ad476..6ea765a 100644 --- a/app/routers/monitor.py +++ b/app/routers/monitor.py @@ -7,7 +7,7 @@ from sqlmodel import Session, select from app.core.security import require_api_key from app.database import get_metrics_session -from app.models.monitor import MetricPoint, SecurityCheck, ServerInfo +from app.models.monitor import MetricPoint, ServerInfo from app.services import cleanup_service, security_service router = APIRouter(prefix="/api/monitor", tags=["monitor"]) @@ -55,18 +55,7 @@ def info(asset_id: int, session: Session = Depends(get_metrics_session)) -> Opti @router.get("/{asset_id}/security", summary="安全检查项（每项最新一条）") def security(asset_id: int, session: Session = Depends(get_metrics_session)): - stmt = ( - select(SecurityCheck) - .where(SecurityCheck.asset_id == asset_id) - .order_by(SecurityCheck.ts.desc()) - .limit(50) - ) - checks = session.exec(stmt).all() - latest_by_item = {} - for check in checks: - if check.check_item not in latest_by_item: - latest_by_item[check.check_item] = check - return list(latest_by_item.values()) + return security_service.latest_checks(session, asset_id) @router.get("/{asset_id}/security-score", summary="单个服务器安全评分与加固建议") diff --git a/app/services/asset_service.py b/app/services/asset_service.py index 22028de..df6eb92 100644 --- a/app/services/asset_service.py +++ b/app/services/asset_service.py @@ -283,14 +283,19 @@ def delete_asset(session: Session, asset_id: int) -> None: def _cleanup_metrics_for_asset(asset_id: int) -> None: - """清理 metrics.db 中该资产的 MetricPoint/ServerInfo/SecurityCheck/EventLog""" + """清理 metrics.db 中该资产的 MetricPoint/ServerInfo/SecurityCheck/EventLog + + 使用批量 DELETE（而非逐行加载后删除）：监控时序数据可能上万行， + 逐行删除会全部载入内存且产生数万次 ORM 操作。 + """ + from sqlmodel import delete + from app.database import metrics_engine from app.models.monitor import EventLog, MetricPoint, SecurityCheck, ServerInfo with Session(metrics_engine) as ms: for model in (MetricPoint, ServerInfo, SecurityCheck, EventLog): - for row in ms.exec(select(model).where(model.asset_id == asset_id)).all(): - ms.delete(row) + ms.exec(delete(model).where(model.asset_id == asset_id)) ms.commit() diff --git a/app/services/security_service.py b/app/services/security_service.py index a0587ea..8b9dc04 100644 --- a/app/services/security_service.py +++ b/app/services/security_service.py @@ -15,19 +15,26 @@ DEFAULT_WEIGHT = 20 def latest_checks(session: Session, asset_id: int) -> list: - """取每个检查项的最新一条""" - stmt = ( - select(SecurityCheck) + """取每个检查项的最新一条 + + 用 GROUP BY max(ts) 子查询精确取每项最新记录：若用 limit(50) 后去重， + 当某个检查项连续上报超过 50 次时会把其他项的最新记录挤出，导致评分失真。 + """ + latest_ts = ( + select( + SecurityCheck.check_item, + func.max(SecurityCheck.ts).label("max_ts"), + ) .where(SecurityCheck.asset_id == asset_id) - .order_by(SecurityCheck.ts.desc()) - .limit(50) + .group_by(SecurityCheck.check_item) + .subquery() ) - checks = session.exec(stmt).all() - latest = {} - for check in checks: - if check.check_item not in latest: - latest[check.check_item] = check - return list(latest.values()) + stmt = select(SecurityCheck).where(SecurityCheck.asset_id == asset_id).join( + latest_ts, + (SecurityCheck.check_item == latest_ts.c.check_item) + & (SecurityCheck.ts == latest_ts.c.max_ts), + ) + return list(session.exec(stmt).all()) def compute_security_score(checks: list): diff --git a/app/services/ssl_service.py b/app/services/ssl_service.py index e4697da..e2b7fcf 100644 --- a/app/services/ssl_service.py +++ b/app/services/ssl_service.py @@ -224,12 +224,36 @@ def delete_site_cert(session: Session, cert_id: int) -> None: 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: - check_one(session, cert) + session.refresh(cert) if cert.status == "error": stats["error"] += 1 problems.append({"hostname": cert.hostname, "detail": cert.error}) diff --git a/app/services/sync_service.py b/app/services/sync_service.py index 11e6402..825cded 100644 --- a/app/services/sync_service.py +++ b/app/services/sync_service.py @@ -299,26 +299,44 @@ def refresh_ai_balance(session: Session, asset_id: int) -> dict: } -def sync_all_ai_balances(session: Session) -> dict: - """遍历所有 AI 账号资产，逐个刷新余额（单个失败不中断整体）""" +def sync_all_ai_balances(session: Session, max_workers: int = 5) -> dict: + """并发刷新所有 AI 账号余额（单个失败不中断整体） + + 每个账号需调用外部 API（单次最长 30s），串行时总耗时随账号数线性增长； + 改为线程池并发后显著提速。注意：SQLite Session 不能跨线程共享， + 每个 worker 使用独立 Session（WAL 模式下多连接读写安全）。 + """ + from concurrent.futures import ThreadPoolExecutor + + from app.database import assets_engine + ai_assets = session.exec( select(Asset).where(Asset.asset_type == AssetType.AI_AGENT) ).all() + asset_ids = [(a.id, a.name) for a in ai_assets] + + def _refresh_one(asset_id: int): + with Session(assets_engine) as worker_session: + refresh_ai_balance(worker_session, asset_id) + success = 0 failed = 0 errors = [] - for asset in ai_assets: - try: - refresh_ai_balance(session, asset.id) - success += 1 - except HTTPException as e: - failed += 1 - errors.append(f"{asset.name}: {e.detail}") - except Exception as e: # noqa: BLE001 - failed += 1 - errors.append(f"{asset.name}: {e}") + with ThreadPoolExecutor(max_workers=max_workers) as pool: + futures = {pool.submit(_refresh_one, aid): name for aid, name in asset_ids} + for future in futures: + name = futures[future] + try: + future.result() + success += 1 + except HTTPException as e: + failed += 1 + errors.append(f"{name}: {e.detail}") + except Exception as e: # noqa: BLE001 + failed += 1 + errors.append(f"{name}: {e}") return { - "total": len(ai_assets), + "total": len(asset_ids), "success": success, "failed": failed, "errors": errors,