#!/usr/bin/env python3
"""sosig 취약점 점검기 (배포본) — 서버 API 로 규칙을 받아 이 호스트를 점검한다.

받는 곳: https://sosig.site/vuln   ·   이 파일에는 DB 접속 정보가 들어 있지 않다.

준비
  1) 운영자에게 host_id 와 secret 을 발급받는다(자동 가입은 없다).
  2) 아래 내용으로 ~/.config/sosig-agent.conf 를 만들고 권한을 600 으로 둔다.

       host_id = h-xxxxxxxx
       secret  = ...
       server  = https://sosig.site

  3) 파이썬 3.10 이상이면 된다. 표준 라이브러리만 쓴다(추가 설치 불필요).

동작
  · POST /api/agent/rules 로 점검 규칙을 받는다. 요청과 응답 모두 HMAC-SHA256 으로 서명한다.
  · 받은 규칙으로 이 호스트를 점검하고 '검출 데이터 + 양호/취약/점검불가' 를 낸다.
  · 읽기 전용이다. 시스템을 고치지 않는다.

현재 제한 (2026-09-06 ~)
  · 자동 생성된 파이썬 점검 코드의 실행을 보류하고 있다. 정규식·AST 검사는 격리가
    아니므로, 검증된 OS 격리 환경을 갖추기 전까지 임의 코드를 실행하지 않는다.
  · 셸 점검도 버전 조회 명령(uname -r, openssl version 등)으로 제한한다.
  · 서버가 보류 상태를 서명해 알려주면 우회 경로 없이 종료한다(종료 코드 75).

사용
  vuln_check.py                      # 이 호스트 점검
  vuln_check.py --engine-py never    # 셸 규칙만 사용
  vuln_check.py --json out.json      # 결과를 JSON 으로도 저장

주의
  · '취약' 은 확정이 아니라 확인이 필요한 후보다. 배포판이 패치를 백포트한 경우가 흔하다.
  · 파이썬 점검 코드는 자동 생성물이라 오탐이 있다. 근거 줄을 함께 읽을 것.
"""
from __future__ import annotations
import argparse, ast, json, os, re, shlex, signal, socket, subprocess, sys, tempfile, time
from datetime import datetime
from pathlib import Path

BASE = Path(__file__).resolve().parent
RULES = BASE / "rules.json"

class ExecutionPaused(RuntimeError):
    pass

# 루트 전체 스캔 차단. 2026-08-21 에 이런 코드가 남긴 find 1,207개가 부하를 1,200까지 올려
# NAS 와 뉴스 파이프라인을 10시간 멈춰 세웠다. sosig_agent 에는 있었는데 여기엔 빠져 있어
# 배포본이 프로덕션 서버에서 그대로 돌 뻔했다(2026-08-22 발견). 해당 명령은 실행하지 않고
# '점검불가' 로 남긴다 — 실패를 안전으로 오인하지 않기 위해서다.
FULLSCAN = re.compile(r"""(?x)
    ["'\s\[]find["'\s,\]]+.{0,12}["']/["'\s,\]]
  | find\s+/\s+-
  | ["']/["']\s*,\s*["']-(name|type|perm)
  | grep\s+-[rR][^|;]{0,20}\s/\s
  | /mnt/nas | os\.walk\(\s*["']/["']\s*\)
""")

# ── 파이썬 점검 코드 격리 실행기 (2026-08-22) ──────────────────────────────
# sosig_agent.py 에 있던 것을 그대로 이식했다. 배포본은 단일 파일이라 그 모듈이 없고,
# 없으면 파이썬 경로를 통째로 못 써서 백포트 확인이 안 되고 오탐이 크게 늘었다.
# 원본과 동작이 달라지면 안 되므로 손대지 말 것 — 고칠 일이 생기면 양쪽을 같이 고친다.
TIMEOUT_EACH = 15

