feat: P1-19 P1-20 实现 - 边界测试强制 + 高风险审计 + UI Skill 路由

P1-19.1: Arc 边界测试强制
- TaskNode 新增 test_required 字段
- _inject_boundary_tests() 为每个模块注入接口测试和单元测试任务
- Done When 验证必须包含"测试通过"

P1-19.2: AirRvr 高风险审计
- 新增 HighRiskAudit, HighRiskFinding 数据类
- ReviewReport 新增 highRiskAudit 字段,含 lifecycle/nullPointer/danglingPointer/exceptionSafety/concurrency + overallRisk + deliveryVerdict
- 序列化/反序列化支持

P1-19.3: block-release 集成
- dispatch_worker_group() 派发前扫描最新审查报告
- deliveryVerdict=block-release 时阻止所有后续派发
- 记录 eng.blocked 事件

P1-20: frontend-design Skill 集成
- is_ui_task() UI 任务检测
- ensure_frontend_design_skill() 自动安装 Skill
- route_ui_task() UI 任务路由决策
- enter_worker() 集成 UI 检测,skill 不可用时阻止执行
- commands/do.md 更新 UI 处理说明
- SKILL.md 新增 INV-12/INV-13/INV-14

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
AirLongDian
2026-06-10 17:06:48 +08:00
commit e73a4da354
54 changed files with 6054 additions and 0 deletions

View File

@@ -0,0 +1,175 @@
"""
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