diff --git a/app/console.py b/app/console.py index 9de5e3b..8e1da55 100644 --- a/app/console.py +++ b/app/console.py @@ -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)