"""AirSec mode — V2 安全扫描器。""" import sys from pathlib import Path from air_runtime.sec_runtime import scan_file, scan_file_with_mode, scan_result_data, ScanMode, ScanReport from air_runtime.io import safe_json_load from air_runtime.paths import airplan_root, event_log_path from air_runtime.events import EventLog, SEC_SCAN def main(args) -> None: project_root = Path(args.project).expanduser().resolve() tid = args.task_id or "unknown" sub = args.sub or "scan" mode = getattr(args, "sec_mode", "blocking") or "blocking" if mode not in ("advisory", "blocking"): print("error: mode must be advisory or blocking", file=sys.stderr) sys.exit(1) if sub == "scan": if args.scan_path: scan_path = Path(args.scan_path).expanduser().resolve() if scan_path.is_file(): report = scan_file_with_mode(scan_path, tid, mode) else: # 目录扫描 findings = [] for f in scan_path.rglob("*"): if f.is_file() and not any(x in f.name for x in [".git", "node_modules", "__pycache__"]): r = scan_file_with_mode(f, tid, mode) findings.extend(r.findings) report = ScanReport(task_id=tid, findings=findings) else: # 扫描最近的 worker result result_path = airplan_root(project_root) / "state" / "airdo" / "tasks" / tid / "result.json" data = safe_json_load(result_path) or {} report = scan_result_data(data, tid) log = EventLog(event_log_path(project_root)) log.emit(SEC_SCAN, { "taskId": tid, "clean": report.clean, "findings": len(report.findings), "whitelisted": report.whitelisted, "mode": mode, }) print("airplan_mode=sec") print(f"task_id={tid}") print(f"scan_path={getattr(args, 'scan_path', '')}") print(f"mode={mode}") print(f"clean={report.clean}") print(f"findings={len(report.findings)}") print(f"whitelisted={report.whitelisted}") if report.findings: for f in report.findings[:5]: print(f" {f.file}:{f.line} [{f.severity}] {f.rule}: {f.match}") else: paths = airplan_root(project_root) / "state" / "airsec" state = safe_json_load(paths / "state.json") or {} print(f"airplan_mode=sec\nenabled={state.get('enabled', False)}")