mirror/scripts/scan.py

#!/usr/bin/env python3
"""Static security scanner for agent skill packages."""

from __future__ import annotations

import argparse
import dataclasses
import hashlib
import json
import re
import shutil
import sys
import tempfile
import urllib.request
import zipfile
from pathlib import Path, PurePosixPath
from typing import Iterable


VERSION = "5.2.1"
REPORT_SCHEMA_VERSION = "skill-security-scan.v1"
MAX_ZIP_BYTES = 10 * 1024 * 1024
MAX_ZIP_FILES = 500
URL_TIMEOUT_SECONDS = 10
URL_MAX_BYTES = 2 * 1024 * 1024
FILE_MARKER = re.compile(r"^### FILE: (.+)$")
EXCLUDED_DIRS = {
    ".git",
    ".github",
    ".mypy_cache",
    ".pytest_cache",
    "__pycache__",
    "node_modules",
    "tests",
    "venv",
    ".venv",
}


class ScanError(Exception):
    """Raised when input cannot be safely scanned."""


@dataclasses.dataclass(frozen=True)
class Evidence:
    file: str
    line: int
    text: str


@dataclasses.dataclass(frozen=True)
class Issue:
    rule_id: str
    dimension: str
    severity: str
    confidence: str
    title: str
    evidence: Evidence
    suggestion: str
    penalty: int
    direct_fail: bool = False
    auto_fix: bool = False

    def to_dict(self) -> dict:
        data = dataclasses.asdict(self)
        return data


@dataclasses.dataclass(frozen=True)
class Target:
    name: str
    source: str
    content: str


@dataclasses.dataclass(frozen=True)
class Report:
    target: str
    input_sha256: str
    rating: str
    score: int
    issues: list[Issue]
    passed_dimensions: list[str]
    ignored_rules: list[str]
    complete: bool

    def to_dict(self) -> dict:
        return {
            "schema_version": REPORT_SCHEMA_VERSION,
            "scanner_version": VERSION,
            "target": self.target,
            "input_sha256": self.input_sha256,
            "ignored_rules": self.ignored_rules,
            "complete": self.complete,
            "rating": self.rating,
            "score": self.score,
            "issues": [issue.to_dict() for issue in self.issues],
            "passed_dimensions": self.passed_dimensions,
        }


DIMENSIONS = {
    "prompt-injection": "Prompt injection",
    "sensitive-data": "Sensitive file access / data exfiltration",
    "compliance": "Compliance violations",
    "malicious-script": "Malicious scripts",
    "dependency": "Dependency safety",
    "description": "Description trigger reasonability",
    "frontmatter": "Frontmatter compliance",
}


def parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Scan skill packages for static security risks."
    )
    parser.add_argument(
        "input",
        nargs="?",
        help="Skill file, skill directory, .zip package, URL, '-' for stdin, or omitted with --text.",
    )
    parser.add_argument("--text", help="Inline skill text to scan.")
    parser.add_argument(
        "--format",
        choices=("text", "json"),
        default="text",
        help="Output format. Defaults to text.",
    )
    parser.add_argument(
        "--ignore",
        action="append",
        default=[],
        metavar="RULE_ID",
        help="Ignore a rule for this run. Can be repeated.",
    )
    parser.add_argument(
        "--include-references",
        action="store_true",
        help="Include references/ markdown files in directory and zip scans.",
    )
    parser.add_argument(
        "--max-zip-bytes",
        type=int,
        default=MAX_ZIP_BYTES,
        help=f"Reject zip files larger than this many bytes. Default: {MAX_ZIP_BYTES}.",
    )
    parser.add_argument(
        "--max-zip-files",
        type=int,
        default=MAX_ZIP_FILES,
        help=f"Reject zip files with more entries than this. Default: {MAX_ZIP_FILES}.",
    )
    parser.add_argument("--version", action="version", version=VERSION)
    return parser.parse_args(argv)


def scan_source(
    source: str | None,
    *,
    text: str | None = None,
    ignored_rules: set[str] | None = None,
    include_references: bool = False,
    max_zip_bytes: int = MAX_ZIP_BYTES,
    max_zip_files: int = MAX_ZIP_FILES,
) -> list[Report]:
    ignored_rules = ignored_rules or set()
    targets = collect_targets(
        source,
        text=text,
        include_references=include_references,
        max_zip_bytes=max_zip_bytes,
        max_zip_files=max_zip_files,
    )
    return [scan_target(target, ignored_rules=ignored_rules) for target in targets]