DANGER = re.compile(r"(?i)\b(rm\s+-|shutil\.rmtree|os\.remove|os\.unlink|os\.rmdir|mkfs|shutdown|"
                    r"reboot|halt|poweroff|systemctl\s+(stop|restart|disable|mask)|chmod|chown|"
                    r"urlopen|urlretrieve|requests\.|socket\.|smtplib|ftplib|paramiko|pickle|"
                    r"eval\(|exec\(|__import__|open\([^)]*['\"][wa])")

def safe_to_run(code):
    """받은 코드 자체 검사: 문법 · 금지 패턴 · check() 시그니처."""
    if "def check" not in code:
        return False, "check() 없음"
    try:
        tree = ast.parse(code)
    except SyntaxError as e:
        return False, f"문법 오류: {str(e)[:60]}"
    m = DANGER.search(code)
    if m:
        return False, f"금지 패턴: {m.group(0)[:30]}"
    m = FULLSCAN.search(code)
    if m:
        return False, f"전체 스캔: {m.group(0)[:30].strip()}"
    for node in ast.walk(tree):
        if isinstance(node, ast.FunctionDef) and node.name == "check":
            a = node.args
            if len(a.posonlyargs) + len(a.args) - len(a.defaults) > 0:
                return False, "check() 가 인자를 요구함"
            return True, ""
    return False, "check() 정의 없음"

SECRET_NAME = r"(?:SECRET|TOKEN|PASSWORD|PASSWD|API[_-]?KEY|ACCESS[_-]?KEY|PRIVATE[_-]?KEY|CREDENTIAL|AUTH)"

REDACTORS = [
    # NAME=값 / NAME: 값  → 값을 가린다
    (re.compile(rf"(?i)([A-Z0-9_]*{SECRET_NAME}[A-Z0-9_]*)\s*[=:]\s*\S+"), r"\1=***"),
    # 이름만 나열된 경우도 가린다(어떤 비밀이 있는지 알려주는 것 자체가 단서다)
    (re.compile(rf"(?i)\b[A-Z0-9_]*{SECRET_NAME}[A-Z0-9_]*\b"), "***"),
    (re.compile(r"-----BEGIN[^-]{0,40}-----.*?-----END[^-]{0,40}-----", re.S), "***"),
    (re.compile(r"\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.?[A-Za-z0-9_-]*"), "***"),  # JWT
    (re.compile(r"\b[A-Fa-f0-9]{32,}\b"), "***"),                                            # 긴 16진 문자열
    (re.compile(r"\b[A-Za-z0-9+/]{40,}={0,2}\b"), "***"),                                    # 긴 base64
]

def redact(text):
    """비밀값으로 보이는 것을 가린다. 서버로 보내기 전 마지막 관문."""
    out = str(text)
    for pat, rep in REDACTORS:
        out = pat.sub(rep, out)
    return out

def _kill_tree(proc):
    """프로세스 그룹째 정리한다. 자식을 남기면 타임아웃이 무의미해진다."""
    try:
        os.killpg(proc.pid, signal.SIGKILL)
    except (ProcessLookupError, PermissionError):
        pass
    try:
        proc.wait(timeout=5)
    except Exception:
        pass

def run_check(code):
    """점검 1건 실행. 반환값은 반드시 redact 를 거친다."""
    r = _run_check_raw(code)
    return {k: (redact(v) if k in ("detected", "reason") else v) for k, v in r.items()}

def _run_check_raw(code):
    """Fail closed: namespace isolation is unavailable on the current host.

    Regex/AST checks are not a sandbox. There is deliberately no environment
    variable or automatic fallback to executing arbitrary generated Python.
    """
    return {"verdict": "점검불가", "detected": "-",
            "reason": "실행 보류: 검증된 OS 격리 환경이 필요합니다."}


_run_py_check = run_check


