#!/usr/bin/env python3 """sosig 점검 에이전트 — 리눅스 클라이언트 (2026-08-17). 동작 ① 설치 조사 : 패키지 목록 + 실제 실행파일 버전(php, nginx, openssl …)을 모은다 ② 코드 수신 : 서버에 목록을 주고 해당 제품의 점검 코드만 받는다 (HMAC 서명) ③ 자체 검사 : 받은 코드를 다시 검사한다 — 서버를 믿지 않는다 ④ CVE 실행 : OS 격리 환경 검증 전 보류(2026-09-06). KISA 경로는 별도. ⑤ 보고 : 결과를 서명해 서버로 보낸다 설정 (둘 중 하나) 환경변수 SOSIG_HOST_ID, SOSIG_SECRET, (선택) SOSIG_SERVER 설정파일 /etc/sosig-agent.conf 또는 ~/.config/sosig-agent.conf host_id = h-xxxx secret = ... server = https://sosig.site 사용 sosig_agent.py # 조사 → 수신 → 실행 → 보고 sosig_agent.py --dry-run # 실행까지만 하고 보고하지 않음 sosig_agent.py --inventory # 설치 조사 결과만 출력 """ import argparse import ast import configparser import hashlib import hmac import io import json import os import platform import re import shutil import signal import socket import subprocess import sys import tempfile import time import urllib.error import urllib.request from pathlib import Path DEFAULT_SERVER = "https://sosig.site" # Cloudflare 가 파이썬 기본 UA(Python-urllib)를 봇으로 보고 1010 으로 막는다 — 식별 가능한 UA 를 쓴다. USER_AGENT = "sosig-agent/1.0 (+https://sosig.site/vuln)" TIMEOUT_EACH = 15 TIMEOUT_TOTAL = 900 STATE = Path(os.environ.get("SOSIG_STATE", str(Path.home() / ".cache/sosig-agent/cursor"))) # 루트 전체 스캔은 서버를 마비시킨다. 2026-08-21 에 이런 코드가 남긴 find 1,207개가 # 부하를 1,200까지 올려 NAS 와 뉴스 파이프라인을 10시간 멈춰 세웠다. # DB 에 있는 옛 코드에는 아직 이런 게 남아 있으므로, 받는 쪽에서 막는다. 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*\) """) # 서버가 보낸 코드라도 그대로 믿지 않는다. 클라이언트에서 한 번 더 막는다. 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])") # 실행파일 → NVD 제품명. 여기 있는 것만 조사한다(모르는 제품은 서버가 어차피 못 매칭). PROBES = [ ("php", ["php", "-v"], r"PHP (\d+\.\d+\.\d+)"), ("nginx", ["nginx", "-v"], r"nginx/(\d+\.\d+\.\d+)"), ("apache_http_server", ["apache2", "-v"], r"Apache/(\d+\.\d+\.\d+)"), ("openssl", ["openssl", "version"], r"OpenSSL (\d+\.\d+\.\d+\w*)"), ("openssh", ["ssh", "-V"], r"OpenSSH_(\d+\.\d+\w*\d*)"), ("mariadb", ["mysql", "--version"], r"Distrib (\d+\.\d+\.\d+)"), ("mysql", ["mysqld", "--version"], r"Ver (\d+\.\d+\.\d+)"), ("postgresql", ["psql", "--version"], r"(\d+\.\d+)"), ("python", ["python3", "-V"], r"Python (\d+\.\d+\.\d+)"), ("git", ["git", "--version"], r"git version (\d+\.\d+\.\d+)"), ("curl", ["curl", "--version"], r"curl (\d+\.\d+\.\d+)"), ("bash", ["bash", "--version"], r"version (\d+\.\d+\.\d+)"), ("sudo", ["sudo", "-V"], r"version (\d+\.\d+\.\d+)"), ("redis", ["redis-server", "-v"], r"v=(\d+\.\d+\.\d+)"), ("docker", ["docker", "--version"], r"version (\d+\.\d+\.\d+)"), ("node.js", ["node", "-v"], r"v(\d+\.\d+\.\d+)"), ] def norm_product(name): """제품 이름을 비교용으로 납작하게. 'Apache HTTP Server' → 'apachehttpserver'. 서버는 감시 목록 이름('Node.js')을, 이쪽은 탐지 이름('node.js')을 쓰므로 양쪽을 맞춘다.""" return re.sub(r"[^a-z0-9]+", "", str(name).lower()) def in_inventory(product, inv_norm): """이 서버에 그 제품이 있는가. 없으면 점검할 이유가 없다. 서버는 감시 제품 키워드로 느슨하게 묶어 보내기 때문에 .NET·Azure·IBM WebSphere 처럼 이 서버에 있을 수 없는 제품의 점검까지 딸려온다. 실측 205건 중 113건이 그런 경우였고, 대상이 없으니 코드가 옆에 있는 아무 버전이나 집어 오탐을 냈다. 이름이 정확히 맞을 때만 통과시킨다 — 부분 일치는 mysql/Microsoft SQL Server 처럼 잘못 붙는다. """ return norm_product(product) in inv_norm def load_config(): cfg = {"server": os.environ.get("SOSIG_SERVER", DEFAULT_SERVER), "host_id": os.environ.get("SOSIG_HOST_ID", ""), "secret": os.environ.get("SOSIG_SECRET", "")} for path in ("/etc/sosig-agent.conf", str(Path.home() / ".config/sosig-agent.conf")): if os.path.exists(path): cp = configparser.ConfigParser() cp.read_string("[a]\n" + Path(path).read_text(encoding="utf-8")) for k in ("host_id", "secret", "server"): if cp["a"].get(k): cfg[k] = cp["a"][k].strip() if not cfg["host_id"] or not cfg["secret"]: sys.exit("설정 없음: SOSIG_HOST_ID/SOSIG_SECRET 환경변수나 /etc/sosig-agent.conf 를 두세요") return cfg def probe_versions(): """실제로 설치돼 실행되는 것만 잡는다(패키지만 깔리고 안 쓰는 경우 제외).""" found = {} for product, argv, pattern in PROBES: if not shutil.which(argv[0]): continue try: # 한국어 로캘은 'GNU bash, 버전 5.2.21'·'Sudo 버전 1.9.15p5' 라 'version' 을 못 찾는다 → 영어로 받는다 p = subprocess.run(argv, capture_output=True, text=True, timeout=10, env=dict(os.environ, LC_ALL="C")) except Exception: continue out = (p.stdout or "") + (p.stderr or "") m = re.search(pattern, out) found[product] = m.group(1) if m else "unknown" return found def package_products(): """패키지 관리자 목록에서 이름만 추린다(버전은 probe 쪽이 정확).""" names = set() if shutil.which("dpkg"): try: p = subprocess.run(["dpkg-query", "-W", "-f=${Package}\n"], capture_output=True, text=True, timeout=30) names.update(l.strip().lower() for l in p.stdout.splitlines() if l.strip()) except Exception: pass elif shutil.which("rpm"): try: p = subprocess.run(["rpm", "-qa", "--qf", "%{NAME}\n"], capture_output=True, text=True, timeout=30) names.update(l.strip().lower() for l in p.stdout.splitlines() if l.strip()) except Exception: pass # 서버는 NVD product 명과 맞춰본다 — 흔한 표기 차이를 여기서 흡수 alias = {"php-cli": "php", "php8.3-cli": "php", "apache2": "apache_http_server", "mariadb-server": "mariadb", "mysql-server": "mysql", "nodejs": "node.js", "openssh-server": "openssh", "openssh-client": "openssh", "python3": "python"} return {alias.get(n, n) for n in names} def inventory(limit=250): """실제로 실행되는 제품을 앞에 둔다 — 서버가 개수를 자르더라도 중요한 게 먼저 가도록.""" ver = probe_versions() pkgs = package_products() - set(ver) prods = list(ver) + sorted(pkgs) return ver, prods[:limit] # ── 공식 보안자료 대조 (2026-10-17) ─────────────────────────────────── # AI 생성 점검 코드 실행은 2026-09-06 부터 멈춰 있다(서버 vuln_policy). 그 대신 설치 패키지 이름·버전만 보내면 서버가 # 배포판이 낸 '수정된 패키지 버전'(Ubuntu: Canonical OVAL·보안 공지, Rocky: RLSA)과 비교해 판정·저장한다. # 여기서는 고정된 조회 명령(dpkg-query 또는 rpm -qa, /etc/os-release, 커널 판)만 돈다. 생성 코드는 받지도 돌리지도 않는다. def os_release(): rel = {} try: for line in Path("/etc/os-release").read_text(encoding="utf-8").splitlines(): if "=" in line: k, v = line.split("=", 1) rel[k.strip()] = v.strip().strip('"') except OSError: pass return {"id": rel.get("ID", ""), "version_id": rel.get("VERSION_ID", ""), "codename": rel.get("VERSION_CODENAME", "")} def installed_packages(): """설치된 패키지 [{name, arch, version, source}]. dpkg 는 설치 완료(ii)만, rpm 은 EPOCH 가 없으면 0.""" env = dict(os.environ, LC_ALL="C") out = [] try: if shutil.which("dpkg-query"): p = subprocess.run(["dpkg-query", "-W", "-f=${db:Status-Abbrev}\t${Package}\t${Architecture}\t${Version}\t${source:Package}\n"], capture_output=True, text=True, timeout=60, env=env) for line in p.stdout.splitlines(): f = line.split("\t") if len(f) == 5 and f[0].startswith("ii") and f[1] and f[3]: out.append({"name": f[1], "arch": f[2], "version": f[3], "source": f[4]}) elif shutil.which("rpm"): p = subprocess.run(["rpm", "-qa", "--qf", "%{NAME}\t%{ARCH}\t%{EPOCHNUM}:%{VERSION}-%{RELEASE}\t%{SOURCERPM}\n"], capture_output=True, text=True, timeout=60, env=env) for line in p.stdout.splitlines(): f = line.split("\t") if len(f) == 4 and f[0] and f[2]: out.append({"name": f[0], "arch": f[1], "version": f[2], "source": f[3]}) except Exception: return [] return out def report_packages(cfg, quiet=False, dry_run=False): """설치 패키지 목록을 보내 공식 보안자료와 대조한다. 서버가 판정·저장(dry_run 이면 판정만). 성공하면 True.""" pk = installed_packages() if not pk: return False try: data, sig, raw = signed_post(cfg, "/api/agent/linux/packages", {"host_id": cfg["host_id"], "ts": int(time.time()), "os": os_release(), "kernel": platform.release(), "packages": pk, "dry_run": dry_run}) except urllib.error.HTTPError as e: if not quiet: print(f" 공식 보안자료 대조: 서버가 받지 않음(HTTP {e.code}) — 이 배포판·서버는 아직 대상이 아닙니다") return False if not verify_server_sig(cfg, raw, sig): sys.exit("서버 응답 서명이 맞지 않습니다 — 실행을 중단합니다") if data.get("error"): if not quiet: print(f" 공식 보안자료 대조: {data['error']}") return False if not quiet: un = sum((data.get("unfixed") or {}).values()) esm = f"(Ubuntu Pro ESM 업데이트 {data['esm_vulnerable']})" if data.get("esm_vulnerable") else "" print(f"[공식 보안자료 대조] {data.get('distro')} {data.get('rel')} · 패키지 {data.get('packages')}" f" · 커널 {data.get('kernel_flavor') or '자료 없음'} — 취약 {data.get('vulnerable')}{esm} · 수정 적용됨 {data.get('ok')}" f" · 배포판 수정 대기 {un} (자료 {data.get('data_at')}){' · 보고 안 함(--dry-run)' if dry_run else ''}") for t in (data.get("top") or [])[:10]: print(f" · {t['cve']:<18} {t['detected'][:44]} {t['reason'][:60]}") return True def signed_post(cfg, path, payload): body = json.dumps(payload, ensure_ascii=False).encode() sig = "sha256=" + hmac.new(cfg["secret"].encode(), body, hashlib.sha256).hexdigest() req = urllib.request.Request(cfg["server"].rstrip("/") + path, 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() return json.loads(raw), r.headers.get("X-Signature", ""), raw def verify_server_sig(cfg, raw, sig): """서버가 준 코드 묶음이 진짜인지 확인 — 중간에서 바꿔치기 못 하게. 파싱한 값이 아니라 '받은 바이트 그대로' 검증한다(재직렬화하면 바이트가 달라진다).""" return hmac.compare_digest(sig, "sha256=" + hmac.new(cfg["secret"].encode(), raw, hashlib.sha256).hexdigest()) def load_cursor(): """직전 실행에서 어디까지 받았는지. 한 바퀴 다 돌면 서버가 빈 값을 주고 처음으로 돌아간다.""" try: return STATE.read_text(encoding="utf-8").strip() except OSError: return "" def save_cursor(v): try: STATE.parent.mkdir(parents=True, exist_ok=True) STATE.write_text(v or "", encoding="utf-8") except OSError: pass 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() 정의 없음" # 점검 결과가 그대로 서버로 가서 DB 에 남는다. 실측으로 어떤 점검이 환경변수를 훑어 # SOSIG_SECRET·OLLAMA_API_KEY 같은 이름을 검출값에 담아 보냈다. 값까지 담는 코드가 나오면 # 그대로 유출이므로, 보내기 전에 가린다. 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 격리 환경이 필요합니다."} # ── KISA 2026 주요정보통신기반시설 점검 ──────────────────────────────── # CVE 점검과 다른 축이다. CVE 는 "이 소프트웨어에 알려진 취약점이 있는가", # KISA 는 "이 서버 설정이 기준에 맞는가" 를 본다. # # CVE 점검과 달리 root 가 필요하다 — 스크립트가 /etc/shadow 같은 것을 읽는다. # root 가 아니면 스크립트 스스로 "root 권한이 필요합니다" 하고 멈추므로, # 그 상태로 돌리면 전 항목이 점검불가로 나온다. 그래서 미리 확인하고 알린다. KISA_TIMEOUT_EACH = 60 KISA_TIMEOUT_TOTAL = 1800 _VMAP = {"GOOD": "양호", "VULNERABLE": "취약", "MANUAL": "수동확인", "N/A": "점검불가"} def kisa_platform(): """KISA 가이드의 플랫폼 구분. 조치 코드(kisa_fix)와 같은 기준을 쓴다.""" rel = {} try: for line in Path("/etc/os-release").read_text(encoding="utf-8").splitlines(): if "=" in line: k, v = line.split("=", 1) rel[k.strip()] = v.strip().strip('"') except OSError: pass ident = " ".join([rel.get("ID", ""), rel.get("ID_LIKE", "")]).lower() if any(x in ident for x in ("debian", "ubuntu")): return "Debian" if any(x in ident for x in ("rhel", "fedora", "centos", "rocky", "almalinux", "ol")): return "RedHat" return {"SunOS": "Solaris", "AIX": "AIX", "HP-UX": "HP-UX"}.get(os.uname().sysname, "") def run_kisa(cfg, quiet=False, dry_run=False): import base64 import tarfile platform = kisa_platform() if not platform: print("이 플랫폼은 KISA 점검 대상이 아닙니다.", file=sys.stderr) return 1 as_root = os.geteuid() == 0 if not quiet: print(f"[1/4] 플랫폼 {platform}" + (" (root)" if as_root else " (일반 계정)")) if not as_root: print(" ⚠️ KISA 점검 스크립트는 root 를 요구합니다. 이대로 돌리면 대부분 점검불가로 나옵니다.\n" " sudo python3 sosig_agent.py --kisa 로 실행하십시오.", file=sys.stderr) data, sig, raw = signed_post(cfg, "/api/agent/kisa/bundle", {"host_id": cfg["host_id"], "ts": int(time.time()), "platform": platform}) if not verify_server_sig(cfg, raw, sig): sys.exit("서버 응답 서명이 맞지 않습니다 — 실행을 중단합니다") blob = base64.b64decode(data["tgz_b64"]) if hashlib.sha256(blob).hexdigest() != data.get("sha256"): sys.exit("점검 스크립트 묶음의 해시가 맞지 않습니다 — 실행을 중단합니다") if not quiet: print(f"[2/4] 점검 스크립트 수신 {len(blob)//1024}KB (서명·해시 확인됨)") work = tempfile.mkdtemp(prefix="sosig-kisa-") try: with tarfile.open(fileobj=io.BytesIO(blob), mode="r:gz") as tf: # 압축 해제 경로가 work 밖으로 나가지 않는지 확인한다. for m in tf.getmembers(): dest = os.path.realpath(os.path.join(work, m.name)) if not dest.startswith(os.path.realpath(work) + os.sep): sys.exit(f"묶음에 비정상 경로가 있습니다: {m.name}") # 파이썬 3.12+ 는 filter 를 요구한다(3.14 는 미지정 시 오류). 3.10 등 구버전에는 없다. try: tf.extractall(work, filter="data") except TypeError: tf.extractall(work) base = os.path.join(work, "01.Unix서버", platform) for d in (base, os.path.join(work, "lib")): for f in os.listdir(d): if f.endswith(".sh"): os.chmod(os.path.join(d, f), 0o755) scripts = sorted(f for f in os.listdir(base) if re.fullmatch(r"U\d{2}_check\.sh", f)) results, t0 = [], time.time() for i, s in enumerate(scripts, 1): if time.time() - t0 > KISA_TIMEOUT_TOTAL: print(f" 전체 제한시간 초과 — {i-1}/{len(scripts)} 까지만 수행", file=sys.stderr) break item = "U-" + s[1:3] try: subprocess.run(["bash", "./" + s], cwd=base, capture_output=True, text=True, timeout=KISA_TIMEOUT_EACH) except subprocess.TimeoutExpired: results.append({"item_id": item, "verdict": "점검불가", "detected": "제한시간 초과", "raw_output": ""}) continue except Exception as e: results.append({"item_id": item, "verdict": "점검불가", "detected": str(e)[:200], "raw_output": ""}) continue results.append(_kisa_read_result(base, item)) if not quiet and i % 10 == 0: print(f" {i}/{len(scripts)} …") tally = {} for r in results: tally[r["verdict"]] = tally.get(r["verdict"], 0) + 1 if not quiet: print(f"[3/4] 실행 {len(results)}건 — " + " · ".join(f"{k} {v}" for k, v in tally.items())) vul = [r for r in results if r["verdict"] == "취약"] if vul: print(f" ⚠️ 취약 {len(vul)}건 — 사람이 확인해야 할 후보입니다") for r in vul[:10]: print(f" · {r['item_id']} {str(r.get('detected'))[:70]}") if dry_run: print("[4/4] --dry-run: 보고 생략") return 0 resp, _, _ = signed_post(cfg, "/api/agent/kisa/report", {"host_id": cfg["host_id"], "ts": int(time.time()), "platform": platform, "hostname": socket.gethostname(), "as_root": as_root, "results": results}) if not quiet: print(f"[4/4] 보고 완료 — 저장 {resp.get('stored')}건 {resp.get('summary')}") return 0 finally: shutil.rmtree(work, ignore_errors=True) def _kisa_read_result(base, item): """스크립트가 남긴 JSON 결과를 읽는다. 로그 파싱보다 형식이 안정적이다.""" rdir = os.path.join(base, "results") newest, newest_t = None, -1 for root, _dirs, files in os.walk(rdir): for f in files: if item in f and f.endswith(".json"): p = os.path.join(root, f) t = os.path.getmtime(p) if t > newest_t: newest, newest_t = p, t if not newest: return {"item_id": item, "verdict": "점검불가", "detected": "결과 파일이 생성되지 않았습니다", "raw_output": ""} try: d = json.loads(Path(newest).read_text(encoding="utf-8", errors="ignore")) except Exception as e: return {"item_id": item, "verdict": "점검불가", "detected": f"결과 파일 파싱 실패: {e}"[:200], "raw_output": ""} insp = d.get("inspection") or {} return {"item_id": item, "verdict": _VMAP.get(str(d.get("final_result", "")).upper(), "점검불가"), "detected": str(insp.get("summary") or d.get("command_result") or "")[:2000], "raw_output": json.dumps(d, ensure_ascii=False)[:60000]} def main(): ap = argparse.ArgumentParser() ap.add_argument("--dry-run", action="store_true", help="보고하지 않음") ap.add_argument("--inventory", action="store_true", help="설치 조사 결과만 출력") ap.add_argument("--quiet", action="store_true") ap.add_argument("--restart", action="store_true", help="이어받기 위치를 지우고 처음부터") ap.add_argument("--kisa", action="store_true", help="KISA 2026 주요정보통신기반시설 점검(설정 기준). root 필요") a = ap.parse_args() # KISA 점검은 root 가 있어야 한다(설정 파일·계정 정보를 읽는다). CVE 점검은 아니다. if os.geteuid() == 0 and not a.kisa: print("경고: root 로 실행 중입니다. CVE 점검은 일반 계정 권한으로 충분합니다.", file=sys.stderr) # sudo 는 HOME 을 /root 로 바꾼다. 설정은 원래 사용자 홈에 있으므로 되돌린다. if os.environ.get("SUDO_USER"): try: import pwd os.environ["HOME"] = pwd.getpwnam(os.environ["SUDO_USER"]).pw_dir except Exception: pass cfg = load_config() if a.kisa: return run_kisa(cfg, quiet=a.quiet, dry_run=a.dry_run) ver, prods = inventory() if a.inventory: print("설치 확인(버전):") for k, v in sorted(ver.items()): print(f" {k:<22} {v}") print(f"\n서버에 보낼 제품명 {len(prods)}개 (앞 20개): {', '.join(prods[:20])}") return 0 if not a.quiet: print(f"[1/4] 설치 조사 — 실행 확인 {len(ver)}종: " + ", ".join(f"{k} {v}" for k, v in sorted(ver.items())[:8])) distro_ok = report_packages(cfg, quiet=a.quiet, dry_run=a.dry_run) cursor = "" if a.restart else load_cursor() data, sig, raw = signed_post(cfg, "/api/agent/bundle", {"host_id": cfg["host_id"], "ts": int(time.time()), "products": prods, "cursor": cursor}) if not verify_server_sig(cfg, raw, sig): sys.exit("서버 응답 서명이 맞지 않습니다 — 실행을 중단합니다") if data.get('execution_paused'): if distro_ok: if not a.quiet: print('AI 생성 점검 코드는 보류 중 — CVE 점검은 위 공식 보안자료 대조로 대신했습니다.') return 0 print('점검 보류: 검증된 OS 격리 환경 준비가 필요합니다. 이전 결과를 유지합니다.') return 75 checks = data.get("checks", []) if not a.quiet: more = " · 남은 항목 있음" if data.get("more") else " · 마지막 묶음" print(f"[2/4] 점검 코드 {len(checks)}건 수신 (서명 확인됨{more})") inv_norm = {norm_product(x) for x in prods} | {norm_product(x) for x in ver} runnable = [c for c in checks if in_inventory(c.get("product", ""), inv_norm)] not_target = len(checks) - len(runnable) if not a.quiet and not_target: absent = sorted({c.get("product", "") for c in checks if not in_inventory(c.get("product", ""), inv_norm)})[:6] print(f" 대상 아님 {not_target}건 제외 (이 서버에 없는 제품: {', '.join(absent)}…)") results, skipped, t0 = [], 0, time.time() for c in runnable: if time.time() - t0 > TIMEOUT_TOTAL: break if hashlib.sha256((c.get("code") or "").encode()).hexdigest() != c.get("sha256"): skipped += 1 continue ok, why = safe_to_run(c["code"]) if not ok: skipped += 1 continue r = run_check(c["code"]) r["cve_id"] = c["cve_id"] r['code_sha256'] = c['sha256'] results.append(r) tally = {} for r in results: tally[r["verdict"]] = tally.get(r["verdict"], 0) + 1 if not a.quiet: print(f"[3/4] 실행 {len(results)}건 (자체검사 탈락 {skipped} · 대상 아님 {not_target}) — " + " · ".join(f"{k} {v}" for k, v in tally.items())) vul = [x for x in results if x["verdict"] == "취약"] if vul: # '취약'이라고 단정하지 않는다. 자동 생성 점검의 취약 판정은 실측에서 대부분 # 오탐이었다 — 사람이 확인하기 전까지는 후보다. print(f" ⚠️ 취약 의심 {len(vul)}건 — 확인이 필요한 후보이며 확정된 취약점이 아닙니다") for r in vul[:10]: print(f" · {r['cve_id']:<18} {str(r.get('detected'))[:36]} {str(r.get('reason'))[:46]}") if a.dry_run: print("[4/4] --dry-run: 보고 생략") return 0 resp, _, _ = signed_post(cfg, "/api/agent/report", {"host_id": cfg["host_id"], "ts": int(time.time()), "results": results}) # 보고까지 끝난 뒤에만 커서를 옮긴다 — 중간에 실패하면 같은 구간을 다시 한다 save_cursor(data.get("cursor", "") if data.get("more") else "") if not a.quiet: print(f"[4/4] 보고 완료 — 저장 {resp.get('stored')}건 {resp.get('summary')}") return 0 if __name__ == "__main__": sys.exit(main())