def collect_targets(
    source: str | None,
    *,
    text: str | None = None,
    include_references: bool = False,
    max_zip_bytes: int = MAX_ZIP_BYTES,
    max_zip_files: int = MAX_ZIP_FILES,
) -> list[Target]:
    if text is not None:
        return [Target(name="<inline-text>", source="<inline-text>", content=text)]

    if not source:
        raise ScanError("missing input: provide a path, URL, '-' for stdin, or --text")

    if source == "-":
        return [Target(name="<stdin>", source="<stdin>", content=read_stdin_utf8())]

    if is_url(source):
        return [
            Target(
                name=source,
                source=source,
                content=fetch_url(source),
            )
        ]

    path = Path(source).expanduser()
    if not path.exists():
        raise ScanError(f"input does not exist: {source}")
    path = path.resolve()

    if path.is_dir():
        return collect_directory_targets(path, include_references=include_references)

    if path.suffix.lower() == ".zip":
        return collect_zip_targets(
            path,
            include_references=include_references,
            max_zip_bytes=max_zip_bytes,
            max_zip_files=max_zip_files,
        )

    return [
        Target(
            name=path.name,
            source=str(path),
            content=read_text(path),
        )
    ]


def read_stdin_utf8() -> str:
    """Read the file protocol as strict UTF-8 on every platform."""
    buffer = getattr(sys.stdin, "buffer", None)
    if buffer is not None:
        raw = buffer.read()
        try:
            return raw.decode("utf-8")
        except UnicodeDecodeError as exc:
            raise ScanError("stdin is not valid UTF-8") from exc

    text = sys.stdin.read()
    try:
        text.encode("utf-8")
    except UnicodeEncodeError as exc:
        raise ScanError("stdin contains invalid Unicode") from exc
    return text


def collect_directory_targets(path: Path, *, include_references: bool) -> list[Target]:
    skill_files = find_skill_files(path)
    if not skill_files:
        return [
            Target(
                name=path.name,
                source=str(path),
                content=read_directory_bundle(path, include_references=include_references),
            )
        ]

    return [
        Target(
            name=target_name(path, skill_file),
            source=str(skill_file.parent),
            content=read_skill_bundle(skill_file, include_references=include_references),
        )
        for skill_file in skill_files
    ]


def find_skill_files(path: Path) -> list[Path]:
    return sorted(
        skill_file
        for skill_file in path.rglob("SKILL.md")
        if not is_excluded_path(skill_file, path)
    )


def target_name(root: Path, skill_file: Path) -> str:
    relative = skill_file.parent.relative_to(root).as_posix()
    if relative in {"", "."}:
        return root.name
    return relative


def collect_zip_targets(
    path: Path,
    *,
    include_references: bool,
    max_zip_bytes: int,
    max_zip_files: int,
) -> list[Target]:
    if path.stat().st_size > max_zip_bytes:
        raise ScanError(f"zip is too large: {path.stat().st_size} bytes > {max_zip_bytes}")

    tempdir = Path(tempfile.mkdtemp(prefix="skill-scan-"))
    try:
        safe_extract_zip(path, tempdir, max_files=max_zip_files)
        targets = collect_directory_targets(tempdir, include_references=include_references)
        if not targets:
            raise ScanError("zip contains no scannable files")
        return [
            Target(
                name=f"{path.name}:{target.name}",
                source=f"{path}:{target.source}",
                content=target.content,
            )
            for target in targets
        ]
    finally:
        shutil.rmtree(tempdir, ignore_errors=True)


def safe_extract_zip(path: Path, destination: Path, *, max_files: int) -> None:
    with zipfile.ZipFile(path) as archive:
        infos = archive.infolist()
        if len(infos) > max_files:
            raise ScanError(f"zip has too many entries: {len(infos)} > {max_files}")

        for info in infos:
            normalized = PurePosixPath(info.filename.replace("\\", "/"))
            if info.filename.endswith("/") or info.is_dir():
                continue
            if normalized.is_absolute() or ".." in normalized.parts:
                raise ScanError(f"unsafe zip entry path: {info.filename}")
            if info.file_size > MAX_ZIP_BYTES:
                raise ScanError(f"zip entry is too large: {info.filename}")

        archive.extractall(destination)


