mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-28 08:56:13 +00:00
658c39ccf7
* Add phase 1 skill static scanning * Rework SkillScan phase 1 as native scanner * refactor(skillscan): align phase 1 with trimmed RFC contract - SecurityFinding: 7 fields (rule_id, severity, file, line, message, remediation, evidence); category/analyzer derive from the rule_id prefix, confidence/column/fingerprint/metadata removed - scan_archive_preflight()/scan_skill_dir() are pure functions: no ScanContext, no policy schema; CRITICAL-blocks is a code constant and skill_scan.enabled is applied by enforce_static_scan()/callers - secret-* evidence is redacted before findings leave the scanner - de-dup keys on (rule_id, file, line) so repeated occurrences keep distinct locations for agent self-correction - cloud-metadata detection consolidated into network-cloud-metadata - nested zip members get a one-level stdlib magic-byte peek; an executable member escalates package-nested-archive to CRITICAL - install metadata sidecar removed (Phase 7 decides if it is needed) - rule specs moved next to their analyzers; skillscan/rules/ removed - tests updated + new anchors: redaction, dedup lines, nested-zip escalation, single cloud-metadata rule, bundled-skill zero-CRITICAL * fix(skillscan): tighten reverse-shell/secret/archive scan rules from review Address PR #3033 review feedback on the native SkillScan analyzers: - Reverse-shell false positives: split shell detection by signal strength (/dev/tcp/, nc -e stay CRITICAL; bash -i, mkfifo -> new HIGH shell-reverse-shell-heuristic, warn->LLM). The Python check is now AST-anchored on real socket.socket/os.dup2/subprocess call sites instead of raw-text substring matching, so prose/docstrings no longer hard-block. - Secret evidence: _redact_secret_evidence returns [redacted] with no secret bytes (was value[:6], which leaked 2 real token bytes past the prefix). - Archive DoS: cap outer archive member count (_MAX_ARCHIVE_MEMBERS=4096); scan_archive_preflight early-aborts with a package-too-many-members CRITICAL finding (routes through the existing blocked->400 fail-closed path). - shell-destructive-command: broaden the rm -rf matcher to sensitive system roots (/home, /usr, /*, --no-preserve-root /) while leaving safe subpaths unflagged. - Dead code: collapse _decode_text_for_analysis to a single decode path and drop the unused _TEXT_SUFFIXES set and _has_text_shebang helper. - local_skill_storage: document why the host_path branch keeps app_config possibly-None (lazy kill-switch resolution; avoids eager get_app_config in config-free environments such as CI). Tests: new negative/positive coverage in test_skillscan_native.py. Full backend suite 6616 passed, 26 skipped.
60 lines
1.8 KiB
Python
60 lines
1.8 KiB
Python
"""Data contracts for DeerFlow SkillScan.
|
|
|
|
Every ``SecurityFinding`` field has a Phase 1 consumer: the blocking policy
|
|
reads ``severity``; the Gateway rejection response, the agent tool error, and
|
|
the LLM scanner context read the rest. The rule category and owning analyzer
|
|
are encoded in the ``rule_id`` prefix (``package-``, ``secret-``,
|
|
``declaration-``, ``python-``, ``shell-``, ``network-``/``resource-``), not
|
|
duplicated as separate fields.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Literal, TypedDict
|
|
|
|
FindingSeverity = Literal["CRITICAL", "HIGH", "MEDIUM", "LOW"]
|
|
|
|
|
|
class SecurityFinding(TypedDict):
|
|
rule_id: str
|
|
severity: FindingSeverity
|
|
file: str | None
|
|
line: int | None
|
|
message: str
|
|
remediation: str
|
|
evidence: str | None
|
|
|
|
|
|
class ScanResult(TypedDict):
|
|
findings: list[SecurityFinding]
|
|
blocked: bool
|
|
scanner_errors: list[str]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RuleSpec:
|
|
"""Static definition of one SkillScan rule; ``remediation`` is authored here once and copied into findings."""
|
|
|
|
rule_id: str
|
|
severity: FindingSeverity
|
|
message: str
|
|
remediation: str
|
|
|
|
|
|
class StaticScannerError(RuntimeError):
|
|
"""Raised when SkillScan cannot evaluate its input at the package boundary."""
|
|
|
|
|
|
class StaticScanBlockedError(ValueError):
|
|
"""Raised when deterministic findings block a skill write or install."""
|
|
|
|
findings: list[SecurityFinding]
|
|
skill_name: str | None
|
|
|
|
def __init__(self, findings: list[SecurityFinding], *, skill_name: str | None = None, message: str | None = None) -> None:
|
|
self.findings = [dict(finding) for finding in findings] # type: ignore[list-item]
|
|
self.skill_name = skill_name
|
|
subject = f"skill '{skill_name}'" if skill_name else "skill content"
|
|
super().__init__(message or f"Static security scan blocked {subject}")
|