# ── 안전장치: DB 에 저장된 명령을 그대로 실행하므로 파괴적/외부변경 동작은 차단 ──
DENY = re.compile(
    r"""(?ix)
    \b(rm|rmdir|mkfs|dd|shutdown|reboot|halt|poweroff|init|kill|pkill|killall)\b
  | \b(useradd|userdel|usermod|passwd|chpasswd|visudo)\b
  | \b(iptables|nft|ufw)\s+(-[AIDF]|add|delete|flush)
  | \b(systemctl|service)\s+(stop|restart|disable|mask)
  | \bcurl\b[^|]*\s(-X\s*(POST|PUT|DELETE)|--data|-d\s)
  | \bwget\b[^|]*\s-O\s
  | (?<![0-9])>\s*/(etc|usr|bin|sbin|boot|sys|proc)/      # 시스템 경로 쓰기 (2>/dev/null 같은 표준에러 버리기는 제외)
  | \bchmod\b|\bchown\b|\bmv\b\s+/|\btruncate\b
  | \b(nc|ncat|netcat)\b.*\s-e\b
""")



# ── 서버 API (2026-08-22) ────────────────────────────────────────────────
# 예전에는 이 파일이 MySQL 에 직접 붙었다. 점검 대상이 NAS·외부 호스트로 늘면
# 그 호스트마다 DB admin 자격증명을 심어야 해서, 에이전트와 같은 HMAC 방식으로 바꿨다.
# 필요한 것은 ~/.config/sosig-agent.conf 의 host_id·secret 뿐이다.
AGENT_CONF_PATHS = ("/etc/sosig-agent.conf", str(Path.home() / ".config/sosig-agent.conf"))
USER_AGENT = "sosig-agent/1.0 (+https://sosig.site/vuln)"
_API_CACHE: dict = {}


def load_agent_cfg():
    """에이전트 설정(host_id·secret·server)을 읽는다."""
    import configparser
    for path in AGENT_CONF_PATHS:
        if Path(path).exists():
            cp = configparser.ConfigParser()
            cp.read_string("[a]\n" + Path(path).read_text(encoding="utf-8"))
            a = cp["a"]
            if a.get("host_id") and a.get("secret"):
                return (a["host_id"].strip(), a["secret"].strip(),
                        (a.get("server") or "https://sosig.site").strip())
    raise RuntimeError("에이전트 설정 없음 — ~/.config/sosig-agent.conf 에 host_id/secret 을 두세요")