def read_directory_bundle(path: Path, *, include_references: bool) -> str:
    parts: list[str] = []
    for child in sorted(path.rglob("*")):
        if not child.is_file():
            continue
        if should_scan_file(child, path, include_references=include_references):
            parts.append(format_file_content(child.relative_to(path).as_posix(), read_text(child)))
    return "\n".join(parts)


def read_skill_bundle(skill_file: Path, *, include_references: bool) -> str:
    root = skill_file.parent
    parts = [format_file_content("SKILL.md", read_text(skill_file))]

    for subdir in ("scripts",):
        candidate = root / subdir
        if not candidate.exists():
            continue
        for child in sorted(candidate.rglob("*")):
            if child.is_file() and should_scan_file(child, root, include_references=include_references):
                parts.append(format_file_content(child.relative_to(root).as_posix(), read_text(child)))

    if include_references:
        references = root / "references"
        if references.exists():
            for child in sorted(references.rglob("*.md")):
                if child.is_file():
                    parts.append(format_file_content(child.relative_to(root).as_posix(), read_text(child)))

    return "\n".join(parts)


def should_scan_file(path: Path, root: Path, *, include_references: bool) -> bool:
    if is_excluded_path(path, root):
        return False
    rel = path.relative_to(root).as_posix()
    suffixes = {".md", ".sh", ".bash", ".py", ".js", ".ts", ".json", ".yaml", ".yml"}
    if path.suffix.lower() not in suffixes:
        return False
    if rel.startswith("references/") and not include_references:
        return False
    return True


def is_excluded_path(path: Path, root: Path) -> bool:
    try:
        parts = path.relative_to(root).parts
    except ValueError:
        return False
    return any(part in EXCLUDED_DIRS for part in parts)


def format_file_content(relative_path: str, content: str) -> str:
    return f"### FILE: {relative_path}\n{content.rstrip()}\n"


def read_text(path: Path) -> str:
    return path.read_text(encoding="utf-8", errors="replace")


def is_url(value: str) -> bool:
    return value.startswith("https://") or value.startswith("http://")


def fetch_url(url: str) -> str:
    request = urllib.request.Request(
        url,
        headers={"User-Agent": f"skill-security-guard/{VERSION}"},
    )
    with urllib.request.urlopen(request, timeout=URL_TIMEOUT_SECONDS) as response:
        content_type = response.headers.get("content-type", "")
        if "text" not in content_type and "json" not in content_type and "markdown" not in content_type:
            raise ScanError(f"URL does not look text-like: {content_type or 'unknown content type'}")
        data = response.read(URL_MAX_BYTES + 1)
    if len(data) > URL_MAX_BYTES:
        raise ScanError(f"URL response is too large: > {URL_MAX_BYTES} bytes")
    return data.decode("utf-8", errors="replace")


def scan_target(target: Target, *, ignored_rules: set[str]) -> Report:
    issues = detect_issues(target.content)
    issues = [issue for issue in issues if issue.rule_id not in ignored_rules]
    score, rating = calculate_score(issues)
    issue_dimensions = {issue.dimension for issue in issues}
    passed = [
        title
        for key, title in DIMENSIONS.items()
        if key not in issue_dimensions
    ]
    return Report(
        target=target.name,
        input_sha256="sha256:" + hashlib.sha256(target.content.encode("utf-8")).hexdigest(),
        rating=rating,
        score=score,
        issues=issues,
        passed_dimensions=passed,
        ignored_rules=sorted(ignored_rules),
        complete=True,
    )


def detect_issues(content: str) -> list[Issue]:
    issues: list[Issue] = []
    seen: set[str] = set()

    def add(issue: Issue) -> None:
        key = f"{issue.rule_id}:{issue.evidence.file}:{issue.evidence.line}"
        if key not in seen:
            seen.add(key)
            issues.append(issue)

    primary_skill = extract_primary_skill_text(content)
    frontmatter = parse_frontmatter(primary_skill)
    description = clean_yaml_scalar(frontmatter.get("description", ""))

    detect_frontmatter(primary_skill, frontmatter, add)
    detect_description(description, add)
    detect_line_rules(content, add)

    return sorted(
        issues,
        key=lambda issue: (
            0 if issue.direct_fail else 1,
            severity_order(issue.severity),
            issue.rule_id,
            issue.evidence.file,
            issue.evidence.line,
        ),
    )


