#!/usr/bin/env python3 # ============================================================================= # odysec-check —— Odysec 主機入侵跡證健檢採集程式 # # 版本 : 0.2.0 # 建置日期 : 2026-07-27 # 授權 : 僅供 Odysec 客戶於自有主機執行 # # 這是一份完整、可讀的 Python 原始碼，沒有經過壓縮或混淆。 # 它需要以 root 執行才能取得完整涵蓋率，因此我們預期你會先讀過它。 # 請讀。任何要求你用 root 執行卻不讓你看原始碼的資安工具，都不值得信任。 # # 相依套件：無。只用 Python 標準函式庫（需 Python 3.9 以上）。 # 網路行為：無。本程式不會建立任何對外連線。 # 寫入行為：只寫出你指定的結果檔，不碰系統上其他任何檔案。 # ============================================================================= #!/usr/bin/env python3 """vigil — 主機入侵跡證評估代理程式 (host compromise assessment agent). 設計原則: * 純 stdlib，單一檔案，scp 過去就能跑，不需安裝任何東西。 * 只讀不寫（除了自己的 state 目錄），不開任何監聽埠，不用 shell=True。 * 偵測核心是「已知良好基準線的漂移」，不是病毒碼比對。 * 權限不足時明講跳過了什麼，不靜默降級。 用法: vigil.py baseline 建立已知良好基準線 vigil.py scan 採集 + 偵測 + 與基準線比對 vigil.py scan --json 機器可讀輸出 vigil.py show 列出目前基準線摘要 """ from __future__ import annotations import argparse import fnmatch import hashlib import json import os import platform import pwd import re import socket import stat import subprocess import sys import time import urllib.error import urllib.request from pathlib import Path VERSION = "0.2.0" # -------------------------------------------------------------------------- # 常數與設定 # -------------------------------------------------------------------------- SEV_ORDER = ["INFO", "LOW", "MEDIUM", "HIGH", "CRITICAL"] SEV_WEIGHT = {"INFO": 0, "LOW": 2, "MEDIUM": 8, "HIGH": 20, "CRITICAL": 45} DEFAULT_STATE = Path(os.environ.get("VIGIL_STATE", "/var/lib/vigil")) CONFIG_CANDIDATES = [ Path("/etc/vigil/config.json"), Path.home() / ".config" / "vigil" / "config.json", ] MAX_READ = 512 * 1024 # 單一檔案雜湊上限，避免誤踩巨大檔案 # 全檔案系統掃描時要剪掉的路徑（虛擬檔案系統、快取、容器層） PRUNE_DIRS = { "/proc", "/sys", "/dev", "/run", "/mnt", "/media", "/lost+found", "/var/cache/pacman/pkg", "/var/lib/libvirt/images", "/swapfile", # 容器執行環境的映像層。基礎映像本來就含 su/passwd/mount 等 SUID 程式， # 掃進去會產生大量「非套件管理的 SUID」假陽性——任何跑容器的主機都會中。 # 那些檔案不在宿主的執行路徑上，威脅要透過容器逃逸才成立，不屬本工具範圍。 "/var/lib/docker", "/var/lib/containers", "/var/lib/containerd", "/var/lib/lxc", "/var/lib/lxd", "/var/lib/machines", "/var/lib/rancher", "/snap", "/var/snap", "/var/lib/snapd", } PRUNE_NAMES = {".git", "node_modules", "__pycache__", ".cache", ".npm", ".venv"} # 執行檔出現在這些位置本身就可疑 SUSPICIOUS_EXEC_DIRS = ("/tmp/", "/var/tmp/", "/dev/shm/", "/run/shm/", "/run/user/") # 各「面向」(surface) 的漂移嚴重度。基準線比對的偵測邏輯全部由這張表驅動。 SURFACES: dict[str, dict] = { "systemd.unit": {"label": "systemd unit", "added": "HIGH", "changed": "HIGH", "removed": "LOW"}, "systemd.enabled": {"label": "systemd 啟用連結", "added": "HIGH", "changed": "MEDIUM", "removed": "LOW"}, "cron": {"label": "cron 排程", "added": "HIGH", "changed": "HIGH", "removed": "LOW"}, "autostart": {"label": "桌面自動啟動", "added": "HIGH", "changed": "MEDIUM", "removed": "INFO"}, "shellrc": {"label": "shell 啟動檔", "added": "HIGH", "changed": "MEDIUM", "removed": "LOW"}, "preload": {"label": "動態連結器預載", "added": "CRITICAL", "changed": "CRITICAL", "removed": "MEDIUM"}, "pam": {"label": "PAM 設定", "added": "HIGH", "changed": "HIGH", "removed": "HIGH"}, "sudoers": {"label": "sudoers", "added": "CRITICAL", "changed": "CRITICAL", "removed": "MEDIUM"}, "ssh.authkeys": {"label": "SSH authorized_keys", "added": "CRITICAL", "changed": "CRITICAL", "removed": "MEDIUM"}, "sshd.config": {"label": "sshd 設定", "added": "HIGH", "changed": "HIGH", "removed": "HIGH"}, "udev": {"label": "udev 規則", "added": "HIGH", "changed": "MEDIUM", "removed": "LOW"}, "dbus": {"label": "D-Bus 系統服務", "added": "HIGH", "changed": "MEDIUM", "removed": "LOW"}, "polkit": {"label": "polkit 規則", "added": "HIGH", "changed": "HIGH", "removed": "LOW"}, "pkghook": {"label": "套件管理器 hook", "added": "HIGH", "changed": "HIGH", "removed": "LOW"}, "modconf": {"label": "核心模組設定", "added": "HIGH", "changed": "MEDIUM", "removed": "LOW"}, "suid": {"label": "SUID/SGID 執行檔", "added": "CRITICAL", "changed": "HIGH", "removed": "LOW"}, "listener": {"label": "監聽埠", "added": "HIGH", "changed": "MEDIUM", "removed": "INFO"}, "kmod": {"label": "已載入核心模組", "added": "MEDIUM", "changed": "INFO", "removed": "INFO"}, "kernel.taint": {"label": "核心 taint 來源模組", "added": "HIGH", "changed": "HIGH", "removed": "LOW"}, "account": {"label": "本機帳號", "added": "HIGH", "changed": "HIGH", "removed": "MEDIUM"}, "unpackaged": {"label": "非套件管理的執行檔", "added": "MEDIUM", "changed": "MEDIUM", "removed": "INFO"}, "aur.pkg": {"label": "AUR 套件", "added": "LOW", "changed": "LOW", "removed": "INFO"}, "aur.source": {"label": "AUR 建置來源", "added": "HIGH", "changed": "HIGH", "removed": "LOW"}, "docker.container": {"label": "Docker 容器", "added": "MEDIUM", "changed": "MEDIUM", "removed": "LOW"}, } # 預設 sysctl 政策：兩台都該成立的加固值。可由 config 的 sysctl_policy 覆寫/擴充。 DEFAULT_SYSCTL_POLICY = { "kernel.kptr_restrict": "1", "kernel.randomize_va_space": "2", "kernel.dmesg_restrict": "1", "kernel.yama.ptrace_scope": "1", "net.ipv4.conf.all.accept_redirects": "0", "net.ipv4.conf.all.rp_filter": "1|2", "net.ipv6.conf.all.accept_redirects": "0", "net.core.bpf_jit_harden": "2", "net.ipv4.conf.all.send_redirects": "0", "net.ipv4.conf.default.send_redirects": "0", } # -------------------------------------------------------------------------- # 小工具 # -------------------------------------------------------------------------- def now_ts() -> str: return time.strftime("%Y-%m-%dT%H:%M:%S%z") def sev_max(a: str, b: str) -> str: return a if SEV_ORDER.index(a) >= SEV_ORDER.index(b) else b def sha256_file(path: Path, limit: int = MAX_READ) -> str | None: try: h = hashlib.sha256() read = 0 with open(path, "rb") as fh: while read < limit: chunk = fh.read(min(65536, limit - read)) if not chunk: break h.update(chunk) read += len(chunk) return h.hexdigest()[:32] except OSError: return None def stat_fingerprint(path: Path, with_hash: bool = True) -> str | None: """檔案指紋：內容雜湊 + 權限 + 擁有者 + 大小。任一項變動即視為 changed。""" try: st = os.lstat(path) except OSError: return None if stat.S_ISLNK(st.st_mode): try: return f"link->{os.readlink(path)}" except OSError: return "link->?" if not stat.S_ISREG(st.st_mode): return f"special:{stat.S_IFMT(st.st_mode):o}" parts = [f"m{oct(st.st_mode & 0o7777)}", f"u{st.st_uid}:{st.st_gid}", f"s{st.st_size}"] if with_hash: digest = sha256_file(path) if digest: parts.insert(0, digest) return "|".join(parts) def safe_isdir(path: Path) -> bool: """Path.is_dir() 在 Python 3.13 遇到權限不足會拋 PermissionError 而非回 False。 掃描器一定會踩到別人的家目錄，這裡一律吞掉錯誤當作「看不到」。""" try: return path.is_dir() except OSError: return False def safe_isfile(path: Path) -> bool: try: return path.is_file() except OSError: return False def safe_exists(path: Path) -> bool: try: return path.exists() except OSError: return False def read_text(path: Path, limit: int = MAX_READ) -> str: try: with open(path, "r", encoding="utf-8", errors="replace") as fh: return fh.read(limit) except OSError: return "" def iter_files(root: Path, pattern: str = "*", depth: int = 3): """有界深度的檔案列舉，讀不到就跳過。""" if not safe_isdir(root): return base_depth = len(root.parts) try: for dirpath, dirnames, filenames in os.walk(root, followlinks=False): if len(Path(dirpath).parts) - base_depth >= depth: dirnames[:] = [] dirnames[:] = [d for d in dirnames if d not in PRUNE_NAMES] for name in filenames: if fnmatch.fnmatch(name, pattern): yield Path(dirpath) / name except OSError: return def _dedupe_dirs(dirs: list[Path]) -> list[Path]: """依真實路徑去重。/lib→/usr/lib、/usr/sbin→/usr/bin 這類相容連結會讓同一個 目錄被走兩次，產生成對的重複跡證。""" out, seen = [], set() for d in dirs: try: real = os.path.realpath(d) except OSError: continue if real in seen or not os.path.isdir(real): continue seen.add(real) out.append(Path(real)) return out def run(argv: list[str], timeout: int = 120) -> tuple[int, str]: """執行外部指令。固定 argv、絕不使用 shell。""" try: proc = subprocess.run( argv, capture_output=True, text=True, timeout=timeout, check=False ) return proc.returncode, (proc.stdout or "") + (proc.stderr or "") except (OSError, subprocess.SubprocessError) as exc: return -1, str(exc) def have(cmd: str) -> bool: from shutil import which return which(cmd) is not None # -------------------------------------------------------------------------- # 掃描結果容器 # -------------------------------------------------------------------------- class Scan: def __init__(self, cfg: dict): self.cfg = cfg self.started = time.time() self.surfaces: dict[str, dict[str, str]] = {} self.findings: list[dict] = [] self.skipped: list[str] = [] self.facts: dict = {} def surface(self, name: str) -> dict[str, str]: return self.surfaces.setdefault(name, {}) def add(self, module: str, severity: str, title: str, subject: str = "", detail: str = "", evidence: list[str] | None = None, standing: tuple[str, str] | None = None) -> None: """記錄一筆跡證。 standing=(面向, 項目鍵) 用於標記「常態觀察」——描述系統長期組態而非事件的 跡證，例如 dockge 掛載 docker.sock。這類跡證若該項目已收錄在基準線中就會被 壓下：基準線本身就代表「這個狀態我看過且接受」。項目一旦變動，漂移比對會 獨立產生告警，所以壓制不會造成漏報。 """ key = f"{module}:{title}:{subject}"[:220] self.findings.append({ "key": key, "module": module, "severity": severity, "title": title, "subject": subject, "detail": detail, "evidence": evidence or [], "standing": list(standing) if standing else None, }) def skip(self, what: str, why: str) -> None: self.skipped.append(f"{what} — {why}") # -------------------------------------------------------------------------- # 主機環境探測 # -------------------------------------------------------------------------- class Host: def __init__(self): self.hostname = socket.gethostname() self.is_root = os.geteuid() == 0 self.kernel = platform.release() self.distro = self._distro() self.pkg = "pacman" if self.distro == "arch" else ("dpkg" if self.distro == "debian" else None) self.nosuid_mounts = self._mounts_with("nosuid") self.noexec_mounts = self._mounts_with("noexec") try: self.root_dev = os.stat("/").st_dev except OSError: self.root_dev = None @staticmethod def _distro() -> str: data = read_text(Path("/etc/os-release")) ids = "" for line in data.splitlines(): if line.startswith("ID="): ids = line.split("=", 1)[1].strip().strip('"') elif line.startswith("ID_LIKE="): ids += " " + line.split("=", 1)[1].strip().strip('"') if "arch" in ids: return "arch" if "debian" in ids or "ubuntu" in ids: return "debian" return ids.split()[0] if ids else "unknown" @staticmethod def _mounts_with(option: str) -> set[str]: out = set() for line in read_text(Path("/proc/mounts")).splitlines(): parts = line.split() if len(parts) >= 4 and option in parts[3].split(","): out.add(parts[1]) return out def under_mount(self, path: str, mounts: set[str]) -> bool: return any(path == m or path.startswith(m.rstrip("/") + "/") for m in mounts) # -------------------------------------------------------------------------- # 套件資料庫：哪些檔案「應該」存在且由套件管理 # -------------------------------------------------------------------------- class PackageDB: """一次性建立「套件擁有的檔案」集合，之後 O(1) 查詢，不逐檔呼叫 pacman -Qo。""" def __init__(self, host: Host, scan: Scan): self.host = host self.owned: set[str] = set() self.available = False if host.pkg == "pacman": rc, out = run(["pacman", "-Qlq"], timeout=180) if rc == 0: self.owned = {ln.rstrip("/") for ln in out.splitlines() if ln.startswith("/")} self.available = True elif host.pkg == "dpkg": listdir = Path("/var/lib/dpkg/info") if safe_isdir(listdir): for f in listdir.glob("*.list"): try: with open(f, "r", encoding="utf-8", errors="replace") as fh: for ln in fh: ln = ln.strip() if ln.startswith("/"): self.owned.add(ln.rstrip("/")) except OSError: continue self.available = bool(self.owned) if not self.available: scan.skip("套件擁有權查詢", f"無法建立 {host.pkg} 檔案索引") def owns(self, path: str) -> bool: p = path.rstrip("/") if p in self.owned: return True # Arch 的 /usr/sbin→/usr/bin、/lib→/usr/lib 之類的相容連結，會讓同一個檔案 # 有多條路徑，而套件資料庫只記錄其中一條。解析到真實路徑再查一次。 try: real = os.path.realpath(p) except OSError: return False return real != p and real in self.owned # -------------------------------------------------------------------------- # 各偵測模組 # -------------------------------------------------------------------------- def parse_pacman_qkk(out: str) -> list[tuple[str, str, str]]: """解析 `pacman -Qkk` 輸出，取出真正有差異的檔案。 有差異的行長這樣: coreutils: /usr/bin/ls (SHA256 checksum mismatch) filesystem: /etc/hosts (Modification time mismatch) 收尾統計行 "444 total files, 0 altered files" 要略過。 """ altered = [] reasons = ("UID", "GID", "Permissions", "Size", "Modification time", "Symlink", "SHA256", "MD5", "Missing", "type") for line in out.splitlines(): if "altered files" in line or not line.strip(): continue m = re.match(r"^([\w@.+-]+):\s+(/\S+)\s+\((.+)\)\s*$", line) if m: pkg, path, reason = m.groups() if any(r in reason for r in reasons): altered.append((pkg, path, reason)) return altered def parse_dpkg_verify(out: str) -> list[tuple[str, str, str]]: """解析 `dpkg -V` 輸出。格式為 9 個屬性旗標，第 3 位是 md5 檢查結果。 ??5?????? /etc/default/grub ??5?????? c /etc/ssh/sshd_config """ altered = [] for line in out.splitlines(): m = re.match(r"^([?\w]{9})\s+(?:(c)\s+)?(/\S+)\s*$", line.strip()) if not m: continue flags, conf, path = m.groups() if flags[2] == "5": altered.append(("dpkg", path, "conffile 內容變更" if conf else "內容雜湊不符")) return altered def collect_integrity(host: Host, scan: Scan) -> None: """套件管理器檔案完整性 —— 在發行版管理的機器上，這是最強的單一訊號。""" if host.pkg == "pacman": _, out = run(["pacman", "-Qkk"], timeout=900) _report_altered(scan, parse_pacman_qkk(out)) scan.facts["integrity_checked"] = True elif host.pkg == "dpkg": _, out = run(["dpkg", "-V"], timeout=900) _report_altered(scan, parse_dpkg_verify(out)) scan.facts["integrity_checked"] = True else: scan.skip("套件完整性驗證", "無法辨識的套件管理器") def _report_altered(scan: Scan, altered: list[tuple[str, str, str]]) -> None: if not altered: return # 系統執行檔/函式庫被改動，遠比設定檔被改動嚴重 hot = [a for a in altered if re.match(r"^/(usr/)?(s?bin|lib\d*|lib/systemd)/", a[1])] for pkg, path, reason in hot[:40]: scan.add("integrity", "CRITICAL", "套件管理的系統二進位檔已被修改", path, f"套件 {pkg} 宣稱擁有此檔，但內容與套件資料庫不符：{reason}", [f"{pkg}: {path} ({reason})"]) cold = [a for a in altered if a not in hot] if cold: scan.add("integrity", "MEDIUM", "套件管理的檔案與資料庫不符", f"{len(cold)} 個檔案", "多為設定檔手動編輯所致，但仍應逐一確認是否為自己所改。", [f"{p[0]}: {p[1]} ({p[2]})" for p in cold[:25]]) def collect_persistence(host: Host, scan: Scan, pkgdb: PackageDB) -> None: """列舉所有能讓程式在重開機後自動執行的位置。後門必須落在其中之一。""" # --- systemd unit 檔與 drop-in --- unit_dirs = _dedupe_dirs([Path("/etc/systemd/system"), Path("/usr/lib/systemd/system"), Path("/run/systemd/system"), Path("/lib/systemd/system")]) s_unit = scan.surface("systemd.unit") s_enabled = scan.surface("systemd.enabled") seen_units = set() for d in unit_dirs: if not safe_isdir(d): continue for f in iter_files(d, "*", depth=2): name = str(f) if f.suffix not in (".service", ".timer", ".socket", ".path", ".mount", ".conf"): continue if name in seen_units: continue seen_units.add(name) if os.path.islink(f): s_enabled[name] = stat_fingerprint(f) or "?" continue fp = stat_fingerprint(f) if fp: s_unit[name] = fp _inspect_unit(f, scan, pkgdb) # *.wants / *.requires 內的啟用連結 for sub in d.glob("*.wants/*"): s_enabled[str(sub)] = stat_fingerprint(sub) or "?" for sub in d.glob("*.requires/*"): s_enabled[str(sub)] = stat_fingerprint(sub) or "?" # 使用者層 systemd for home in _home_dirs(): ud = home / ".config" / "systemd" / "user" for f in iter_files(ud, "*", depth=2): fp = stat_fingerprint(f) if fp: s_unit[str(f)] = fp _inspect_unit(f, scan, pkgdb) # --- cron 各種形式 --- s_cron = scan.surface("cron") cron_targets = [Path("/etc/crontab"), Path("/etc/anacrontab")] cron_dirs = [Path("/etc/cron.d"), Path("/etc/cron.hourly"), Path("/etc/cron.daily"), Path("/etc/cron.weekly"), Path("/etc/cron.monthly"), Path("/var/spool/cron"), Path("/var/spool/cron/crontabs")] for p in cron_targets: fp = stat_fingerprint(p) if fp: s_cron[str(p)] = fp unreadable_cron = 0 for d in cron_dirs: if not safe_isdir(d): continue if not os.access(d, os.R_OK): unreadable_cron += 1 continue for f in iter_files(d, "*", depth=1): fp = stat_fingerprint(f) if fp: s_cron[str(f)] = fp if unreadable_cron: scan.skip("cron 排程列舉", f"{unreadable_cron} 個 cron 目錄無讀取權限（需 root）") # --- 桌面自動啟動 --- s_auto = scan.surface("autostart") auto_dirs = [Path("/etc/xdg/autostart")] + [h / ".config" / "autostart" for h in _home_dirs()] for d in auto_dirs: for f in iter_files(d, "*.desktop", depth=1): fp = stat_fingerprint(f) if fp: s_auto[str(f)] = fp # --- shell 啟動檔 --- s_rc = scan.surface("shellrc") rc_targets = [Path("/etc/profile"), Path("/etc/bash.bashrc"), Path("/etc/bashrc"), Path("/etc/zsh/zshrc"), Path("/etc/zshenv"), Path("/etc/environment")] for d in (Path("/etc/profile.d"),): rc_targets.extend(iter_files(d, "*", depth=1)) for home in _home_dirs(): for nm in (".bashrc", ".bash_profile", ".profile", ".zshrc", ".zshenv", ".bash_login", ".bash_logout", ".config/fish/config.fish"): rc_targets.append(home / nm) for p in rc_targets: fp = stat_fingerprint(p) if fp: s_rc[str(p)] = fp # --- 動態連結器預載（經典 rootkit 手法）--- s_pre = scan.surface("preload") preload = Path("/etc/ld.so.preload") if safe_exists(preload): content = read_text(preload).strip() s_pre[str(preload)] = stat_fingerprint(preload) or "?" if content: scan.add("persistence", "CRITICAL", "/etc/ld.so.preload 非空", str(preload), "此檔會把指定的 .so 注入每一個動態連結程式，是使用者層 rootkit 的典型落點。" "乾淨的 Arch/Debian 系統上此檔通常根本不存在。", content.splitlines()[:10]) for f in iter_files(Path("/etc/ld.so.conf.d"), "*.conf", depth=1): s_pre[str(f)] = stat_fingerprint(f) or "?" env_content = read_text(Path("/etc/environment")) if "LD_PRELOAD" in env_content or "LD_LIBRARY_PATH" in env_content: scan.add("persistence", "HIGH", "/etc/environment 含連結器注入變數", "/etc/environment", "LD_PRELOAD / LD_LIBRARY_PATH 出現在全域環境檔中。", [l for l in env_content.splitlines() if "LD_" in l][:5]) # --- 認證與授權相關 --- for surface_name, paths in ( ("pam", list(iter_files(Path("/etc/pam.d"), "*", depth=1))), ("sudoers", [Path("/etc/sudoers")] + list(iter_files(Path("/etc/sudoers.d"), "*", depth=1))), ("udev", list(iter_files(Path("/etc/udev/rules.d"), "*", depth=1))), ("dbus", list(iter_files(Path("/usr/share/dbus-1/system-services"), "*", depth=1)) + list(iter_files(Path("/etc/dbus-1/system.d"), "*", depth=1))), ("polkit", list(iter_files(Path("/etc/polkit-1/rules.d"), "*", depth=1)) + list(iter_files(Path("/etc/polkit-1/localauthority"), "*", depth=3))), ("pkghook", list(iter_files(Path("/etc/pacman.d/hooks"), "*", depth=1)) + list(iter_files(Path("/usr/share/libalpm/hooks"), "*", depth=1)) + list(iter_files(Path("/etc/apt/apt.conf.d"), "*", depth=1))), ("modconf", list(iter_files(Path("/etc/modules-load.d"), "*", depth=1)) + list(iter_files(Path("/etc/modprobe.d"), "*", depth=1)) + [Path("/etc/modules")]), ("sshd.config", [Path("/etc/ssh/sshd_config")] + list(iter_files(Path("/etc/ssh/sshd_config.d"), "*", depth=1))), ): surf = scan.surface(surface_name) for p in paths: fp = stat_fingerprint(p) if fp: surf[str(p)] = fp # sudoers 內容啟發式 for sf in [Path("/etc/sudoers")] + list(iter_files(Path("/etc/sudoers.d"), "*", depth=1)): for line in read_text(sf).splitlines(): line = line.strip() if line.startswith("#") or not line: continue if "NOPASSWD" in line and "ALL" in line: scan.add("persistence", "HIGH", "sudoers 含 NOPASSWD:ALL 規則", line[:80], f"定義於 {sf}。任何取得該帳號的程式都能免密碼提權至 root。", [line], standing=("sudoers", str(sf))) # --- SSH authorized_keys --- s_keys = scan.surface("ssh.authkeys") unreadable_keys = 0 for home in _home_dirs(include_root=True): for nm in ("authorized_keys", "authorized_keys2"): p = home / ".ssh" / nm if not safe_exists(p): continue if not os.access(p, os.R_OK): unreadable_keys += 1 continue for idx, line in enumerate(read_text(p).splitlines()): line = line.strip() if not line or line.startswith("#"): continue fields = line.split() fp = hashlib.sha256(line.encode()).hexdigest()[:16] comment = fields[-1] if len(fields) > 2 else "(無註解)" s_keys[f"{p}#{idx}"] = fp if line.startswith(("command=", "no-pty", "restrict")): continue if "root" in str(home) or str(home) == "/root": scan.add("persistence", "HIGH", "root 帳號存在 SSH 公鑰", comment, f"{p} 內有可直接登入 root 的授權金鑰。", [line[:100]], standing=("ssh.authkeys", f"{p}#{idx}")) if unreadable_keys: scan.skip("authorized_keys 列舉", f"{unreadable_keys} 個檔案無讀取權限（需 root）") def _inspect_unit(path: Path, scan: Scan, pkgdb: PackageDB) -> None: """檢查 systemd unit 的執行內容是否指向可疑位置。""" text = read_text(path, 64 * 1024) if not text: return # 由套件本身提供的 unit，其 ExecStart 指向什麼是套件維護者的決定，不是異常訊號。 # 真正有意義的是「unit 檔本身不受套件管理」，或執行目標落在暫存目錄。 unit_is_packaged = pkgdb.available and pkgdb.owns(str(path)) for m in re.finditer(r"^\s*(ExecStart\w*|ExecStop\w*)\s*=\s*[-@+!]*(\S+)", text, re.M): directive, target = m.group(1), m.group(2) if not target.startswith("/"): continue if any(target.startswith(d) for d in SUSPICIOUS_EXEC_DIRS): scan.add("persistence", "CRITICAL", "systemd unit 從暫存目錄執行程式", str(path), f"{directive}={target} —— 開機自動執行的程式位於世界可寫的暫存目錄。", [m.group(0).strip()]) continue if unit_is_packaged or not pkgdb.available: continue # 目標不存在＝該 unit 是休眠的（對應套件沒裝），不可能執行，不算跡證。 if not os.path.exists(target): continue if not pkgdb.owns(target) and target.startswith(("/usr/", "/bin", "/sbin", "/opt/")): scan.add("persistence", "MEDIUM", "非套件 unit 執行非套件管理的程式", str(path), f"{directive}={target} —— unit 檔與執行目標都不屬於任何已安裝套件。", [m.group(0).strip()], standing=("systemd.unit", str(path))) def _home_dirs(include_root: bool = True) -> list[Path]: out = [] try: for entry in pwd.getpwall(): if entry.pw_uid == 0 and not include_root: continue if entry.pw_uid != 0 and entry.pw_uid < 1000: continue home = Path(entry.pw_dir) if safe_isdir(home) and home not in out: out.append(home) except OSError: pass return out def collect_processes(host: Host, scan: Scan, pkgdb: PackageDB) -> None: """執行中行程的異常特徵。多數判斷需要 root 才看得到別人的 /proc//exe。""" unreadable = 0 total = 0 for pid_dir in Path("/proc").iterdir(): if not pid_dir.name.isdigit(): continue total += 1 pid = pid_dir.name try: cmdline = read_text(pid_dir / "cmdline", 8192).replace("\x00", " ").strip() except OSError: continue if not cmdline: continue # 核心執行緒，無使用者空間映像，正常 try: exe = os.readlink(pid_dir / "exe") except PermissionError: unreadable += 1 continue except OSError: continue if exe.endswith(" (deleted)"): real = exe[: -len(" (deleted)")] # 套件升級後舊行程仍在跑會出現此狀態，屬常見良性情形，故列 MEDIUM 讓人自行判斷 scan.add("process", "MEDIUM", "行程的執行檔已從磁碟刪除", f"pid {pid} {Path(real).name}", "映像檔已不在磁碟上。套件升級後未重啟的服務會呈現此狀態（良性），" "但「落地即刪」也是惡意程式隱藏自身的標準手法。請確認該行程啟動時間是否早於最近一次升級。", [f"pid={pid} exe={real}", f"cmdline={cmdline[:160]}"]) elif exe.startswith("/memfd:") or "memfd:" in exe: scan.add("process", "CRITICAL", "行程由匿名記憶體檔案執行", f"pid {pid}", "memfd 執行代表程式碼從未落地到檔案系統，是無檔案惡意程式的典型特徵。", [f"pid={pid} exe={exe}", f"cmdline={cmdline[:160]}"]) elif any(exe.startswith(d) for d in SUSPICIOUS_EXEC_DIRS): scan.add("process", "HIGH", "行程從暫存目錄執行", f"pid {pid} {Path(exe).name}", "執行檔位於世界可寫的暫存目錄。", [f"pid={pid} exe={exe}", f"cmdline={cmdline[:160]}"]) elif pkgdb.available and exe.startswith(("/usr/bin/", "/usr/sbin/", "/bin/", "/sbin/")) \ and not pkgdb.owns(exe): scan.add("process", "MEDIUM", "行程執行系統路徑下的非套件檔案", f"pid {pid} {Path(exe).name}", f"{exe} 位於系統目錄但不屬於任何套件。", [f"pid={pid} exe={exe}", f"cmdline={cmdline[:160]}"]) scan.facts["process_count"] = total if unreadable: scan.skip("行程執行檔檢查", f"{unreadable}/{total} 個行程無法讀取 /proc//exe（需 root）") def collect_network(host: Host, scan: Scan) -> None: """直接解析 /proc/net，不依賴 ss/netstat。""" inode_to_pid = _socket_inode_map() s_listen = scan.surface("listener") for proto, path, family in ( ("tcp", "/proc/net/tcp", socket.AF_INET), ("tcp6", "/proc/net/tcp6", socket.AF_INET6), ("udp", "/proc/net/udp", socket.AF_INET), ("udp6", "/proc/net/udp6", socket.AF_INET6), ): text = read_text(Path(path)) for line in text.splitlines()[1:]: f = line.split() if len(f) < 10: continue local, state, inode = f[1], f[3], f[9] listening = (proto.startswith("tcp") and state == "0A") or \ (proto.startswith("udp") and f[2].split(":")[0] == "00000000") if not listening: continue addr, port = _parse_hex_addr(local, family) if addr is None: continue owner = inode_to_pid.get(inode, "?") exposed = addr not in ("127.0.0.1", "::1") and not addr.startswith("127.") key = f"{proto}/{addr}:{port}" s_listen[key] = owner if exposed: scan.facts.setdefault("exposed_listeners", []).append(f"{key} ({owner})") if not inode_to_pid: scan.skip("監聽埠對應行程", "無法讀取其他使用者的 /proc//fd（需 root）") def _socket_inode_map() -> dict[str, str]: out = {} for pid_dir in Path("/proc").iterdir(): if not pid_dir.name.isdigit(): continue fd_dir = pid_dir / "fd" try: entries = list(os.scandir(fd_dir)) except OSError: continue name = read_text(pid_dir / "comm").strip() or "?" for entry in entries: try: target = os.readlink(entry.path) except OSError: continue if target.startswith("socket:["): out[target[8:-1]] = f"{name}[{pid_dir.name}]" return out def _parse_hex_addr(token: str, family) -> tuple[str | None, int]: try: hexaddr, hexport = token.split(":") port = int(hexport, 16) raw = bytes.fromhex(hexaddr) if family == socket.AF_INET: return socket.inet_ntop(family, raw[::-1]), port packed = b"".join(raw[i:i + 4][::-1] for i in range(0, 16, 4)) return socket.inet_ntop(family, packed), port except (ValueError, OSError): return None, 0 def collect_filesystem(host: Host, scan: Scan, pkgdb: PackageDB, fast: bool) -> None: """SUID/SGID 清查與系統路徑下的非套件執行檔。""" s_suid = scan.surface("suid") s_unpack = scan.surface("unpackaged") if fast: roots = _dedupe_dirs([Path("/usr/bin"), Path("/usr/sbin"), Path("/usr/local/bin"), Path("/usr/local/sbin"), Path("/opt")]) scan.skip("全檔案系統 SUID 掃描", "--fast 模式只掃系統執行檔目錄") else: roots = [Path("/")] scanned = 0 for root in roots: for path, st in _walk_device(root, host): scanned += 1 mode = st.st_mode if mode & (stat.S_ISUID | stat.S_ISGID): if host.under_mount(str(path), host.nosuid_mounts): continue # 掛載於 nosuid，SUID 位無效 bits = ("u" if mode & stat.S_ISUID else "") + ("g" if mode & stat.S_ISGID else "") digest = sha256_file(path) or "?" s_suid[str(path)] = f"{bits}|{oct(mode & 0o7777)}|{st.st_uid}|{digest}" if pkgdb.available and not pkgdb.owns(str(path)): scan.add("filesystem", "CRITICAL", "非套件管理的 SUID/SGID 執行檔", str(path), "此檔具備提權能力但不屬於任何已安裝套件——乾淨系統上不應存在。", [f"mode={oct(mode & 0o7777)} uid={st.st_uid}"]) if mode & 0o111 and str(path).startswith( ("/usr/bin/", "/usr/sbin/", "/usr/local/bin/", "/usr/local/sbin/", "/opt/")): if pkgdb.available and not pkgdb.owns(str(path)): s_unpack[str(path)] = sha256_file(path) or "?" scan.facts["files_scanned"] = scanned scan.facts["suid_count"] = len(s_suid) # 暫存目錄內的執行檔 for d in (Path("/tmp"), Path("/var/tmp"), Path("/dev/shm")): for f in iter_files(d, "*", depth=2): try: st = os.lstat(f) except OSError: continue if stat.S_ISREG(st.st_mode) and st.st_mode & 0o111: # 決定嚴重度的是 noexec 而非 nosuid：nosuid 只擋提權，不擋執行。 noexec = host.under_mount(str(f), host.noexec_mounts) sev = "MEDIUM" if noexec else "HIGH" scan.add("filesystem", sev, "暫存目錄下存在可執行檔", str(f), "世界可寫目錄中的執行檔常是攻擊者落地的第一站。" + ("（該掛載點為 noexec，無法直接執行，但仍可被直譯器讀取執行）" if noexec else "（該掛載點未設 noexec，可直接執行）"), [f"mode={oct(st.st_mode & 0o7777)} uid={st.st_uid} size={st.st_size}"]) def _walk_device(root: Path, host: Host): """只在 root 所在的區塊裝置上行走，避免跨進虛擬檔案系統或外接磁碟。""" stack = [root] while stack: cur = stack.pop() if str(cur) in PRUNE_DIRS: continue try: entries = list(os.scandir(cur)) except OSError: continue for entry in entries: try: st = entry.stat(follow_symlinks=False) is_dir = entry.is_dir(follow_symlinks=False) is_file = entry.is_file(follow_symlinks=False) except OSError: continue if host.root_dev is not None and st.st_dev != host.root_dev: continue if is_dir: if entry.name in PRUNE_NAMES or entry.path in PRUNE_DIRS: continue stack.append(Path(entry.path)) elif is_file: yield Path(entry.path), st def collect_accounts(host: Host, scan: Scan) -> None: s_acct = scan.surface("account") uid0 = [] for line in read_text(Path("/etc/passwd")).splitlines(): parts = line.split(":") if len(parts) < 7: continue name, _, uid, gid, _, home, shell = parts[:7] s_acct[name] = f"{uid}:{gid}:{shell}" if uid == "0": uid0.append(name) if len(uid0) > 1: scan.add("account", "CRITICAL", "存在多個 UID 0 帳號", ",".join(uid0), "除 root 外的 UID 0 帳號是後門帳號的經典形式。", uid0) shadow = Path("/etc/shadow") if os.access(shadow, os.R_OK): for line in read_text(shadow).splitlines(): parts = line.split(":") if len(parts) > 1 and parts[1] == "": scan.add("account", "CRITICAL", "帳號無密碼", parts[0], "該帳號的密碼欄位為空，可能允許無密碼登入。", [parts[0]]) else: scan.skip("空密碼帳號檢查", "/etc/shadow 無讀取權限（需 root）") def collect_kernel(host: Host, scan: Scan) -> None: s_mod = scan.surface("kmod") culprits = [] for line in read_text(Path("/proc/modules")).splitlines(): parts = line.split() if not parts: continue name = parts[0] # 把每個模組自身的 taint 旗標一起納入指紋：日後有新的樹外模組被載入， # 基準線比對會直接抓到，不必依賴整體 taint 數值。 mtaint = read_text(Path("/sys/module") / name / "taint").strip() s_mod[name] = mtaint or "-" if mtaint: culprits.append(f"{name} ({mtaint})") if culprits: scan.surface("kernel.taint")["modules"] = ",".join(sorted(culprits)) tainted = read_text(Path("/proc/sys/kernel/tainted")).strip() if tainted.isdigit() and int(tainted): flags = _decode_taint(int(tainted)) interesting = {"O": "載入了樹外(out-of-tree)模組", "E": "載入了未簽章模組", "P": "載入了專有模組", "F": "模組被強制載入"} hits = [f"{k}={v}" for k, v in interesting.items() if k in flags] if hits: # 桌機裝 NVIDIA/VirtualBox/DKMS 驅動就會永久 tainted，設 HIGH 只會變成長期噪音。 # 這裡給 MEDIUM 並點名是哪些模組造成的，新出現的模組由基準線比對負責升級告警。 scan.add("kernel", "MEDIUM", "核心處於 tainted 狀態", ",".join(sorted(flags)), "有非發行版簽章的程式碼進入核心。專有顯卡驅動或 DKMS 模組會正常造成此狀態；" "重點是下面點名的模組是否都是你自己裝的。", hits + ([f"造成 taint 的模組: {', '.join(culprits[:12])}"] if culprits else []), standing=("kernel.taint", "modules") if culprits else None) def _decode_taint(value: int) -> set[str]: names = "PFSRMBUDAWCIOELKXTN" return {names[i] for i in range(len(names)) if value & (1 << i)} def collect_policy(host: Host, scan: Scan, cfg: dict) -> None: """把加固決策寫成可機器驗證的斷言 —— 抓的是「設定悄悄跑掉」而非入侵。""" policy = dict(DEFAULT_SYSCTL_POLICY) policy.update(cfg.get("sysctl_policy", {})) drift = [] for key, expected in policy.items(): if key.startswith("_"): continue # 設定檔內以底線開頭的鍵是註解，不是 sysctl path = Path("/proc/sys") / key.replace(".", "/") actual = read_text(path).strip() if not actual: continue # 該鍵不存在於此核心，不算違規 allowed = str(expected).split("|") if actual not in allowed: drift.append((key, expected, actual)) for key, expected, actual in drift: scan.add("policy", "MEDIUM", "sysctl 加固值與政策不符", key, f"預期 {expected}，實際 {actual}。設定可能被開機順序競態、套件更新或人工變更覆蓋。", [f"{key} = {actual} (expected {expected})"]) # send_redirects 是 OR 邏輯：任一介面為 1 即會送出，all=0 蓋不掉既有介面 per_iface = Path("/proc/sys/net/ipv4/conf") if safe_isdir(per_iface): bad = [] for iface in per_iface.iterdir(): # lo 不轉送封包，其 send_redirects 值無實際意義，排除以免成為永久雜訊 if iface.name in ("all", "default", "lo"): continue v = read_text(iface / "send_redirects").strip() if v == "1": bad.append(iface.name) if bad: scan.add("policy", "LOW", "個別介面仍會送出 ICMP redirect", ",".join(sorted(bad)), "send_redirects 是 OR 邏輯，all=0 無法覆蓋既有介面；" "開機後才建立的介面（如 virbr0）需靠 default=0 繼承。", sorted(bad)) def collect_aur(host: Host, scan: Scan) -> None: """AUR 供應鏈追蹤 —— 針對「本地編譯的第三方 PKGBUILD」這條攻擊路徑。""" if host.pkg != "pacman": return rc, out = run(["pacman", "-Qm"], timeout=60) if rc != 0: scan.skip("AUR 套件列舉", "pacman -Qm 執行失敗") return s_pkg = scan.surface("aur.pkg") s_src = scan.surface("aur.source") pkgs = [] for line in out.splitlines(): parts = line.split() if len(parts) == 2: s_pkg[parts[0]] = parts[1] pkgs.append(parts[0]) scan.facts["aur_count"] = len(pkgs) cache_roots = [Path(h) / ".cache" / "yay" for h in [str(x) for x in _home_dirs(include_root=False)]] cache_roots += [Path("/var/cache/private/yay")] found_pkgbuild = 0 for pkg in pkgs: for cache in cache_roots: pb = cache / pkg / "PKGBUILD" if not safe_isfile(pb): continue found_pkgbuild += 1 text = read_text(pb, 256 * 1024) s_src[f"{pkg}:PKGBUILD"] = hashlib.sha256(text.encode()).hexdigest()[:32] for url in sorted(set(re.findall(r"(https?://[^\s\"')]+)", text))): s_src[f"{pkg}:url:{url}"] = "present" for inst in (cache / pkg).glob("*.install"): s_src[f"{pkg}:install:{inst.name}"] = sha256_file(inst) or "?" scan.add("supplychain", "LOW", "AUR 套件含 .install 腳本", f"{pkg}/{inst.name}", ".install 腳本會以 root 在安裝/升級時執行，是供應鏈投毒的高價值落點。" "本身不是異常，內容已納入基準線追蹤，日後被改動會升級為 HIGH 告警。", [str(inst)], standing=("aur.source", f"{pkg}:install:{inst.name}")) for pattern, label, sev in ( (r"\b(curl|wget)\b[^\n|]*\|\s*(ba|z|k)?sh\b", "下載後直接管線給 shell 執行", "HIGH"), (r"base64\s+-d[^\n|]*\|\s*(ba)?sh\b", "base64 解碼後直接執行", "HIGH"), (r"\beval\b[^\n]*(\$\(|`)", "eval 搭配命令替換", "MEDIUM"), (r"\bchmod\s+[+]?[0-7]*s\b", "建置期設定 SUID 位", "HIGH"), ): # 純 eval "package() {...}" 是 PKGBUILD 常見的樣板寫法，不算跡證， # 因此只在 eval 同時出現命令替換時才記錄。 if re.search(pattern, text): scan.add("supplychain", sev, "AUR PKGBUILD 含高風險建置指令", f"{pkg}: {label}", "建置腳本會以你的身分執行；此類指令可在編譯期取得任意程式碼。", [ln.strip() for ln in text.splitlines() if re.search(pattern, ln)][:5], standing=("aur.source", f"{pkg}:PKGBUILD")) break if pkgs and not found_pkgbuild: scan.skip("AUR PKGBUILD 追蹤", "找不到 yay 建置快取（升級後被清掉屬正常，" "代價是無法比對 source= 是否被換掉）") def collect_docker(host: Host, scan: Scan) -> None: if not have("docker"): return rc, out = run(["docker", "ps", "--format", "{{.Names}}\t{{.Image}}\t{{.Ports}}"], timeout=60) if rc != 0: scan.skip("Docker 容器列舉", "docker 指令無法執行（權限或 daemon 未啟動）") return s_ctr = scan.surface("docker.container") names = [] for line in out.splitlines(): parts = line.split("\t") if len(parts) >= 2: s_ctr[parts[0]] = parts[1] names.append(parts[0]) scan.facts["container_count"] = len(names) if not names: return rc, out = run(["docker", "inspect", "--format", "{{.Name}}\t{{.HostConfig.Privileged}}\t{{.HostConfig.NetworkMode}}\t" "{{range .HostConfig.Binds}}{{.}} {{end}}"] + names, timeout=90) if rc != 0: return for line in out.splitlines(): parts = line.split("\t") if len(parts) < 4: continue name, priv, netmode, binds = parts[0].lstrip("/"), parts[1], parts[2], parts[3] if priv == "true": scan.add("container", "HIGH", "容器以 privileged 模式執行", name, "privileged 容器可直接存取宿主裝置，逃逸幾乎沒有阻力。", [line], standing=("docker.container", name)) if "docker.sock" in binds: scan.add("container", "HIGH", "容器掛載 docker.sock", name, "掛載 docker socket 等同賦予宿主 root 權限。" "若為 dockge/uptime-kuma 這類管理工具屬已知取捨。", [binds[:200]], standing=("docker.container", name)) if netmode == "host": scan.add("container", "MEDIUM", "容器使用 host 網路", name, "容器直接使用宿主網路命名空間，繞過 Docker 網路隔離。", [line], standing=("docker.container", name)) def collect_ioc(host: Host, scan: Scan, cfg: dict, ioc_path: Path | None) -> None: """比對使用者提供的 IOC 清單（雜湊 / IP / 路徑樣式）。""" if ioc_path is None or not safe_isfile(ioc_path): return try: ioc = json.loads(read_text(ioc_path, 4 * 1024 * 1024)) except (ValueError, OSError) as exc: scan.skip("IOC 比對", f"無法解析 {ioc_path}: {exc}") return hashes = {h.lower()[:32] for h in ioc.get("sha256", [])} paths = ioc.get("paths", []) hits = 0 if hashes: for surface in ("suid", "unpackaged"): for path, fp in scan.surface(surface).items(): for token in fp.split("|"): if token.lower() in hashes: hits += 1 scan.add("ioc", "CRITICAL", "檔案雜湊命中 IOC 清單", path, f"雜湊 {token} 出現在 {ioc_path.name} 中。", [path]) for pattern in paths: for match in Path("/").glob(pattern.lstrip("/")): hits += 1 scan.add("ioc", "HIGH", "路徑命中 IOC 清單", str(match), f"符合樣式 {pattern}。", [str(match)]) scan.facts["ioc_hits"] = hits # -------------------------------------------------------------------------- # 基準線比對 # -------------------------------------------------------------------------- def diff_baseline(scan: Scan, baseline: dict, allowlist: list[str]) -> None: base_surfaces = baseline.get("surfaces", {}) for name, current in scan.surfaces.items(): spec = SURFACES.get(name) if spec is None: continue prior = base_surfaces.get(name) if prior is None: continue # 基準線建立後才新增的面向，不視為漂移 label = spec["label"] added = [k for k in current if k not in prior] removed = [k for k in prior if k not in current] changed = [k for k in current if k in prior and current[k] != prior[k]] for key, kind, sev in ( (added, "added", spec["added"]), (changed, "changed", spec["changed"]), (removed, "removed", spec["removed"]), ): items = [i for i in key if not _allowed(f"{name}:{i}", allowlist)] if not items: continue verb = {"added": "新增", "changed": "內容變更", "removed": "消失"}[kind] # 大量同時變動通常是升級，單獨一兩項變動才是可疑訊號 if len(items) > 25: scan.add("drift", sev_max("LOW", _downgrade(sev)), f"{label}：{len(items)} 項{verb}", name, "變動數量龐大，通常代表系統升級而非入侵；仍建議掃過清單。", items[:30]) else: for item in items: detail = f"{label} 自基準線以來{verb}。" if kind == "changed": detail += f"\n 基準線指紋: {prior.get(item, '?')[:64]}\n 目前指紋: {current.get(item, '?')[:64]}" scan.add("drift", sev, f"{label}{verb}", item, detail, [item]) def _downgrade(sev: str) -> str: idx = max(0, SEV_ORDER.index(sev) - 1) return SEV_ORDER[idx] def _allowed(key: str, allowlist: list[str]) -> bool: return any(fnmatch.fnmatch(key, pat) for pat in allowlist) # -------------------------------------------------------------------------- # 評分與輸出 # -------------------------------------------------------------------------- def score(findings: list[dict]) -> int: total = sum(SEV_WEIGHT.get(f["severity"], 0) for f in findings) return min(100, total) def render_text(report: dict, verbose: bool) -> str: out = [] host = report["host"] add = out.append add("=" * 78) add(f" vigil {VERSION} — 主機入侵跡證評估") add(f" 主機 {host['hostname']} ({host['distro']}, kernel {host['kernel']})") add(f" 掃描時間 {report['timestamp']} 耗時 {report['duration_sec']}s " f"權限 {'root' if host['is_root'] else 'user(受限)'}") add("=" * 78) counts = report["severity_counts"] risk = report["risk_score"] band = "乾淨" if risk == 0 else "低" if risk < 15 else "中" if risk < 40 else "高" if risk < 70 else "嚴重" add("") add(f"風險分數 {risk}/100 [{band}] " + " ".join(f"{s}:{counts.get(s, 0)}" for s in reversed(SEV_ORDER) if counts.get(s))) quiet = [] if report.get("suppressed_by_allowlist"): quiet.append(f"允許清單壓制 {report['suppressed_by_allowlist']} 筆") if report.get("accepted_via_baseline"): quiet.append(f"基準線已接受 {report['accepted_via_baseline']} 筆常態組態") if quiet: add("（" + "，".join(quiet) + "；這些項目變動時仍會告警）") if report.get("baseline_ts"): add(f"基準線建立於 {report['baseline_ts']}") else: add("尚未建立基準線 —— 只做了啟發式檢查，漂移偵測未啟用（跑 `vigil.py baseline`）") if report["skipped"]: add("") add("─ 覆蓋率缺口 " + "─" * 63) for s in report["skipped"]: add(f" ! {s}") findings = report["findings"] if not findings: add("") add(" 未發現任何跡證。") else: shown = 0 for sev in reversed(SEV_ORDER): group = [f for f in findings if f["severity"] == sev] if not group: continue add("") add(f"─ {sev} ({len(group)}) " + "─" * max(0, 68 - len(sev) - len(str(len(group))))) limit = len(group) if verbose else min(len(group), 12) for f in group[:limit]: add(f" ▸ {f['title']}") if f["subject"]: add(f" 對象: {f['subject']}") for line in f["detail"].splitlines(): if line.strip(): add(f" {line.strip()}") for ev in f["evidence"][:4]: add(f" · {ev[:150]}") shown += 1 if limit < len(group): add(f" … 另有 {len(group) - limit} 筆同級跡證（加 -v 顯示全部）") add("") facts = report.get("facts", {}) if facts: add("─ 統計 " + "─" * 70) for k, v in facts.items(): if isinstance(v, list): v = f"{len(v)} 項: " + ", ".join(str(x) for x in v[:6]) add(f" {k}: {v}") add("") return "\n".join(out) def notify(report: dict, cfg: dict) -> str | None: url = cfg.get("ntfy_url") if not url: return "未設定 ntfy_url，略過通知" threshold = cfg.get("notify_min_severity", "HIGH") tidx = SEV_ORDER.index(threshold) if threshold in SEV_ORDER else 3 hot = [f for f in report["findings"] if SEV_ORDER.index(f["severity"]) >= tidx] if not hot: return "無達到通知門檻的跡證" token = cfg.get("ntfy_token") token_file = cfg.get("ntfy_token_file") if not token and token_file and safe_isfile(Path(token_file)): token = read_text(Path(token_file)).strip() host = report["host"]["hostname"] lines = [f"風險分數 {report['risk_score']}/100，{len(hot)} 筆 >= {threshold}"] for f in hot[:12]: lines.append(f"[{f['severity']}] {f['title']}: {f['subject']}"[:180]) body = "\n".join(lines).encode("utf-8") req = urllib.request.Request(url, data=body, method="POST") req.add_header("Title", f"vigil: {host} 發現 {len(hot)} 筆跡證") req.add_header("Priority", "urgent" if any(f["severity"] == "CRITICAL" for f in hot) else "high") req.add_header("Tags", "rotating_light") if token: req.add_header("Authorization", f"Bearer {token}") try: with urllib.request.urlopen(req, timeout=20) as resp: return f"ntfy 已送出 (HTTP {resp.status})" except (urllib.error.URLError, OSError) as exc: return f"ntfy 送出失敗: {exc}" # -------------------------------------------------------------------------- # 主流程 # -------------------------------------------------------------------------- def load_config(explicit: Path | None) -> tuple[dict, Path | None]: candidates = [explicit] if explicit else CONFIG_CANDIDATES for path in candidates: if path and safe_isfile(path): try: return json.loads(read_text(path)), path except ValueError as exc: print(f"vigil: 設定檔 {path} 解析失敗: {exc}", file=sys.stderr) return {}, None def perform_scan(host: Host, cfg: dict, fast: bool, ioc_path: Path | None) -> Scan: scan = Scan(cfg) pkgdb = PackageDB(host, scan) if not host.is_root: scan.skip("整體覆蓋率", "非 root 執行 —— 其他使用者的行程、/etc/shadow、" "root 的 cron 與金鑰皆不可見。結果只能當作粗篩。") collect_persistence(host, scan, pkgdb) collect_processes(host, scan, pkgdb) collect_network(host, scan) collect_accounts(host, scan) collect_kernel(host, scan) collect_policy(host, scan, cfg) collect_aur(host, scan) collect_docker(host, scan) if not fast: collect_integrity(host, scan) else: scan.skip("套件完整性驗證", "--fast 模式略過（完整驗證需數分鐘）") collect_filesystem(host, scan, pkgdb, fast) collect_ioc(host, scan, cfg, ioc_path) return scan def build_report(host: Host, scan: Scan, baseline: dict | None, allowlist: list[str] | None = None) -> dict: # 允許清單對所有跡證生效，不只漂移比對——已知且已接受的取捨（例如 dockge 掛 # docker.sock）若每次都報，久了就會讓人整份跳過不看。 allowlist = allowlist or [] suppressed = 0 accepted = 0 base_surfaces = (baseline or {}).get("surfaces", {}) kept = [] for f in scan.findings: if allowlist and _allowed(f["key"], allowlist): suppressed += 1 continue ref = f.get("standing") if ref and base_surfaces.get(ref[0], {}).get(ref[1]) is not None: # 該項目在建立基準線時就存在且被接受；若其後有任何變動， # diff_baseline 會另外產生一筆漂移跡證，不會漏報。 accepted += 1 continue kept.append(f) scan.findings = kept # 同一個 key 可能被觸發多次（例如一個 unit 有多行 ExecStart），合併證據只留一筆。 merged: dict[str, dict] = {} for f in scan.findings: prior = merged.get(f["key"]) if prior is None: merged[f["key"]] = f else: prior["severity"] = sev_max(prior["severity"], f["severity"]) for ev in f["evidence"]: if ev not in prior["evidence"]: prior["evidence"].append(ev) scan.findings = list(merged.values()) counts: dict[str, int] = {} for f in scan.findings: counts[f["severity"]] = counts.get(f["severity"], 0) + 1 order = {s: i for i, s in enumerate(reversed(SEV_ORDER))} scan.findings.sort(key=lambda f: (order.get(f["severity"], 9), f["module"], f["subject"])) return { "vigil_version": VERSION, "timestamp": now_ts(), "duration_sec": round(time.time() - scan.started, 1), "host": { "hostname": host.hostname, "distro": host.distro, "kernel": host.kernel, "is_root": host.is_root, }, "baseline_ts": (baseline or {}).get("timestamp"), "risk_score": score(scan.findings), "severity_counts": counts, "suppressed_by_allowlist": suppressed, "accepted_via_baseline": accepted, "findings": scan.findings, "skipped": scan.skipped, "facts": scan.facts, } def cmd_baseline(args, host: Host, cfg: dict) -> int: state = Path(args.state) state.mkdir(parents=True, exist_ok=True) print("正在採集基準線…（完整模式需要幾分鐘）", file=sys.stderr) scan = perform_scan(host, cfg, args.fast, _ioc_path(args, cfg)) baseline = { "vigil_version": VERSION, "timestamp": now_ts(), "host": host.hostname, "distro": host.distro, "is_root": host.is_root, "surfaces": scan.surfaces, } target = state / "baseline.json" tmp = target.with_suffix(".tmp") tmp.write_text(json.dumps(baseline, indent=1, sort_keys=True), encoding="utf-8") os.chmod(tmp, 0o600) tmp.replace(target) total = sum(len(v) for v in scan.surfaces.values()) print(f"基準線已寫入 {target}") print(f" 記錄 {len(scan.surfaces)} 個面向、{total} 個項目") if not host.is_root: print(" ⚠ 非 root 建立的基準線覆蓋不完整；日後以 root 掃描時會出現大量假漂移。") for name in sorted(scan.surfaces): print(f" {name:20s} {len(scan.surfaces[name]):5d}") if scan.skipped: print(" 覆蓋率缺口:") for s in scan.skipped: print(f" ! {s}") return 0 def cmd_scan(args, host: Host, cfg: dict) -> int: state = Path(args.state) baseline = None bpath = state / "baseline.json" if safe_isfile(bpath): try: baseline = json.loads(read_text(bpath, 64 * 1024 * 1024)) except ValueError as exc: print(f"vigil: 基準線損毀 ({exc})，本次僅做啟發式檢查", file=sys.stderr) scan = perform_scan(host, cfg, args.fast, _ioc_path(args, cfg)) if baseline: if baseline.get("is_root") and not host.is_root: scan.skip("基準線比對", "基準線以 root 建立、本次以一般使用者執行，" "差異多為權限造成的假陽性") diff_baseline(scan, baseline, cfg.get("allowlist", [])) report = build_report(host, scan, baseline, cfg.get("allowlist", [])) if args.json: print(json.dumps(report, indent=1, ensure_ascii=False)) else: print(render_text(report, args.verbose)) if args.save: rdir = state / "reports" try: rdir.mkdir(parents=True, exist_ok=True) rp = rdir / f"{time.strftime('%Y%m%d-%H%M%S')}.json" rp.write_text(json.dumps(report, ensure_ascii=False), encoding="utf-8") os.chmod(rp, 0o600) _prune_reports(rdir, keep=int(cfg.get("keep_reports", 60))) if not args.json: print(f"報告已存至 {rp}") except OSError as exc: print(f"vigil: 無法寫入報告: {exc}", file=sys.stderr) if args.notify: msg = notify(report, cfg) if not args.json: print(f"通知: {msg}") if args.fail_on: idx = SEV_ORDER.index(args.fail_on) if any(SEV_ORDER.index(f["severity"]) >= idx for f in report["findings"]): return 2 return 0 def _prune_reports(rdir: Path, keep: int) -> None: files = sorted(rdir.glob("*.json")) for old in files[:-keep] if keep > 0 else []: try: old.unlink() except OSError: pass def _ioc_path(args, cfg: dict) -> Path | None: if args.ioc: return Path(args.ioc) if cfg.get("ioc_file"): return Path(cfg["ioc_file"]) default = Path(__file__).resolve().parent / "rules" / "ioc.json" return default if safe_isfile(default) else None def cmd_show(args, host: Host, cfg: dict) -> int: bpath = Path(args.state) / "baseline.json" if not safe_isfile(bpath): print(f"找不到基準線 {bpath}", file=sys.stderr) return 1 baseline = json.loads(read_text(bpath, 64 * 1024 * 1024)) print(f"基準線 {bpath}") print(f" 建立於 {baseline['timestamp']} on {baseline['host']} " f"({baseline['distro']}, root={baseline.get('is_root')})") for name in sorted(baseline.get("surfaces", {})): items = baseline["surfaces"][name] print(f" {name:20s} {len(items):5d} {SURFACES.get(name, {}).get('label', '')}") if args.verbose: for k in sorted(items)[:2000]: print(f" {k}") return 0 def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser( prog="vigil", description="主機入侵跡證評估代理程式") ap.add_argument("--state", default=str(DEFAULT_STATE), help="狀態目錄") ap.add_argument("--config", help="設定檔路徑") ap.add_argument("--ioc", help="IOC JSON 路徑") ap.add_argument("--fast", action="store_true", help="略過套件完整性驗證與全檔案系統掃描") ap.add_argument("-v", "--verbose", action="store_true") ap.add_argument("--version", action="version", version=f"vigil {VERSION}") sub = ap.add_subparsers(dest="cmd", required=True) p_base = sub.add_parser("baseline", help="建立已知良好基準線") p_base.set_defaults(func=cmd_baseline) p_scan = sub.add_parser("scan", help="採集、偵測並與基準線比對") p_scan.add_argument("--json", action="store_true", help="輸出 JSON") p_scan.add_argument("--notify", action="store_true", help="達門檻時推送 ntfy") p_scan.add_argument("--save", action="store_true", help="保存報告到 state 目錄") p_scan.add_argument("--fail-on", choices=SEV_ORDER, help="出現該等級以上跡證時以 exit code 2 結束") p_scan.set_defaults(func=cmd_scan) p_show = sub.add_parser("show", help="檢視目前基準線") p_show.set_defaults(func=cmd_show) args = ap.parse_args(argv) cfg, cfg_path = load_config(Path(args.config) if args.config else None) if cfg_path and args.verbose: print(f"vigil: 使用設定檔 {cfg_path}", file=sys.stderr) host = Host() try: return args.func(args, host, cfg) except KeyboardInterrupt: print("\n已中斷", file=sys.stderr) return 130 # ========================================================================== # Odysec 自助健檢 —— 客戶端介面 # # 本段由 build.py 附加在 vigil 引擎之後，組成單一可散布檔案。 # 上面的程式碼是採集引擎，這一段是給客戶操作的介面。 # ========================================================================== ODYSEC_FORMAT = 1 ODYSEC_SERVICE = "https://check.odysec.org" BANNER = f""" ╔══════════════════════════════════════════════════════════════════════╗ ║ Odysec 主機入侵跡證健檢 —— 採集程式 ║ ╚══════════════════════════════════════════════════════════════════════╝ 這支程式會做什麼： • 讀取本機的持久化設定、行程清單、監聽埠、帳號、核心模組與檔案完整性 • 把結果寫成一個 JSON 檔案，放在你指定的位置 • 結束 這支程式不會做什麼： • 不連線到任何地方。採集過程完全離線，結果檔案由你自己決定要不要上傳 • 不修改系統上的任何檔案或設定 • 不讀取你的文件、郵件、資料庫內容或任何業務資料 • 不常駐，不安裝，執行完就結束，不留任何東西 這是一份可讀的 Python 原始碼，執行前歡迎自行審閱。 建議以 root 執行以取得完整涵蓋率；非 root 也能跑，但報告會列出未涵蓋的項目。 """ def odysec_collect(args) -> int: if not args.quiet: print(BANNER, file=sys.stderr) host = Host() if not host.is_root: print("⚠ 目前不是以 root 執行 —— 其他使用者的行程、/etc/shadow、" "root 的排程與金鑰都看不到。", file=sys.stderr) print(" 報告會如實列出這些缺口，但涵蓋率會明顯下降。" "建議改用 sudo 重跑。\n", file=sys.stderr) print("採集中…（完整模式視主機規模約需 15 秒至數分鐘）", file=sys.stderr) started = time.time() scan = perform_scan(host, {}, args.fast, None) report = build_report(host, scan, None, []) bundle = { "odysec": { "format": ODYSEC_FORMAT, "token": args.token, "collector_version": VERSION, "generated_at": now_ts(), "label": args.label or host.hostname, }, "report": report, } stamp = time.strftime("%Y%m%d-%H%M%S") safe = "".join(c if c.isalnum() or c in "-_." else "_" for c in host.hostname) out = Path(args.output) if args.output else Path(f"odysec-{safe}-{stamp}.json") try: out.write_text(json.dumps(bundle, ensure_ascii=False, indent=1), encoding="utf-8") os.chmod(out, 0o600) # 以 sudo 執行時，結果檔會是 root 所有、權限 600——使用者回到自己的帳號 # 後會讀不到自己剛產生的檔案，也就無法上傳。把擁有者交還給實際操作的人。 sudo_uid, sudo_gid = os.environ.get("SUDO_UID"), os.environ.get("SUDO_GID") if sudo_uid and sudo_gid and os.geteuid() == 0: os.chown(out, int(sudo_uid), int(sudo_gid)) except (OSError, ValueError) as exc: print(f"寫入結果檔失敗：{exc}", file=sys.stderr) return 1 counts = report.get("severity_counts", {}) summary = "、".join(f"{k} {v}" for k, v in counts.items()) or "無跡證" print(f"\n完成（{round(time.time() - started, 1)} 秒）", file=sys.stderr) print(f"結果檔：{out.resolve()}", file=sys.stderr) print(f"採集項目：{sum(len(v) for v in scan.surfaces.values())} 項　" f"初步跡證：{summary}", file=sys.stderr) if scan.skipped: print(f"涵蓋率缺口：{len(scan.skipped)} 項（報告中會逐項說明）", file=sys.stderr) print(f""" ──────────────────────────────────────────────────────────────────────── 下一步：上傳結果檔以取得完整報告 方式一（指令）： curl -X POST {ODYSEC_SERVICE}/api/upload \\ -H "X-Odysec-Token: {args.token or '你的訂單代碼'}" \\ --data-binary @{out.name} 方式二（網頁）： 開啟 {ODYSEC_SERVICE}/upload 上傳這個檔案 結果檔內含本機的組態指紋（檔案清單、雜湊、監聽埠、帳號名稱）， 屬敏感資料。上傳前歡迎自行檢視內容，也可自行決定不上傳。 ────────────────────────────────────────────────────────────────────────""", file=sys.stderr) return 0 def odysec_main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser( prog="odysec-check", description="Odysec 主機入侵跡證健檢採集程式", epilog="採集過程完全離線；結果檔由你自行決定是否上傳。") ap.add_argument("-t", "--token", help="訂單代碼（會寫進結果檔以對應你的訂單）") ap.add_argument("-o", "--output", help="結果檔輸出路徑") ap.add_argument("-l", "--label", help="這台主機在報告上顯示的名稱") ap.add_argument("--fast", action="store_true", help="快速模式：略過套件完整性驗證與全磁碟掃描") ap.add_argument("-q", "--quiet", action="store_true", help="不顯示說明橫幅") ap.add_argument("--version", action="version", version=f"odysec-check (vigil engine {VERSION})") args = ap.parse_args(argv) try: return odysec_collect(args) except KeyboardInterrupt: print("\n已中斷，未產生結果檔。", file=sys.stderr) return 130 if __name__ == "__main__": sys.exit(odysec_main())