def fetch_rules_api(host: str | None):
    """/api/agent/rules 를 한 번 호출해 규칙·NVD 범위·호스트 명령·파이썬 코드를 모두 받는다.

    응답 서명을 확인한다 — 서버가 준 것이 맞는지 클라이언트가 검증할 수 있어야 한다.
    같은 인자로 두 번 부르지 않도록 결과를 캐시한다(로더 4개가 각각 부르기 때문).
    """
    key = host or ""
    if key in _API_CACHE:
        return _API_CACHE[key]
    import hashlib, hmac, urllib.request
    host_id, secret, server = load_agent_cfg()
    payload = {"host_id": host_id, "ts": int(time.time())}
    if host:
        payload["host"] = host
    body = json.dumps(payload).encode()
    sig = "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
    req = urllib.request.Request(server + "/api/agent/rules", data=body, headers={
        "Content-Type": "application/json", "X-Signature": sig, "User-Agent": USER_AGENT})
    with urllib.request.urlopen(req, timeout=60) as r:
        raw = r.read()
        srv_sig = r.headers.get("X-Signature", "")
    want = "sha256=" + hmac.new(secret.encode(), raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(srv_sig, want):
        raise RuntimeError("서버 응답 서명 불일치 — 받은 규칙을 쓰지 않는다")
    data = json.loads(raw)
    if data.get('execution_paused'):
        raise ExecutionPaused('실행 격리 준비 중; 파일·셸 규칙으로 우회하지 않습니다.')
    _API_CACHE[key] = data
    return data

def load_rules_db(host: str | None = None):
    """서버 API 에서 규칙을 받아 파일 스키마와 같은 형태로 돌려준다."""
    rows = fetch_rules_api(host)["rules"]
    out = []
    for r in rows:
        rule = json.loads(r["rule_json"]) if isinstance(r["rule_json"], str) else (r["rule_json"] or {})
        out.append({"cve_id": r["cve_id"], "engine": r["engine"], "severity": r["severity"],
                    "hosts": r["hosts"], "nvt": r["nvt_name"], "command": r["command"],
                    "extract_regex": rule.get("extract_regex"),
                    "verdict": rule.get("verdict") or {"type": "always_manual"},
                    "note": r["note"]})
    return out


def load_rules(source: str):
    """(규칙목록, 출처설명) — DB 실패 시 파일로 폴백."""
    if source != "file":
        try:
            rules = load_rules_db()
            if rules:
                return rules, f"서버 API ({len(rules)}건)"
        except ExecutionPaused:
            raise
        except Exception as e:
            print(f"[알림] 서버에서 규칙을 받지 못해 로컬 rules.json 으로 넘어갑니다: {str(e)[:70]}", file=sys.stderr)
    if not RULES.exists():
        return [], "규칙 없음"
    return json.loads(RULES.read_text(encoding="utf-8")), f"파일 {RULES.name}"


def load_nvd_ranges(host: str | None = None):
    """{cve_id: [range,...]}. 실패하면 빈 dict (LLM 규칙으로 진행)."""
    try:
        rows = fetch_rules_api(host)["nvd_ranges"]
    except Exception:
        return {}
    out = {}
    for r in rows:
        out.setdefault(r["cve_id"], []).append(r)
    return out


def load_py_codes(host: str | None = None):
    """{cve_id: py_code}. 서버가 syntax_ok=1 AND safe_ok=1 만 내려준다. 실패 시 빈 dict."""
    try:
        rows = fetch_rules_api(host)["rules"]
    except Exception:
        return {}
    return {r["cve_id"]: r["py_code"] for r in rows if r.get("py_code")}


def in_nvd_range(ver: str, rng: dict) -> bool | None:
    """버전이 NVD 범위 안인가."""
    checks = []
    if rng.get("start_incl"):
        checks.append(cmp_version(ver, rng["start_incl"], ">="))
    if rng.get("start_excl"):
        checks.append(cmp_version(ver, rng["start_excl"], ">"))
    if rng.get("end_incl"):
        checks.append(cmp_version(ver, rng["end_incl"], "<="))
    if rng.get("end_excl"):
        checks.append(cmp_version(ver, rng["end_excl"], "<"))
    if not checks:
        return cmp_version(ver, rng["ver_exact"], "==") if rng.get("ver_exact") else None
    if any(c is None for c in checks):
        return None
    return all(checks)


def decide_by_nvd(ranges: list, raw: str, data: str):
    """NVD 범위로 판정. (검출데이터, 판정, 사유) 또는 None(판정 불가)."""
    bounded = [r for r in ranges if any(r.get(k) for k in ("start_incl", "start_excl", "end_incl", "end_excl"))]
    use = bounded or [r for r in ranges if r.get("ver_exact")]
    if not use:
        return None
    ref = None
    for r in use:
        ref = r.get("start_incl") or r.get("end_excl") or r.get("end_incl") or r.get("ver_exact")
        if ref:
            break
    ver = pick_version(data, ref) or pick_version(raw, ref)
    if not ver:
        return None
    unknown = False
    for r in use:
        match = in_nvd_range(ver, r)
        if match is None:
            unknown = True
        if match is True:
            span = f"{r.get('start_incl') or r.get('start_excl') or '-'} ~ {r.get('end_excl') or r.get('end_incl') or r.get('ver_exact') or '-'}"
            return ver, VERDICT_BAD, f"NVD: {r.get('product')} {ver} ∈ [{span}]"
    if unknown:
        return ver, VERDICT_NA, 'NVD 영향 범위 비교 불가: 양호로 확정하지 않음'
    return ver, VERDICT_OK, f"NVD: {use[0].get('product')} {ver} 는 영향 범위 밖({len(use)}개 범위 확인)"



def load_host_cmds(host: str | None):
    """{product: command} — 이 호스트에서 제품 버전을 얻는 명령."""
    if not host:
        return {}
    try:
        rows = fetch_rules_api(None)["host_cmds"]   # 전체 조회 캐시를 재사용
    except Exception:
        return {}
    return {r["product"]: r["command"] for r in rows if r.get("host") == host}


def product_of(nvd_rows: list | None):
    """이 CVE 가 어떤 제품에 대한 것인가 (NVD 범위에서 최빈 제품)."""
    if not nvd_rows:
        return None
    names = [r.get("product") for r in nvd_rows if r.get("product")]
    if not names:
        return None
    return max(set(names), key=names.count)


VERDICT_OK, VERDICT_BAD, VERDICT_NA = "양호", "취약", "점검불가"



# 저장된 명령 상당수가 판정 단어를 직접 출력한다(VULNERABLE / PATCHED / 취약 / 양호).
# 규칙이 값을 못 뽑았을 때 이 단어를 마지막 근거로 쓴다 — '점검불가'로 버리는 것보다 정확하다.
BAD_WORDS = re.compile(r"(?i)\b(vulnerable|affected|취약|위험)\b")
OK_WORDS = re.compile(r"(?i)\b(patched|not\s+vulnerable|not\s+affected|safe|양호|안전|정상)\b")


def keyword_fallback(raw: str):
    """(판정, 사유) 또는 None."""
    t = raw or ""
    ok, bad = OK_WORDS.search(t), BAD_WORDS.search(t)
    if ok and not bad:
        return VERDICT_OK, f"출력의 '{ok.group(0)}' 로 판정(폴백)"
    if bad and not ok:
        return VERDICT_BAD, f"출력의 '{bad.group(0)}' 로 판정(폴백)"
    return None


def norm_version(v: str):
    """'3.4p1' → (3,4,1) 형태로 비교 가능하게."""
    parts = re.findall(r"\d+", v or "")
    return tuple(int(p) for p in parts) if parts else None



VER_RE = re.compile(r"\d+(?:\.\d+){1,3}(?:p\d+)?")


def pick_version(text: str, ref: str | None):
    """출력에서 버전 후보를 모두 뽑고, 기준값(ref)과 major 가 같은 것을 고른다.
    'mysql Ver 15.1 Distrib 10.3.32-MariaDB' 처럼 여러 숫자가 섞인 출력에서
    엉뚱한 값(15.1)을 집던 문제를 막는다."""
    cands = VER_RE.findall(text or "")
    if not cands:
        return None
    if ref:
        rmaj = norm_version(ref)
        if rmaj:
            same = [c for c in cands if (norm_version(c) or (None,))[0] == rmaj[0]]
            if same:
                return same[0]
    return cands[-1] if len(cands) > 1 else cands[0]


def cmp_version(a: str, b: str, op: str) -> bool | None:
    x, y = norm_version(a), norm_version(b)
    if not x or not y:
        return None
    n = max(len(x), len(y))
    x = x + (0,) * (n - len(x))
    y = y + (0,) * (n - len(y))
    return {"<": x < y, "<=": x <= y, ">": x > y, ">=": x >= y, "==": x == y}.get(op)


def run_command(cmd: str, timeout: int = 20, ssh_host: str | None = None):
    probes = {('uname', '-r'), ('ssh', '-V'), ('nginx', '-v'), ('openssl', 'version'),
              ('python3', '-V'), ('git', '--version'), ('curl', '--version')}
    try:
        argv = shlex.split(cmd)
    except ValueError:
        return None, '점검 명령 형식 오류'
    if ssh_host or tuple(argv) not in probes:
        return None, '실행 보류: 임의 셸 명령은 격리 환경 확인 전 실행하지 않습니다.'
    m = FULLSCAN.search(cmd or "")
    if m:
        return "", f"루트 전체 스캔이라 실행하지 않음: {m.group(0).strip()[:30]}"
    """검출 데이터 수집. (출력, 오류) 반환. ssh_host 를 주면 원격에서 실행한다."""
    if DENY.search(cmd):
        return None, "안전장치 차단(파괴적/변경 명령 포함)"
    try:
        p = subprocess.run(argv, capture_output=True, text=True, timeout=timeout,
                           env={'PATH': '/usr/sbin:/usr/bin:/sbin:/bin', 'LC_ALL': 'C'})
    except subprocess.TimeoutExpired:
        return None, f"시간초과({timeout}초)"
    except Exception as e:
        return None, f"실행오류: {str(e)[:60]}"
    if p.returncode != 0:
        return None, f'점검 명령 실패(exit={p.returncode})'
    out = (p.stdout or "").strip() or (p.stderr or "").strip()
    if not out:
        return None, '점검 명령 출력 없음'
    return out, None


def decide(rule: dict, raw: str, nvd: list | None = None):
    """검출 데이터와 판정을 만든다. 반환 (검출데이터, 판정, 사유).
    NVD 범위가 있으면 그것을 우선한다 — LLM 이 추측한 범위보다 권위 있는 근거다."""
    v = rule.get("verdict") or {}
    vtype = v.get("type", "always_manual")
    data = raw

    rx = rule.get("extract_regex")
    if rx:
        try:
            m = re.search(rx, raw or "", re.M)
            if m:
                data = (m.group(1) if m.groups() else m.group(0)).strip()
        except re.error:
            pass

    if nvd:
        got = decide_by_nvd(nvd, raw, data)
        if got:
            return got

    if vtype == "always_manual":
        fb = keyword_fallback(raw)
        if fb:
            return data, fb[0], fb[1]
        return data, VERDICT_NA, "자동 판정 규칙 없음 — 사람이 확인"

    if vtype == "empty":
        cond = not (raw or "").strip()
        bad = cond if v.get("vulnerable_if_true", False) else not cond
        return data, (VERDICT_BAD if bad else VERDICT_OK), ("출력 없음" if cond else "출력 있음")

    if vtype == "contains":
        needle = str(v.get("needle", ""))
        cond = needle.lower() in (raw or "").lower()
        bad = cond if v.get("vulnerable_if_true", True) else not cond
        return data, (VERDICT_BAD if bad else VERDICT_OK), f"'{needle}' {'발견' if cond else '없음'}"

    if vtype == "version_range":
        lo, hi = v.get("min"), v.get("max")
        cand = pick_version(data, lo or hi) or pick_version(raw, lo or hi)
        if not cand:
            fb = keyword_fallback(raw)
            if fb:
                return data, fb[0], fb[1]
            return data, VERDICT_NA, "버전 값을 찾지 못함"
        ge = cmp_version(cand, lo, ">=") if lo else True
        le = cmp_version(cand, hi, "<=") if hi else True
        if ge is None or le is None:
            fb = keyword_fallback(raw)      # 출력이 버전이 아니라 판정어인 경우
            if fb:
                return data, fb[0], fb[1]
            return cand, VERDICT_NA, "버전 비교 불가"
        inside = bool(ge and le)
        bad = inside if v.get("vulnerable_if_true", True) else not inside
        span = f"{lo or '-'} ~ {hi or '-'}"
        return cand, (VERDICT_BAD if bad else VERDICT_OK), f"{cand} in [{span}] = {inside}"

    if vtype == "version":
        target, op = v.get("target", ""), v.get("op", "<")
        cand = pick_version(data, target) or pick_version(raw, target)
        if not cand:
            fb = keyword_fallback(raw)
            if fb:
                return data, fb[0], fb[1]
            return data, VERDICT_NA, "버전 값을 찾지 못함"
        res = cmp_version(cand, target, op)
        if res is None:
            fb = keyword_fallback(raw)
            if fb:
                return data, fb[0], fb[1]
            return cand, VERDICT_NA, "버전 비교 불가"
        bad = res if v.get("vulnerable_if_true", True) else not res
        return cand, (VERDICT_BAD if bad else VERDICT_OK), f"{cand} {op} {target} = {res}"

    return data, VERDICT_NA, f"알 수 없는 규칙({vtype})"


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--all", action="store_true", help="호스트 필터 없이 전부")
    ap.add_argument("--host", help="이 호스트 IP 로 필터 (기본: 자동 감지)")
    ap.add_argument("--json", help="결과 JSON 저장 경로")
    ap.add_argument("--only", help="특정 CVE 만 (쉼표 구분)")
    ap.add_argument("--source", choices=["db", "file"], default="db", help="규칙 출처 (기본 db)")
    ap.add_argument("--engine", help="엔진으로 거르기 (예: 1,2 / 0=자동화 불가)")
    ap.add_argument("--via-ssh", dest="via_ssh",
                    help="원격 호스트에서 실행 (~/.ssh/config 별칭) — 현재 보류 중")
    ap.add_argument("--engine-py", dest="engine_py", choices=["never", "auto", "only"], default="auto",
                    help="파이썬 점검 경로 사용 (기본 auto: py_code 있으면 사용, never: 사용하지 않음, only: py_code 있는 항목만)")
    a = ap.parse_args()

    if a.via_ssh:
        # 원격 셸 실행은 격리 환경 검증 전까지 보류한다(2026-09-06).
        # 그대로 진행하면 모든 항목이 조용히 '점검불가'가 되어 결과를 오해하게 된다.
        print("원격 점검(--via-ssh)은 현재 보류 중입니다: 검증된 OS 격리 환경이 필요합니다.\n"
              "대상 호스트에서 이 파일을 직접 실행하세요.", file=sys.stderr)
        return 75

    try:
        rules, origin = load_rules(a.source)
    except ExecutionPaused as exc:
        print('점검 보류:', exc, file=sys.stderr)
        return 75
    nvd_map = load_nvd_ranges()
    host_cmd = load_host_cmds(a.host)
    if host_cmd:
        print(f"호스트 전용 명령 {len(host_cmd)}개 적용 ({a.host})")
    if nvd_map:
        origin += f" + NVD 범위 {len(nvd_map)}CVE"
    if not rules:
        print("점검 규칙을 받지 못했습니다.\n"
              "  · ~/.config/sosig-agent.conf 에 host_id/secret/server 가 있는지 확인하세요(권한 600).\n"
              "  · 발급이 필요하면 https://sosig.site/vuln 안내를 따르세요.\n"
              "  · --source file 로 쓰려면 이 파일과 같은 폴더에 rules.json 이 있어야 합니다.",
              file=sys.stderr)
        return 1
    if a.engine:
        want_e = {int(x) for x in a.engine.split(",") if x.strip().isdigit()}
        rules = [r for r in rules if int(r.get("engine", 1)) in want_e]

    # 이 호스트의 IP 는 여러 개일 수 있다(유선+tailnet 등). 스캔 기록과 대조하려면 전부 모아야 한다.
    myips = [a.host] if a.host else []
    if not myips and not a.all:
        try:
            out = subprocess.run(["ip", "-4", "-o", "addr"], capture_output=True, text=True, timeout=5).stdout
            myips = re.findall(r"inet (\d+\.\d+\.\d+\.\d+)", out)
        except Exception:
            pass
        myips = [x for x in myips if not x.startswith("127.")]
    myip = ", ".join(myips)
    if a.only:
        want = {x.strip().upper() for x in a.only.split(",")}
        rules = [r for r in rules if r["cve_id"].upper() in want]
    elif not a.all:
        rules = [r for r in rules if any(ip in (r.get("hosts") or "") for ip in myips)]

    # 파이썬 점검 코드 로드 (로컬 점검만 가능, 원격에서는 사용하지 않음)
    py_codes = {}
    if a.engine_py != "never" and not a.via_ssh:
        py_codes = load_py_codes()
        if py_codes:
            origin += f" + py_code {len(py_codes)}건"

    # engine-py=only: py_code 가 있는 항목만 남김
    if a.engine_py == "only" and not a.via_ssh:
        rules = [r for r in rules if r["cve_id"] in py_codes]

    print(f"규칙 출처: {origin}")
    if a.via_ssh:
        print(f"실행 위치: 원격 {a.via_ssh}")
    print(f"점검 대상 {len(rules)}건" + (f" (호스트 {myip})" if myip and not a.all else " (전체)"))
    print("=" * 96)
    if a.engine_py != "never" and _run_py_check is None:
        print("[알림] sosig_agent 모듈이 없어 파이썬 점검 경로를 쓸 수 없습니다 — 셸 규칙으로만 점검합니다.",
              file=sys.stderr)
    results, tally = [], {VERDICT_OK: 0, VERDICT_BAD: 0, VERDICT_NA: 0}
    py_count, sh_count = 0, 0

    for r in sorted(rules, key=lambda x: ({"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3}.get(x.get("severity", ""), 4), x["cve_id"])):
        prod = product_of(nvd_map.get(r["cve_id"]))
        cmd = host_cmd.get(prod) or r["command"]      # 호스트 전용 명령이 있으면 그것으로
        used_override = cmd != r["command"]

        # 파이썬 점검 경로: 로컬 + engine-py 가 auto/only + py_code 존재
        use_py = (a.engine_py != "never" and not a.via_ssh
                  and r["cve_id"] in py_codes and _run_py_check is not None)

        if use_py:
            try:
                res = _run_py_check(py_codes[r["cve_id"]])
                data = res.get("detected", "")
                verdict = res.get("verdict", VERDICT_NA)
                why = res.get("reason", "")
            except Exception as e:
                data, verdict, why = "-", VERDICT_NA, f"파이썬 점검 오류: {str(e)[:60]}"
            path = "py"
            py_count += 1
        else:
            # 기존 셸 명령 경로 (변경 없음)
            raw, err = run_command(cmd, ssh_host=a.via_ssh)
            if err:
                data, verdict, why = "-", VERDICT_NA, err
            else:
                data, verdict, why = decide(r, raw, nvd_map.get(r["cve_id"]))
            path = "sh"
            sh_count += 1

        tally[verdict] += 1
        mark = {"양호": "✅", "취약": "❌", "점검불가": "⚠️"}[verdict]
        one = (data or "").replace("\n", " ⏎ ")[:70] or "(출력 없음)"
        print(f"{mark} {r['cve_id']:<18} [{(r.get('severity') or '?'):<8}] {verdict} [{path}]")
        print(f"     검출 데이터: {one}")
        tag = f" [{prod} 전용명령]" if used_override else ""
        print(f"     근거: {why}{tag}  |  {(r.get('nvt') or '')[:52]}")
        results.append({"cve_id": r["cve_id"], "engine": r.get("engine"), "target": a.via_ssh or "local",
                        "severity": r.get("severity"), "nvt": r.get("nvt"),
                        "command": r["command"], "detected": data, "verdict": verdict,
                        "reason": why, "path": path,
                        "checked_at": datetime.now().isoformat(timespec="seconds")})

    print("=" * 96)
    print(f"합계 {len(results)}건 — ✅ 양호 {tally[VERDICT_OK]} / ❌ 취약 {tally[VERDICT_BAD]} / ⚠️ 점검불가 {tally[VERDICT_NA]}")
    print(f"점검 경로: 파이썬 {py_count}건 / 셸 {sh_count}건")
    if a.json:
        Path(a.json).write_text(json.dumps(results, ensure_ascii=False, indent=1), encoding="utf-8")
        print("JSON 저장:", a.json)
    return 0


if __name__ == "__main__":
    sys.exit(main())