def detect_frontmatter(primary_skill: str, frontmatter: dict[str, str], add) -> None:
    evidence = Evidence("SKILL.md", 1, "---")
    if not primary_skill.lstrip().startswith("---"):
        add(
            Issue(
                rule_id="F7-MISSING-FRONTMATTER",
                dimension="frontmatter",
                severity="medium",
                confidence="confirmed",
                title="SKILL.md is missing YAML frontmatter",
                evidence=evidence,
                suggestion="Add standard frontmatter with name and description.",
                penalty=30,
            )
        )
        return

    allowed = {"name", "description"}
    extra = sorted(set(frontmatter) - allowed)
    if extra:
        add(
            Issue(
                rule_id="F7-NONSTANDARD-FIELDS",
                dimension="frontmatter",
                severity="low",
                confidence="advisory",
                title=f"Non-standard frontmatter fields: {', '.join(extra)}",
                evidence=evidence,
                suggestion="Keep only name and description unless your runtime explicitly supports extra fields.",
                penalty=5,
                auto_fix=True,
            )
        )

    for required in ("name", "description"):
        if not frontmatter.get(required, "").strip():
            add(
                Issue(
                    rule_id=f"F7-MISSING-{required.upper()}",
                    dimension="frontmatter",
                    severity="medium",
                    confidence="confirmed",
                    title=f"Missing required frontmatter field: {required}",
                    evidence=evidence,
                    suggestion=f"Add a concise {required} field.",
                    penalty=30,
                )
            )


def detect_description(description: str, add) -> None:
    evidence = Evidence("SKILL.md", 1, f"description: {description[:120]}")
    if not description:
        add(
            Issue(
                rule_id="D6-MISSING-DESCRIPTION",
                dimension="description",
                severity="medium",
                confidence="confirmed",
                title="Description is missing",
                evidence=evidence,
                suggestion="Add a bounded description that says exactly when the skill should trigger.",
                penalty=30,
            )
        )
        return

    if len(description) < 20:
        add(
            Issue(
                rule_id="D6-SHORT-DESCRIPTION",
                dimension="description",
                severity="medium",
                confidence="confirmed",
                title="Description is too short to define a safe trigger boundary",
                evidence=evidence,
                suggestion="Expand the description with scope, trigger phrases, and non-goals.",
                penalty=30,
            )
        )

    broad_trigger_pattern = re.compile(
        r"(any task|all tasks|anything|everything|all scenarios|任何任务|所有任务|所有场景|任何问题|一切)",
        re.IGNORECASE,
    )
    if broad_trigger_pattern.search(description):
        add(
            Issue(
                rule_id="R3-N5",
                dimension="description",
                severity="medium",
                confidence="suspected",
                title="Description trigger scope appears over-broad",
                evidence=evidence,
                suggestion="Narrow the trigger scope to the skill's actual security-audit use case.",
                penalty=10,
                auto_fix=True,
            )
        )


