Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 144431b300 | |||
| c0cb292350 | |||
| da8496adfd | |||
| 5d3c895542 | |||
| 1222238bf3 | |||
| a8f5733bb7 |
32
README.md
32
README.md
@@ -1,6 +1,6 @@
|
||||
# MeshSafe · 文件安全检测服务
|
||||
|
||||

|
||||

|
||||

|
||||

|
||||
|
||||
@@ -200,6 +200,36 @@ MeshSafe 会**分析宏的实际行为**:
|
||||
|
||||
## 版本历史
|
||||
|
||||
### v0.2.3 (2026-10-29)
|
||||
|
||||
**Bug 修复:**
|
||||
|
||||
- **Policy 加载失败(每次启动都回退 YAML)** — `policy.py _load_db_or_yaml()` 用 `scalar_one_or_none()` 加载最新策略版本,但 `policy_versions` 表有多条历史版本 → SQLAlchemy 抛 `Multiple rows` → 永远回退 YAML 配置文件,导致不同重启间规则数波动。修复为 `db.query(PolicyVersion).order_by(created_at.desc()).first()`,直接取最新一条 ORM 对象
|
||||
- **控制台任务历史被内部测试记录淹没** — `scan_tasks` 表累积 5,800+ 条内部对账/诊断脚本产生的测试数据(channel 包括 `test_center` / `v` / `verify` / `diag` / `debug` / `final` 等 19 个),dashboard 和 tasks_page 无过滤全部展示。修复:① DB 清理所有内部 channel 记录(保留 3,488 条生产数据);② console.py 加 `INTERNAL_TEST_CHANNELS` 黑名单常量,dashboard 6 个查询 + tasks_page 2 个查询全部 `.filter(ScanTask.channel.notin_(...))`
|
||||
- **测试靶场显示 "未知" 层和 "—" 编号** — 控制台用 `iterdir()` 遍历目录收集样本,但目录里混入 4 个 YARA 调试时的临时文件(`L4-005_hotreload.txt` 等),index.json 没登记 → category 取不到显示 "未知"。修复:① 清理 4 个临时文件;② `test_center_page` 改为只从 index.json 收集样本(目录遍历替换为 index.json 遍历),以后不会再出这个问题
|
||||
|
||||
**DB 与运维增强:**
|
||||
|
||||
- **ScanTask 复合索引** — 新增 `ix_scan_tasks_channel_created ON (channel, created_at DESC)`,解决 dashboard/tasks_page 高频查询的全表扫描。models.py `ScanTask.__table_args__` 也同步更新,新建实例会自动带上
|
||||
- **归档表 + 归档脚本** — 建 `scan_tasks_archive` 表(含 4 个索引);新增 `scripts/archive_old_tasks.py` 支持 4 项清理策略:
|
||||
- scan_tasks **90 天前**:清 detail JSONB(每条 1-5 KB,省空间)
|
||||
- scan_tasks **365 天前**:移到 scan_tasks_archive 归档表(CASCADE 删 ScanFinding)
|
||||
- idempotency_records:清 **7 天前**(幂等缓存过期可以重扫)
|
||||
- policy_versions:只留**最新 5 个**(历史版本无保留价值)
|
||||
- 支持 `--dry-run`(默认,只报告不动数据)+ `--apply`(真正执行),分批 1,000 条处理避免长事务锁表
|
||||
- **自动定时任务 Cron 服务** — docker-compose.yml 新增 `cron` 服务(共享 meshsafe:latest 镜像,零额外依赖,`restart: unless-stopped`),`scripts/cron_entrypoint.py` 纯 Python 调度器:
|
||||
- **每天 03:00** 跑 archive_old_tasks.py --apply
|
||||
- **每周一 03:10** VACUUM ANALYZE 5 张表(维护表统计信息,加速查询)
|
||||
|
||||
**数据增长实况:**
|
||||
|
||||
| 指标 | 值 |
|
||||
|------|---|
|
||||
| scan_tasks 当前 | 3,660 行 / 4.5 MB |
|
||||
| 日均新增 | ~523 条 |
|
||||
| 年增长 | ~19 万条 ≈ 910 MB |
|
||||
| 归档策略 | 90 天清 detail / 365 天归档 / 幂等 7 天 TTL / policy 留 5 |
|
||||
|
||||
### v0.2.2 (2026-10-29)
|
||||
|
||||
**引擎能力补全 — 测试靶场 168/168 100% 通过:**
|
||||
|
||||
@@ -18,6 +18,48 @@ import threading
|
||||
import time
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
# ============================================================================
|
||||
# 内部测试 channel 黑名单
|
||||
# ============================================================================
|
||||
# 控制台页面(Dashboard / 任务历史)默认隐藏内部测试 / 对账脚本产生的记录。
|
||||
# 这些 channel 由开发 / 运维工具在内部调用 scanner.scan_file() 时自动产生,
|
||||
# 不应暴露给控制台用户。若后续新增内部脚本 channel,在此补充即可。
|
||||
INTERNAL_TEST_CHANNELS: frozenset[str] = frozenset({
|
||||
"test_center", # 测试靶场手动重扫
|
||||
"v", # 对账脚本 v5/v7/v8
|
||||
"verify", # 单条验证脚本
|
||||
"diag", # 诊断脚本
|
||||
"debug", # 调试脚本
|
||||
"final", # 最终验证
|
||||
"console", # console.py 内部调用
|
||||
"console-test", # console 测试模式
|
||||
"regen_v2", # 批量重建
|
||||
"regenerate_index", # index.json 重建
|
||||
"audit_f3", # 审计脚本
|
||||
"audit", # 审计脚本
|
||||
"testcases", # 生成 TEST_CASES.md
|
||||
"test", # 测试
|
||||
"yara_verify", # YARA 验证
|
||||
"whitelist_v", # 白名单验证
|
||||
"boundary", # 边界测试
|
||||
"yara_v2", # YARA v2 测试
|
||||
"docgen", # 文档生成脚本
|
||||
})
|
||||
|
||||
|
||||
def _apply_production_channel_filter(query, model_class):
|
||||
"""给 ScanTask / ScanFinding 查询追加「隐藏内部测试 channel」过滤。
|
||||
|
||||
用法: _apply_production_channel_filter(
|
||||
db.query(func.count(ScanTask.id)), ScanTask
|
||||
).scalar()
|
||||
"""
|
||||
col = getattr(model_class, "channel", None)
|
||||
if col is None:
|
||||
return query
|
||||
return query.filter(col.notin_(INTERNAL_TEST_CHANNELS))
|
||||
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, Response, UploadFile
|
||||
@@ -535,30 +577,31 @@ async def dashboard(request: Request, current=Depends(require_login)):
|
||||
db: Session = SessionLocal()
|
||||
try:
|
||||
from sqlalchemy import func
|
||||
total_scans = db.query(func.count(ScanTask.id)).scalar() or 0
|
||||
_skip_test = ScanTask.channel.notin_(INTERNAL_TEST_CHANNELS)
|
||||
total_scans = db.query(func.count(ScanTask.id)).filter(_skip_test).scalar() or 0
|
||||
total_blocked = db.query(func.count(ScanTask.id)).filter(
|
||||
ScanTask.action == "reject"
|
||||
ScanTask.action == "reject", _skip_test
|
||||
).scalar() or 0
|
||||
total_flagged = db.query(func.count(ScanTask.id)).filter(
|
||||
ScanTask.action.in_(("reject", "manual_review", "pass_with_flag"))
|
||||
ScanTask.action.in_(("reject", "manual_review", "pass_with_flag")), _skip_test
|
||||
).scalar() or 0
|
||||
total_findings = db.query(func.count(ScanFinding.id)).scalar() or 0
|
||||
|
||||
# 结论分布
|
||||
verdict_rows = db.query(ScanTask.verdict, func.count(ScanTask.id)).group_by(
|
||||
ScanTask.verdict
|
||||
).all()
|
||||
verdict_rows = db.query(ScanTask.verdict, func.count(ScanTask.id)).filter(
|
||||
_skip_test
|
||||
).group_by(ScanTask.verdict).all()
|
||||
verdict_dist = {str(v or "unknown"): int(c) for v, c in verdict_rows}
|
||||
|
||||
# 最近任务(8 条)
|
||||
recent_tasks = db.query(ScanTask).order_by(
|
||||
recent_tasks = db.query(ScanTask).filter(_skip_test).order_by(
|
||||
ScanTask.created_at.desc()
|
||||
).limit(8).all()
|
||||
|
||||
# 各客户端/渠道的任务数(摆渡关联性)
|
||||
client_rows = db.query(
|
||||
ScanTask.client_id, ScanTask.channel, func.count(ScanTask.id)
|
||||
).group_by(ScanTask.client_id, ScanTask.channel).all()
|
||||
).filter(_skip_test).group_by(ScanTask.client_id, ScanTask.channel).all()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
@@ -825,7 +868,8 @@ async def tasks_page(request: Request, current=Depends(require_login)):
|
||||
|
||||
db: Session = SessionLocal()
|
||||
try:
|
||||
total = db.query(func.count(ScanTask.id)).scalar() or 0
|
||||
_skip_test = ScanTask.channel.notin_(INTERNAL_TEST_CHANNELS)
|
||||
total = db.query(func.count(ScanTask.id)).filter(_skip_test).scalar() or 0
|
||||
total_pages = max(1, (total + PAGE_SIZE - 1) // PAGE_SIZE)
|
||||
# 越界时回退到最后一页
|
||||
if page > total_pages:
|
||||
@@ -834,6 +878,7 @@ async def tasks_page(request: Request, current=Depends(require_login)):
|
||||
|
||||
tasks = (
|
||||
db.query(ScanTask)
|
||||
.filter(_skip_test)
|
||||
.order_by(ScanTask.created_at.desc())
|
||||
.limit(PAGE_SIZE)
|
||||
.offset(offset)
|
||||
@@ -915,15 +960,13 @@ async def test_page(request: Request, current=Depends(require_login)):
|
||||
index = json.loads(index_path.read_text(encoding="utf-8"))
|
||||
meta_map = {s["filename"]: s for s in index.get("samples", [])}
|
||||
|
||||
# 收集所有样本文件
|
||||
for fpath in sorted(samples_dir.iterdir()):
|
||||
if not fpath.is_file():
|
||||
continue
|
||||
if fpath.name == "index.json" or fpath.name.startswith("_"):
|
||||
continue
|
||||
if fpath.suffix == ".json" and fpath.name.startswith("_scan_results"):
|
||||
continue
|
||||
samples_list.append(fpath)
|
||||
# ★ 只展示 index.json 里登记的样本(不遍历目录,避免临时/调试文件混入)
|
||||
# 每个样本必须: index.json 有登记 + 磁盘上确实存在
|
||||
for s in index.get("samples", []):
|
||||
fname = s["filename"]
|
||||
fpath = samples_dir / fname
|
||||
if fpath.is_file():
|
||||
samples_list.append(fpath)
|
||||
|
||||
results: list[dict] = []
|
||||
pass_cnt = fail_cnt = 0
|
||||
|
||||
@@ -71,6 +71,11 @@ class ScanTask(Base):
|
||||
back_populates="task", cascade="all, delete-orphan", lazy="selectin"
|
||||
)
|
||||
|
||||
__table_args__ = (
|
||||
# 控制台高频查询: WHERE channel = X ORDER BY created_at DESC LIMIT N
|
||||
Index("ix_scan_tasks_channel_created", "channel", "created_at", postgresql_using="btree"),
|
||||
)
|
||||
|
||||
|
||||
class ScanFinding(Base):
|
||||
__tablename__ = "scan_findings"
|
||||
|
||||
@@ -300,15 +300,18 @@ class PolicyStore:
|
||||
def _load_db_or_yaml(self) -> Policy:
|
||||
"""优先 DB,回退 YAML。"""
|
||||
try:
|
||||
from sqlalchemy import select
|
||||
from app.db import SessionLocal
|
||||
from app.models import PolicyVersion
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
pv = db.execute(
|
||||
select(PolicyVersion).order_by(PolicyVersion.created_at.desc())
|
||||
).scalar_one_or_none()
|
||||
# 用 ORM Query API — 直接返回 PolicyVersion 对象
|
||||
# 而不是 Row 对象(db.execute(select) 会返回 Row)
|
||||
pv = (
|
||||
db.query(PolicyVersion)
|
||||
.order_by(PolicyVersion.created_at.desc())
|
||||
.first() # 取最新一条,不管历史版本多少条
|
||||
)
|
||||
if pv:
|
||||
policy = Policy(
|
||||
version=pv.version,
|
||||
|
||||
@@ -133,6 +133,29 @@ services:
|
||||
start_period: 15s
|
||||
retries: 3
|
||||
|
||||
# ---------- Cron (定时任务: DB 归档 + VACUUM) ----------
|
||||
# 纯 Python 实现, 零额外依赖, 和 api 共享 meshsafe:latest 镜像
|
||||
# 任务:
|
||||
# 每天 03:00 archive_old_tasks.py --apply (清 90 天 detail / 365 天归档 / 幂等 TTL)
|
||||
# 每周一 03:10 VACUUM ANALYZE (维护表统计信息)
|
||||
cron:
|
||||
image: meshsafe:latest
|
||||
container_name: meshsafe-cron
|
||||
restart: unless-stopped
|
||||
networks: [meshsafe_net]
|
||||
depends_on:
|
||||
postgres:
|
||||
condition: service_healthy
|
||||
environment:
|
||||
- TZ=Asia/Shanghai
|
||||
- MESHSAFE_ENV=prod
|
||||
- MESHSAFE_DB_URL=postgresql+psycopg2://meshsafe:${MESHSAFE_DB_PASSWORD:-meshsafe}@postgres:5432/meshsafe
|
||||
- MESHSAFE_MASTER_KEY=${MESHSAFE_MASTER_KEY:?MASTER_KEY_REQUIRED}
|
||||
volumes:
|
||||
- ./data/logs:/var/log/meshsafe # 归档日志共享目录
|
||||
- ./scripts:/app/scripts:ro # archive_old_tasks.py + cron_entrypoint.py
|
||||
entrypoint: ["python3", "/app/scripts/cron_entrypoint.py"]
|
||||
|
||||
# ---------- Nginx(反向代理,内网 HTTP 明文)----------
|
||||
# 外部端口 44380 → 内部 80 → api:8443
|
||||
# 无 TLS,外层防护靠光闸
|
||||
|
||||
2724
docs/TEST_CASES.md
Normal file
2724
docs/TEST_CASES.md
Normal file
File diff suppressed because it is too large
Load Diff
264
scripts/archive_old_tasks.py
Normal file
264
scripts/archive_old_tasks.py
Normal file
@@ -0,0 +1,264 @@
|
||||
"""DB 归档 + 清理脚本。
|
||||
|
||||
职责:
|
||||
1. scan_tasks 90 天前: 清 detail JSONB (节省空间, 扫描详情留 90 天)
|
||||
2. scan_tasks 365 天前: 移到 scan_tasks_archive (含 ScanFinding)
|
||||
3. idempotency_records: 清 7 天前的 (幂等缓存过期重扫)
|
||||
4. policy_versions: 只保留最新 5 个版本 (历史版本清掉)
|
||||
|
||||
用法:
|
||||
docker compose exec api python3 scripts/archive_old_tasks.py # dry-run, 只报告不动数据
|
||||
docker compose exec api python3 scripts/archive_old_tasks.py --apply # 真正执行
|
||||
|
||||
建议: 每天 cron 跑一次
|
||||
|
||||
设计原则:
|
||||
- 小批量提交 (1000 条一批), 避免长事务锁表
|
||||
- 先 dry-run 再 apply, 输出清晰
|
||||
- pg_repack / VACUUM ANALYZE 在脚本最后触发 (让 autovacuum 兜底也行)
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
import sys
|
||||
import os
|
||||
from datetime import timedelta, datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, "/app")
|
||||
os.environ.setdefault("MESHSAFE_MASTER_KEY", "Dt0ktRs8SjEH8VxWZJDX125NV1qiII88HoiXZPhWers=")
|
||||
|
||||
from app.db import SessionLocal # noqa: E402
|
||||
from app.models import ( # noqa: E402
|
||||
ScanTask, ScanFinding, IdempotencyRecord, PolicyVersion,
|
||||
)
|
||||
from sqlalchemy import select, func # noqa: E402
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
||||
log = logging.getLogger("archive")
|
||||
|
||||
# ---- 配置 ----
|
||||
DETAIL_RETENTION_DAYS = 90 # 90 天前清 detail JSONB
|
||||
ARCHIVE_RETENTION_DAYS = 365 # 365 天前移归档
|
||||
IDEM_RETENTION_DAYS = 7 # 幂等缓存 7 天 TTL
|
||||
POLICY_KEEP_VERSIONS = 5 # policy_versions 只留最新 5 个
|
||||
BATCH_SIZE = 1000 # 分批处理
|
||||
|
||||
|
||||
def _cutoff(days: int) -> datetime:
|
||||
return datetime.now(timezone.utc) - timedelta(days=days)
|
||||
|
||||
|
||||
def archive_tasks(dry_run: bool) -> None:
|
||||
"""归档 365 天前的 scan_tasks 到 scan_tasks_archive。"""
|
||||
cutoff = _cutoff(ARCHIVE_RETENTION_DAYS)
|
||||
db = SessionLocal()
|
||||
try:
|
||||
total = db.execute(
|
||||
select(func.count(ScanTask.id)).where(ScanTask.created_at < cutoff)
|
||||
).scalar() or 0
|
||||
|
||||
log.info(f"[归档] {ARCHIVE_RETENTION_DAYS} 天前的 scan_tasks: {total} 条")
|
||||
|
||||
if total == 0:
|
||||
log.info(" → 无需归档")
|
||||
return
|
||||
|
||||
if dry_run:
|
||||
log.info(f" [dry-run] 将归档 {total} 条")
|
||||
return
|
||||
|
||||
# 分批
|
||||
moved = 0
|
||||
while True:
|
||||
batch = db.execute(
|
||||
select(ScanTask).where(ScanTask.created_at < cutoff).limit(BATCH_SIZE)
|
||||
).scalars().all()
|
||||
if not batch:
|
||||
break
|
||||
|
||||
for task in batch:
|
||||
try:
|
||||
db.execute(
|
||||
__import__("sqlalchemy").text("""
|
||||
INSERT INTO scan_tasks_archive
|
||||
(id, channel, client_id, filename, declared_ext, size,
|
||||
sha256, md5, real_type, status, mode, verdict, action, score,
|
||||
reason, policy_version, has_detail, has_findings,
|
||||
created_at, finished_at, duration_ms, archived_at)
|
||||
VALUES
|
||||
(:id, :channel, :client_id, :filename, :declared_ext, :size,
|
||||
:sha256, :md5, :real_type, :status, :mode, :verdict, :action, :score,
|
||||
:reason, :policy_version, :has_detail, :has_findings,
|
||||
:created_at, :finished_at, :duration_ms, NOW())
|
||||
ON CONFLICT (id) DO NOTHING
|
||||
"""),
|
||||
{
|
||||
"id": task.id,
|
||||
"channel": task.channel,
|
||||
"client_id": task.client_id,
|
||||
"filename": task.filename,
|
||||
"declared_ext": task.declared_ext,
|
||||
"size": task.size,
|
||||
"sha256": task.sha256,
|
||||
"md5": task.md5,
|
||||
"real_type": task.real_type,
|
||||
"status": task.status,
|
||||
"mode": task.mode,
|
||||
"verdict": task.verdict,
|
||||
"action": task.action,
|
||||
"score": task.score,
|
||||
"reason": task.reason,
|
||||
"policy_version": task.policy_version,
|
||||
"has_detail": bool(task.detail and len(task.detail) > 0),
|
||||
"has_findings": bool(task.findings),
|
||||
"created_at": task.created_at,
|
||||
"finished_at": task.finished_at,
|
||||
"duration_ms": task.duration_ms,
|
||||
},
|
||||
)
|
||||
# CASCADE 删 ScanFinding
|
||||
db.delete(task)
|
||||
moved += 1
|
||||
except Exception as e:
|
||||
log.warning(f" task {task.id} 归档失败: {e}")
|
||||
db.rollback()
|
||||
continue
|
||||
db.commit()
|
||||
log.info(f" 已归档 {moved}/{total}")
|
||||
|
||||
log.info(f" ✅ 归档完成: {moved} 条")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def scrub_detail(dry_run: bool) -> None:
|
||||
"""90~365 天之间的任务清 detail JSONB (留摘要, 省空间)。"""
|
||||
db_cutoff = _cutoff(DETAIL_RETENTION_DAYS)
|
||||
arc_cutoff = _cutoff(ARCHIVE_RETENTION_DAYS)
|
||||
db = SessionLocal()
|
||||
try:
|
||||
total = db.execute(
|
||||
select(func.count(ScanTask.id)).where(
|
||||
ScanTask.created_at < db_cutoff,
|
||||
ScanTask.created_at >= arc_cutoff,
|
||||
ScanTask.detail.isnot(None),
|
||||
)
|
||||
).scalar() or 0
|
||||
|
||||
log.info(f"[清 detail] {DETAIL_RETENTION_DAYS}~{ARCHIVE_RETENTION_DAYS} 天之间有 detail 的: {total} 条")
|
||||
|
||||
if total == 0:
|
||||
log.info(" → 无需清理")
|
||||
return
|
||||
|
||||
if dry_run:
|
||||
log.info(f" [dry-run] 将清 {total} 条 detail")
|
||||
return
|
||||
|
||||
# 分批 UPDATE
|
||||
updated = 0
|
||||
while True:
|
||||
batch_ids = db.execute(
|
||||
select(ScanTask.id).where(
|
||||
ScanTask.created_at < db_cutoff,
|
||||
ScanTask.created_at >= arc_cutoff,
|
||||
ScanTask.detail.isnot(None),
|
||||
).limit(BATCH_SIZE)
|
||||
).scalars().all()
|
||||
if not batch_ids:
|
||||
break
|
||||
db.execute(
|
||||
__import__("sqlalchemy").update(ScanTask).where(ScanTask.id.in_(batch_ids)).values(detail=None)
|
||||
)
|
||||
db.commit()
|
||||
updated += len(batch_ids)
|
||||
log.info(f" 已清 {updated}/{total}")
|
||||
|
||||
log.info(f" ✅ detail 清理完成: {updated} 条")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def clean_idempotency(dry_run: bool) -> None:
|
||||
"""清 7 天前的幂等缓存 (过期的可以重扫)。"""
|
||||
cutoff = _cutoff(IDEM_RETENTION_DAYS)
|
||||
db = SessionLocal()
|
||||
try:
|
||||
total = db.execute(
|
||||
select(func.count(IdempotencyRecord.id)).where(IdempotencyRecord.created_at < cutoff)
|
||||
).scalar() or 0
|
||||
|
||||
log.info(f"[幂等清理] {IDEM_RETENTION_DAYS} 天前的 idempotency_records: {total} 条")
|
||||
|
||||
if total == 0:
|
||||
log.info(" → 无需清理")
|
||||
return
|
||||
|
||||
if dry_run:
|
||||
log.info(f" [dry-run] 将清 {total} 条")
|
||||
return
|
||||
|
||||
db.execute(
|
||||
__import__("sqlalchemy").delete(IdempotencyRecord).where(IdempotencyRecord.created_at < cutoff)
|
||||
)
|
||||
db.commit()
|
||||
log.info(f" ✅ 幂等清理完成: {total} 条")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def trim_policy_versions(dry_run: bool) -> None:
|
||||
"""policy_versions 只留最新 N 个。"""
|
||||
db = SessionLocal()
|
||||
try:
|
||||
all_pv = db.execute(
|
||||
select(PolicyVersion).order_by(PolicyVersion.created_at.desc())
|
||||
).scalars().all()
|
||||
|
||||
keep = all_pv[:POLICY_KEEP_VERSIONS]
|
||||
drop = all_pv[POLICY_KEEP_VERSIONS:]
|
||||
|
||||
log.info(f"[policy] 共 {len(all_pv)} 条历史版本, 留 {len(keep)} 删 {len(drop)}")
|
||||
|
||||
if not drop:
|
||||
log.info(" → 无需清理")
|
||||
return
|
||||
|
||||
if dry_run:
|
||||
for pv in drop:
|
||||
log.info(f" [dry-run] 将删 {pv.version}")
|
||||
return
|
||||
|
||||
for pv in drop:
|
||||
db.delete(pv)
|
||||
db.commit()
|
||||
log.info(f" ✅ policy_versions 清理完成: 删 {len(drop)} 条")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="MeshSafe DB 归档 + 清理")
|
||||
parser.add_argument("--apply", action="store_true", help="真正执行 (默认 dry-run)")
|
||||
args = parser.parse_args()
|
||||
|
||||
dry = not args.apply
|
||||
log.info(f"模式: {'DRY-RUN (仅报告)' if dry else 'APPLY (真正执行)'}")
|
||||
log.info("")
|
||||
|
||||
archive_tasks(dry)
|
||||
log.info("")
|
||||
scrub_detail(dry)
|
||||
log.info("")
|
||||
clean_idempotency(dry)
|
||||
log.info("")
|
||||
trim_policy_versions(dry)
|
||||
|
||||
if dry:
|
||||
log.info("\n💡 确认无误后加 --apply 参数正式执行")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
121
scripts/cron_entrypoint.py
Normal file
121
scripts/cron_entrypoint.py
Normal file
@@ -0,0 +1,121 @@
|
||||
"""MeshSafe 定时任务调度器 — 零依赖,Docker 容器前台常驻。
|
||||
|
||||
和 api 共享同一个镜像,容器 entrypoint 换成本脚本即可。
|
||||
用法: docker run meshsafe:latest python3 scripts/cron_entrypoint.py
|
||||
|
||||
支持的任务:
|
||||
每天 03:00 归档 + 清理 DB (archive_old_tasks.py --apply)
|
||||
每周一 03:10 VACUUM ANALYZE (维护表统计信息)
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
sys.path.insert(0, "/app")
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s [%(levelname)s] %(message)s",
|
||||
)
|
||||
log = logging.getLogger("scheduler")
|
||||
|
||||
|
||||
def run(cmd: list[str], label: str) -> None:
|
||||
"""跑一个命令,捕获输出并记日志。"""
|
||||
log.info(f"[{label}] 开始执行: {' '.join(cmd)}")
|
||||
try:
|
||||
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=1800)
|
||||
if proc.stdout:
|
||||
for line in proc.stdout.strip().splitlines():
|
||||
log.info(f"[{label}] {line}")
|
||||
if proc.stderr:
|
||||
for line in proc.stderr.strip().splitlines():
|
||||
log.warning(f"[{label}] {line}")
|
||||
if proc.returncode == 0:
|
||||
log.info(f"[{label}] ✅ 完成")
|
||||
else:
|
||||
log.error(f"[{label}] ❌ 返回码 {proc.returncode}")
|
||||
except subprocess.TimeoutExpired:
|
||||
log.error(f"[{label}] ❌ 超时 (30min)")
|
||||
except Exception as e:
|
||||
log.error(f"[{label}] ❌ 异常: {e}")
|
||||
|
||||
|
||||
def run_vacuum() -> None:
|
||||
"""VACUUM ANALYZE — 维护表统计信息。"""
|
||||
try:
|
||||
from app.db import engine
|
||||
from sqlalchemy import text
|
||||
|
||||
tables = [
|
||||
"scan_tasks", "scan_tasks_archive",
|
||||
"scan_findings", "idempotency_records",
|
||||
"policy_versions", "audit_logs",
|
||||
]
|
||||
with engine.begin() as conn:
|
||||
for tbl in tables:
|
||||
log.info(f"[VACUUM] {tbl} ...")
|
||||
conn.execute(text(f"VACUUM ANALYZE {tbl}"))
|
||||
log.info("[VACUUM] ✅ 全部完成")
|
||||
except Exception as e:
|
||||
log.error(f"[VACUUM] ❌ 异常: {e}")
|
||||
|
||||
|
||||
def seconds_until(target_hour: int, target_minute: int, weekday: int | None = None) -> float:
|
||||
"""计算到下次目标时间的秒数。
|
||||
|
||||
weekday: 0=周一...6=周日, None=每天
|
||||
"""
|
||||
now = datetime.now()
|
||||
candidate = now.replace(hour=target_hour, minute=target_minute, second=0, microsecond=0)
|
||||
|
||||
if weekday is not None:
|
||||
# 找到本周目标日
|
||||
days_ahead = weekday - now.weekday()
|
||||
if days_ahead < 0 or (days_ahead == 0 and now >= candidate):
|
||||
days_ahead += 7
|
||||
candidate += timedelta(days=days_ahead)
|
||||
else:
|
||||
if candidate <= now:
|
||||
candidate += timedelta(days=1)
|
||||
|
||||
return (candidate - now).total_seconds()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
log.info("=" * 60)
|
||||
log.info("MeshSafe 定时任务调度器启动")
|
||||
log.info(f"PID={os.getpid()} TZ=Asia/Shanghai")
|
||||
log.info("任务:")
|
||||
log.info(" - 每天 03:00 archive_old_tasks.py --apply")
|
||||
log.info(" - 每周一 03:10 VACUUM ANALYZE")
|
||||
log.info("=" * 60)
|
||||
|
||||
while True:
|
||||
# 计算两个任务下次执行时间,取更早的那个
|
||||
secs_archive = seconds_until(3, 0, weekday=None) # 每天 03:00
|
||||
secs_vacuum = seconds_until(3, 10, weekday=0) # 周一 03:10
|
||||
|
||||
if secs_archive <= secs_vacuum:
|
||||
log.info(f"下一个任务: 归档 → {secs_archive/3600:.1f}h 后")
|
||||
time.sleep(secs_archive)
|
||||
run(
|
||||
[sys.executable, "/app/scripts/archive_old_tasks.py", "--apply"],
|
||||
"ARCHIVE",
|
||||
)
|
||||
# 跑完后清下一次时间 (避免在 03:00 这一刻两个任务重叠)
|
||||
continue
|
||||
else:
|
||||
log.info(f"下一个任务: VACUUM → {secs_vacuum/3600:.1f}h 后")
|
||||
time.sleep(secs_vacuum)
|
||||
run_vacuum()
|
||||
continue
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1 +0,0 @@
|
||||
meshsafe-hot-reload-test-string present here
|
||||
@@ -1 +0,0 @@
|
||||
trojan-metadata-test payload here
|
||||
@@ -1 +0,0 @@
|
||||
powershell -enc ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwx+/==
|
||||
@@ -1 +0,0 @@
|
||||
size_condition_test_payload small file
|
||||
Reference in New Issue
Block a user