AirPlan-V2/lib/air_runtime/sec_runtime.py

"""
AirSec 安全扫描运行时 — V2 新增组件。
制品敏感数据扫描 + 自动脱敏 + 误报白名单 + 确认流程 + advisory/blocking 模式。
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from pathlib import Path

SECRET_PATTERNS: list[tuple[str, str]] = [
    ("api_key", r'(?:api[_-]?key|apikey)\s*[:=]\s*["\']?([A-Za-z0-9_\-]{16,})["\']?'),
    ("aws_key", r'AKIA[0-9A-Z]{16}'),
    ("private_key", r'-----BEGIN (?:RSA|EC|DSA|OPENSSH) PRIVATE KEY-----'),
    ("token", r'(?:token|secret|password)\s*[:=]\s*["\']?([^\s"\']{8,})["\']?'),
    ("jwt", r'eyJ[A-Za-z0-9_\-]+\.eyJ[A-Za-z0-9_\-]+\.[A-Za-z0-9_\-]+'),
    ("url_credential", r'https?://[^:@]+:([^@]+)@'),
]

ALLOWLIST_PATTERNS: list[str] = [
    r'EXAMPLE',
    r'example',
    r'YOUR_API_KEY',
    r'TODO',
    r'<your-',
    r'placeholder',
    r'xxxxxxxx',
]

# 文件名白名单：命中则跳过该文件的所有发现
WHITELIST_FILE_PATTERNS: list[str] = [
    r"\.example\.",
    r"^test_",
    r"^mock_",
    r"_test\.py$",
    r"\.fixture\.",
]

# 用户确认记录（首次发现需确认，后续同模式自动放行）
USER_CONFIRMATIONS: dict[str, dict] = {}  # {fingerprint: {pattern, confirmed_at, user}}


class ScanMode:
    """扫描模式：advisory（只报告，不阻止）/ blocking（阻止合并）"""
    ADVISORY = "advisory"
    BLOCKING = "blocking"


@dataclass
class ScanFinding:
    rule: str
    file: str
    line: int
    match: str  # 截断显示，不包含完整密钥
    severity: str = "high"  # high | medium | low


@dataclass
class ScanReport:
    task_id: str
    findings: list[ScanFinding] = field(default_factory=list)
    whitelisted: int = 0
    clean: bool = True
    advisory_blocked: bool = False


def scan_file(file_path: Path, task_id: str = "") -> ScanReport:
    findings: list[ScanFinding] = []
    whitelisted = 0

    try:
        content = file_path.read_text(encoding="utf-8", errors="replace")
    except Exception:
        return ScanReport(task_id=task_id, clean=True)

    for rule, pattern in SECRET_PATTERNS:
        for match in re.finditer(pattern, content, re.IGNORECASE):
            matched_text = match.group(0)
            if any(re.search(ap, matched_text) for ap in ALLOWLIST_PATTERNS):
                whitelisted += 1
                continue

            line_no = content[:match.start()].count("\n") + 1
            display = matched_text[:60] + "..." if len(matched_text) > 60 else matched_text
            findings.append(ScanFinding(
                rule=rule, file=str(file_path), line=line_no, match=display,
            ))

    return ScanReport(
        task_id=task_id,
        findings=findings,
        whitelisted=whitelisted,
        clean=len(findings) == 0,
    )


def scan_result_data(result: dict, task_id: str = "") -> ScanReport:
    """扫描 Worker result.json 中的敏感数据。"""
    import json
    text = json.dumps(result, ensure_ascii=False)
    findings: list[ScanFinding] = []
    whitelisted = 0

    for rule, pattern in SECRET_PATTERNS:
        for match in re.finditer(pattern, text, re.IGNORECASE):
            matched_text = match.group(0)
            if any(re.search(ap, matched_text) for ap in ALLOWLIST_PATTERNS):
                whitelisted += 1
                continue
            display = matched_text[:60] + "..." if len(matched_text) > 60 else matched_text
            findings.append(ScanFinding(
                rule=rule, file="result.json", line=0, match=display,
            ))

    return ScanReport(
        task_id=task_id,
        findings=findings,
        whitelisted=whitelisted,
        clean=len(findings) == 0,
    )


def scan_file_with_mode(
    file_path: Path,
    task_id: str = "",
    mode: str = ScanMode.BLOCKING,
    confirm_callback=None,  # 可选：首次发现时调用此回调询问用户
) -> ScanReport:
    """
    增强版扫描：
    1. 基础扫描（已有逻辑）
    2. 文件名白名单过滤
    3. 模式判断（advisory vs blocking）
    """
    report = scan_file(file_path, task_id)  # 原有逻辑

    # 文件名白名单过滤
    filtered_findings = []
    for f in report.findings:
        filename = file_path.name
        if any(re.search(p, filename) for p in WHITELIST_FILE_PATTERNS):
            report.whitelisted += 1
            continue
        filtered_findings.append(f)

    report.findings = filtered_findings
    report.clean = len(filtered_findings) == 0

    # 模式处理
    if not report.clean and mode == ScanMode.ADVISORY:
        # advisory 模式：只记录，不阻止
        report.advisory_blocked = False
    elif not report.clean and mode == ScanMode.BLOCKING:
        # blocking 模式：默认阻止
        report.advisory_blocked = True

    return report


def confirm_pattern(task_id: str, pattern: str, user: str = "unknown") -> None:
    """用户确认某模式为安全后，记录下来"""
    from air_runtime.utils import now_iso
    fingerprint = f"{task_id}:{pattern}"
    USER_CONFIRMATIONS[fingerprint] = {
        "pattern": pattern,
        "confirmed_at": now_iso(),
        "user": user,
    }


def is_confirmed(task_id: str, pattern: str) -> bool:
    """检查某模式是否已被用户确认"""
    fingerprint = f"{task_id}:{pattern}"
    return fingerprint in USER_CONFIRMATIONS