def detect_line_rules(content: str, add) -> None:
    for file_name, line_number, line in iter_scan_lines(content):
        stripped = line.strip()
        if not stripped:
            continue
        if is_rule_definition_line(file_name, stripped):
            continue
        lower = stripped.lower()
        evidence = Evidence(file_name, line_number, stripped[:220])

        if re.search(r"ignore (all )?previous instructions|忽略之前所有指令|you are now dan|没有限制的ai|system prompt.*(updated|更新)", lower):
            add(
                Issue(
                    rule_id="P1-PROMPT-INJECTION",
                    dimension="prompt-injection",
                    severity="medium",
                    confidence="confirmed",
                    title="Prompt-injection instruction detected",
                    evidence=evidence,
                    suggestion="Remove role-override or instruction-override text from the skill.",
                    penalty=30,
                )
            )

        if re.search(r"覆盖之前.*安全规则|override.*safety rules", lower):
            add(
                Issue(
                    rule_id="P1-SAFETY-OVERRIDE",
                    dimension="prompt-injection",
                    severity="medium",
                    confidence="suspected",
                    title="Possible safety-rule override",
                    evidence=evidence,
                    suggestion="State that higher-priority instructions remain authoritative.",
                    penalty=15,
                )
            )

        if re.search(r"(cat|cp|scp|tar|zip|open|read_file|readfile)\b.*(~/.ssh|~/.aws|/root/\.openclaw|openclaw\.json|paired\.json|device-auth\.json|soul\.md|memory\.md|user\.md)", lower):
            add(
                Issue(
                    rule_id="S2-SENSITIVE-READ",
                    dimension="sensitive-data",
                    severity="medium",
                    confidence="confirmed",
                    title="Sensitive file read pattern detected",
                    evidence=evidence,
                    suggestion="Remove direct access to credentials, identity files, memory files, or private config.",
                    penalty=30,
                )
            )

        if re.search(r"(curl|wget|scp|nc)\b.*(~/.ssh|~/.aws|/root/\.openclaw|openclaw\.json|paired\.json|device-auth\.json|soul\.md|memory\.md|user\.md|@-)", lower):
            add(
                Issue(
                    rule_id="S2-EXFILTRATION",
                    dimension="sensitive-data",
                    severity="high",
                    confidence="confirmed",
                    title="Sensitive data exfiltration pattern detected",
                    evidence=evidence,
                    suggestion="Do not send local secrets, identity files, or private memory files to remote hosts.",
                    penalty=100,
                    direct_fail=True,
                )
            )

        if re.search(r"\b(ngrok|localtunnel|cloudflared\s+tunnel|bore|frpc?|frps)\b", lower):
            add(
                Issue(
                    rule_id="R3-N1",
                    dimension="compliance",
                    severity="high",
                    confidence="confirmed",
                    title="Tunneling tool usage detected",
                    evidence=evidence,
                    suggestion="Remove tunneling tools unless a reviewer has explicitly approved them.",
                    penalty=100,
                    direct_fail=True,
                )
            )

        if re.search(r"(hr|okr|performance|绩效|人才|人事).*(internal|corp|内网|内部|系统)", lower):
            add(
                Issue(
                    rule_id="R3-N2",
                    dimension="compliance",
                    severity="high",
                    confidence="confirmed",
                    title="Restricted HR/OKR system access detected",
                    evidence=evidence,
                    suggestion="Remove access to restricted organizational systems.",
                    penalty=100,
                    direct_fail=True,
                )
            )

        if re.search(r"(c4|绝密|内部高密).*(input|upload|paste|输入|上传|粘贴|处理)", lower):
            add(
                Issue(
                    rule_id="R3-N3",
                    dimension="compliance",
                    severity="high",
                    confidence="suspected",
                    title="High-sensitivity data handling detected",
                    evidence=evidence,
                    suggestion="State that the skill must not process highly sensitive data.",
                    penalty=100,
                    direct_fail=True,
                )
            )

        if re.search(r"(cp|tar|zip|upload|curl|备份|上传).*(openclaw\.json|device\.json|device-auth\.json)", lower):
            add(
                Issue(
                    rule_id="R3-N4",
                    dimension="compliance",
                    severity="high",
                    confidence="confirmed",
                    title="Sensitive config backup or upload pattern detected",
                    evidence=evidence,
                    suggestion="Do not back up, upload, or duplicate sensitive device/auth config.",
                    penalty=100,
                    direct_fail=True,
                )
            )

        if re.search(r"rm\s+-rf\s+/(?:\s|$)|rm\s+-rf\s+~|dd\s+if=/dev/zero", lower):
            add(
                Issue(
                    rule_id="M4-DESTRUCTIVE-COMMAND",
                    dimension="malicious-script",
                    severity="high",
                    confidence="confirmed",
                    title="Destructive command detected",
                    evidence=evidence,
                    suggestion="Remove destructive shell commands.",
                    penalty=100,
                    direct_fail=True,
                )
            )

        if re.search(r"(curl|wget)\b.*\|\s*(bash|sh)\b", lower):
            add(
                Issue(
                    rule_id="M4-REMOTE-SCRIPT-EXEC",
                    dimension="malicious-script",
                    severity="high",
                    confidence="confirmed",
                    title="Remote script execution detected",
                    evidence=evidence,
                    suggestion="Download artifacts separately, verify integrity, and avoid piping remote code to a shell.",
                    penalty=100,
                    direct_fail=True,
                )
            )

        if re.search(r"bash\s+-i.*(/dev/tcp|>&)|nc\s+-e\s+(/bin/)?(bash|sh)", lower):
            add(
                Issue(
                    rule_id="M4-REVERSE-SHELL",
                    dimension="malicious-script",
                    severity="high",
                    confidence="confirmed",
                    title="Reverse shell pattern detected",
                    evidence=evidence,
                    suggestion="Remove reverse shell behavior.",
                    penalty=100,
                    direct_fail=True,
                )
            )

        if re.search(r"sudo\s+su\b|chmod\s+777\s+/etc|xmrig|minerd|\.mining", lower):
            add(
                Issue(
                    rule_id="M4-PRIVILEGED-OR-MINING",
                    dimension="malicious-script",
                    severity="high",
                    confidence="confirmed",
                    title="Privileged shell or mining behavior detected",
                    evidence=evidence,
                    suggestion="Remove privilege escalation or mining behavior.",
                    penalty=100,
                    direct_fail=True,
                )
            )

        if re.search(r"\beval\b|\bexec\(", lower):
            add(
                Issue(
                    rule_id="M4-DYNAMIC-EXECUTION",
                    dimension="malicious-script",
                    severity="medium",
                    confidence="suspected",
                    title="Dynamic code execution detected",
                    evidence=evidence,
                    suggestion="Avoid dynamic code execution, or document why it is necessary and constrained.",
                    penalty=15,
                )
            )

        if re.search(r"\bpip\s+install\b.*(--index-url|--extra-index-url|https?://|git\+)", lower):
            add(
                Issue(
                    rule_id="D5-PIP-UNTRUSTED-SOURCE",
                    dimension="dependency",
                    severity="low",
                    confidence="advisory",
                    title="pip install uses a non-default source",
                    evidence=evidence,
                    suggestion="Pin dependencies and prefer the default package index unless a trusted mirror is required.",
                    penalty=5,
                )
            )

        if re.search(r"\bnpm\s+install\b.*(https?://|git\+|--registry)", lower):
            add(
                Issue(
                    rule_id="D5-NPM-UNTRUSTED-SOURCE",
                    dimension="dependency",
                    severity="low",
                    confidence="advisory",
                    title="npm install uses a non-default source",
                    evidence=evidence,
                    suggestion="Pin dependencies and prefer the default npm registry unless a trusted mirror is required.",
                    penalty=5,
                )
            )

        if re.search(r"\b(requets|reqeusts|lodahs|expres)\b", lower):
            add(
                Issue(
                    rule_id="D5-TYPOSQUATTING",
                    dimension="dependency",
                    severity="low",
                    confidence="advisory",
                    title="Possible typosquatted dependency name",
                    evidence=evidence,
                    suggestion="Check dependency names against the intended package before installation.",
                    penalty=5,
                )
            )


def is_rule_definition_line(file_name: str, line: str) -> bool:
    suffix = Path(file_name).suffix.lower()
    if suffix not in {".py", ".js", ".ts"}:
        return False
    return "re.search(" in line or "re.compile(" in line


def parse_frontmatter(text: str) -> dict[str, str]:
    lines = text.splitlines()
    if not lines or lines[0].strip() != "---":
        return {}

    frontmatter_lines: list[str] = []
    for line in lines[1:]:
        if line.strip() == "---":
            break
        frontmatter_lines.append(line)

    frontmatter_text = "\n".join(frontmatter_lines)
    try:
        import yaml  # type: ignore[import-not-found]

        parsed = yaml.safe_load(frontmatter_text) or {}
        if isinstance(parsed, dict):
            return {str(key).strip().lower(): "" if value is None else str(value) for key, value in parsed.items()}
    except Exception:
        pass

    values: dict[str, str] = {}
    index = 0
    while index < len(frontmatter_lines):
        line = frontmatter_lines[index]
        match = re.match(r"^([A-Za-z0-9_-]+):\s*(.*)$", line)
        if match:
            key = match.group(1).strip().lower()
            value = match.group(2).strip()
            if value in {">", ">-", ">+", "|", "|-", "|+"}:
                block_lines: list[str] = []
                index += 1
                while index < len(frontmatter_lines):
                    next_line = frontmatter_lines[index]
                    if re.match(r"^[A-Za-z0-9_-]+:\s*", next_line):
                        index -= 1
                        break
                    block_lines.append(next_line.strip())
                    index += 1
                if value.startswith("|"):
                    values[key] = "\n".join(line for line in block_lines if line)
                else:
                    values[key] = " ".join(line for line in block_lines if line)
            else:
                values[key] = value
        index += 1
    return values


def clean_yaml_scalar(value: str) -> str:
    value = value.strip()
    if len(value) >= 2 and value[0] == value[-1] and value[0] in {'"', "'"}:
        value = value[1:-1]
    return value.strip()


def extract_primary_skill_text(content: str) -> str:
    if "### FILE: " not in content:
        return content

    lines = content.splitlines()
    capture = False
    selected: list[str] = []
    for line in lines:
        marker = FILE_MARKER.match(line)
        if marker:
            capture = marker.group(1).replace("\\", "/").endswith("SKILL.md")
            if selected and capture:
                break
            continue
        if capture:
            selected.append(line)
    return "\n".join(selected) if selected else content


def iter_scan_lines(content: str) -> Iterable[tuple[str, int, str]]:
    current_file = "<input>"
    current_line = 0
    for raw_line in content.splitlines():
        marker = FILE_MARKER.match(raw_line)
        if marker:
            current_file = marker.group(1)
            current_line = 0
            continue
        current_line += 1
        yield current_file, current_line, raw_line


def severity_order(severity: str) -> int:
    return {"high": 0, "medium": 1, "low": 2}.get(severity, 3)


def calculate_score(issues: list[Issue]) -> tuple[int, str]:
    if any(issue.direct_fail for issue in issues):
        return 0, "F"

    score = 100
    confirmed_medium_count = 0
    advisory_penalty = 0
    for issue in issues:
        if issue.confidence == "advisory":
            advisory_penalty += issue.penalty
            continue
        score -= issue.penalty
        if issue.severity == "medium" and issue.confidence == "confirmed":
            confirmed_medium_count += 1

    score = max(0, score - min(advisory_penalty, 10))
    if confirmed_medium_count >= 2:
        score = min(score, 69)

    if score == 100:
        return score, "A"
    if score >= 85:
        return score, "B"
    if score >= 70:
        return score, "C"
    if score >= 50:
        return score, "D"
    return score, "F"


def render_text(reports: list[Report]) -> str:
    chunks: list[str] = []
    for report in reports:
        chunks.append(f"Skill Security Report: {report.target}")
        chunks.append(f"Scanner: {VERSION} ({REPORT_SCHEMA_VERSION})")
        chunks.append(f"Input SHA-256: {report.input_sha256}")
        chunks.append(f"Complete: {str(report.complete).lower()}")
        if report.ignored_rules:
            chunks.append(f"Ignored rules: {', '.join(report.ignored_rules)}")
        chunks.append(f"Rating: {report.rating} ({report.score}/100)")
        chunks.append("")
        if report.issues:
            chunks.append(f"Issues ({len(report.issues)}):")
            for issue in report.issues:
                chunks.append(
                    f"- [{issue.severity}/{issue.confidence}] {issue.rule_id}: {issue.title}"
                )
                chunks.append(
                    f"  evidence: {issue.evidence.file}:{issue.evidence.line}: {issue.evidence.text}"
                )
                chunks.append(f"  fix: {issue.suggestion}")
        else:
            chunks.append("Issues: none")
        chunks.append("")
        chunks.append("Passed dimensions:")
        for dimension in report.passed_dimensions:
            chunks.append(f"- {dimension}")
        chunks.append("")
    return "\n".join(chunks).rstrip()


def write_stdout(text: str) -> None:
    try:
        sys.stdout.write(text)
        sys.stdout.write("\n")
    except UnicodeEncodeError:
        sys.stdout.buffer.write((text + "\n").encode("utf-8"))


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv or sys.argv[1:])
    try:
        reports = scan_source(
            args.input,
            text=args.text,
            ignored_rules=set(args.ignore),
            include_references=args.include_references,
            max_zip_bytes=args.max_zip_bytes,
            max_zip_files=args.max_zip_files,
        )
    except (OSError, zipfile.BadZipFile, urllib.error.URLError, ScanError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2

    if args.format == "json":
        write_stdout(json.dumps([report.to_dict() for report in reports], indent=2, ensure_ascii=False))
    else:
        write_stdout(render_text(reports))

    return 1 if any(report.rating in {"D", "F"} for report in reports) else 0


if __name__ == "__main__":
    raise SystemExit(main())