1#!/usr/bin/env python3
 2"""web-demo 主機的只讀維護檢查。
 3
 4在主機上讀同目錄的 host.toml，逐項執行 where = "host" 的只讀檢查，主機外
 5（OCI Console、Cloudflare dashboard、開發機）的項目印成 MANUAL。
 6
 7用法（在主機上，以 root 執行；firewalld 查詢、sshd -T、collector 的資料庫、nginx 的 log 目錄、
 8audit.log 與 journal 一般帳號讀不到，專用帳號的實測見 README〈以專用帳號執行〉）：
 9
10    sudo python3 host-check.py                       # 例行維護：入口、log、累積的資料、更新
11    sudo python3 host-check.py --phase suspend       # 暫停上線前要做的步驟
12    sudo python3 host-check.py --phase resume        # 恢復上線後要做的步驟
13    sudo python3 host-check.py --manifest other.toml --set repo_dir=/opt/monitor
14    sudo python3 host-check.py --state-file PATH     # 結果檔換路徑（預設 /var/lib/host-check/last.json）
15    sudo python3 host-check.py --no-state            # 不讀也不寫結果檔
16
17輸出：每項一行 `PASS` / `WARN` / `MANUAL`、項目名與理由；接著是與前一次的差（`since:` 開頭，
18其後每個差一行：`CHANGED` 狀態改變、`DELTA` 值改變、`NEW` 新出現、`GONE` 消失）；結尾一行
19`summary: N PASS, M WARN, K MANUAL`；沒有印出 summary 代表腳本中途中止。
20
21結果檔：每次執行結束把這次的計數與量到的值（探針值、累積的 bytes、可用 bytes）寫進結果檔
22（JSON，每個 phase 一段，只替換本次的 phase），下一次拿它算差。目錄不存在時以 0700 建立，檔案
230600，用同目錄暫存檔加 os.replace 寫入。目錄已存在時不改它的權限：由 systemd 的 StateDirectory=
24建立的目錄，權限照 StateDirectoryMode=（預設 0755，每次啟動都會把既有目錄改回這個值），沙箱裡
25執行時要設 StateDirectoryMode=0700。讀不到、壞掉或寫不進去只在 stderr 印 `note:`，不改變
26exit code；清單本身只描述要檢查什麼，不存放結果。
27
28systemd 沙箱（在 Oracle Linux 10 容器裡實測）：ProtectSystem=strict 下要加 ReadWritePaths=/var/log /var/cache/dnf
29（dnf needs-restarting 要寫 dnf.log 與 metadata lock），StateDirectory=host-check 加 StateDirectoryMode=0700；
30NoNewPrivileges、PrivateTmp、IPAddressDeny=any 不影響檢查；不開 PrivateNetwork（ss 看不到主機的 socket，
31入口與 listen-audit 的檢查都會失準）。完整設定見 README〈在 systemd 沙箱裡執行〉。
32
33exit code：
34    0  全部 PASS，或只有 PASS 與 MANUAL
35    1  至少一項 WARN
36    2  清單格式錯（缺必填欄位、型別錯、要執行的指令不在只讀白名單裡），一項都沒有執行
37
38只讀的保證：
39    - 指令不經過 shell（shlex 切詞後直接 exec），沒有管線、轉向與命令替換。
40    - 要執行的每一條指令在執行任何一條之前先比對白名單（READONLY_RULES）；
41      不在白名單裡、或帶了會改變狀態的參數（find -delete、
42      firewall-cmd --add-*、sqlite3 沒有同時帶 -init /dev/null、-readonly 與 -safe 這類）時，整份清單判成格式錯。
43    - systemctl、dnf、rpm、journalctl、ss、find 的選項是列允許：依各工具自己的選項語法（getopt_long，
44      吃值的選項會吃掉下一個字，值可黏在選項後）切分，表外的選項一律拒絕，所以工具日後新增的
45      寫入選項預設就是拒絕（例：rpm --pipe/--eval/-D、systemctl -H/-M、dnf -x、journalctl --cursor-file）。
46    - sshd 只接受單獨的 -T；指令的路徑只接受裸名或 /usr/bin、/usr/sbin、/bin、/sbin 下的檔案。
47    - command 一律會被執行（且必須通過白名單）；manual_command、purge_command 是要人執行的動作，
48      一律只印出來。只印的位置（[[log]]、[[suspend]] / [[resume]]、主機外的 [[entrypoint.verify]]）
49      寫舊名 command 判成格式錯。
50    - 寫結果檔之外不改動任何受檢狀態；結果檔是腳本自己唯一會寫的檔案。例外：collector 停止後
51      資料庫的 -shm、-wal 已被收掉，此時 sqlite3 -readonly 會自己建出這兩個檔（擁有者同資料庫）；
52      在唯讀檔案系統（ProtectSystem=strict）上建不出來，sqlite3 exit 14，該項印成 WARN。
53    - 檔案大小與時間用 Python 的 os.lstat 量，不執行 du。
54
55只用標準函式庫，需要 Python 3.11 以上（tomllib）。
56"""
57
58from __future__ import annotations
59
60import argparse
61import configparser
62import datetime as dt
63import glob
64import json
65import os
66import re
67import shlex
68import stat
69import subprocess
70import sys
71import tempfile
72import time
73import tomllib
74from dataclasses import dataclass, field
75from typing import Any, Callable
76
77WHERE_VALUES = ("host", "external", "console")
78EXPOSURE_VALUES = ("public", "cdn-only", "admin-only", "loopback")
79PROTOCOL_VALUES = ("tcp", "udp")
80BILLED_VALUES = ("yes", "no", "unknown", "n/a")
81MEASURE_VALUES = ("lines", "value", "age_days")
82PHASES = ("maintain", "suspend", "resume")
83COMMAND_TIMEOUT_S = 30
84DEFAULT_STATE_FILE = "/var/lib/host-check/last.json"
85STATE_FORMAT = 1
86
87VAR_RE = re.compile(r"\{\{([A-Za-z_][A-Za-z0-9_]*)\}\}")

1class ManifestError(Exception):
2    """清單格式錯；errors 是逐條的原因。"""
3
4    def __init__(self, errors: list[str]):
5        super().__init__("\n".join(errors))
6        self.errors = errors

 1# ── 只讀白名單 ──────────────────────────────────────────────
 2# 每條規則收 argv（argv[0] 已是 basename），回傳 None 代表放行，字串代表拒絕的理由。
 3
 4
 5# 一支工具允許的選項（列允許，表外一律拒絕）；選項名連同開頭的 - 一起寫。
 6@dataclass(frozen=True)
 7class OptionSpec:
 8    flags: frozenset[str] = frozenset()  # 不吃值的旗標
 9    values: frozenset[str] = frozenset()  # 一定吃值的選項
10    optional: frozenset[str] = frozenset()  # 選填值：只吃黏在一起的值，或下一個字是整數時
11
12
13def _spec(flags: str = "", values: str = "", optional: str = "") -> OptionSpec:
14    return OptionSpec(frozenset(flags.split()), frozenset(values.split()), frozenset(optional.split()))
15
16
17def _parse_options(args: list[str], spec: OptionSpec, stop_at_first: bool = False) -> tuple[list[str], set[str], str | None]:
18    # 依 getopt_long 的語法把 args 切成（位置參數, 出現過的選項, 拒絕理由）：
19    # 短選項可合併（-qa），吃值的短選項值黏在後面或在下一個字；長選項 --name[=value]；-- 之後全是位置參數。
20    # stop_at_first 時，第一個位置參數起的全部字都放進位置參數（不再解析選項）。
21    positionals: list[str] = []
22    seen: set[str] = set()
23    i = 0
24    while i < len(args):
25        tok = args[i]
26        i += 1
27        if tok == "--":
28            positionals.extend(args[i:])
29            break
30        if not tok.startswith("-") or tok == "-":
31            positionals.append(tok)
32            if stop_at_first:
33                positionals.extend(args[i:])
34                break
35            continue
36        if tok.startswith("--"):
37            name, eq, _ = tok.partition("=")
38            consumed = _long_option(name, bool(eq), spec, args, i)
39            if isinstance(consumed, str):
40                return positionals, seen, consumed
41            seen.add(name)
42            i += consumed
43            continue
44        for pos in range(1, len(tok)):
45            name = "-" + tok[pos]
46            if name in spec.flags:
47                seen.add(name)
48                continue
49            consumed = _value_option(name, pos + 1 < len(tok), spec, args, i)
50            if isinstance(consumed, str):
51                return positionals, seen, consumed
52            seen.add(name)
53            i += consumed
54            break
55    return positionals, seen, None
56
57
58def _long_option(name: str, has_value: bool, spec: OptionSpec, args: list[str], i: int) -> int | str:
59    # 回傳這個長選項多吃掉幾個後續字；拒絕時回傳理由。
60    if name in spec.flags:
61        return f"選項 {name} 不吃值" if has_value else 0
62    return _value_option(name, has_value, spec, args, i)
63
64
65def _value_option(name: str, has_glued_value: bool, spec: OptionSpec, args: list[str], i: int) -> int | str:
66    if name not in spec.values and name not in spec.optional:
67        return f"選項 {name} 不在允許清單"
68    if has_glued_value:
69        return 0
70    nxt = args[i] if i < len(args) else None
71    if name in spec.optional:
72        return 1 if nxt is not None and nxt.isdigit() else 0
73    return 1 if nxt is not None else f"選項 {name} 缺少值"

 1SYSTEMCTL_READ = {"is-active", "is-enabled", "is-failed", "show", "status", "cat", "list-timers", "list-units", "list-unit-files"}
 2SYSTEMCTL_OPTIONS = _spec(
 3    flags="--no-pager --no-legend --plain --value -a --all -l --full -q --quiet --failed",
 4    values="-p --property -P -t --type --state -n --lines -o --output --timestamp",
 5)
 6
 7
 8def _rule_systemctl(argv: list[str]) -> str | None:
 9    # -p、-t 這類選項會吃掉下一個字，所以子命令要先依選項語法切出來，不能取第一個不以 - 開頭的字。
10    # -H/--host 與 -M/--machine 會把操作送到別台主機或容器（-H 以 root 的 SSH 身分連出去），不在允許清單。
11    positionals, _, err = _parse_options(argv[1:], SYSTEMCTL_OPTIONS)
12    if err:
13        return f"systemctl {err}"
14    if not positionals or positionals[0] not in SYSTEMCTL_READ:
15        return f"systemctl 只允許 {sorted(SYSTEMCTL_READ)}"
16    return None

 1# --cursor-file 與 --vacuum-*、--rotate、--flush、-f 都會寫檔或改變 journal，不在允許清單。
 2JOURNALCTL_OPTIONS = _spec(
 3    flags="-q --quiet --no-pager -r --reverse --utc --no-hostname -k --dmesg -a --all --disk-usage --list-boots",
 4    values="-u --unit --since --until -o --output -g --grep -p --priority -t --identifier",
 5    optional="-n --lines",
 6)
 7
 8
 9def _rule_journalctl(argv: list[str]) -> str | None:
10    _, _, err = _parse_options(argv[1:], JOURNALCTL_OPTIONS)
11    return f"journalctl {err}" if err else None

1FIREWALL_READ = re.compile(r"^--(permanent|zone=\S+|state|query-\S+|list-\S+|get-\S+|info-\S+)$")
2
3
4def _rule_firewall_cmd(argv: list[str]) -> str | None:
5    for a in argv[1:]:
6        if not FIREWALL_READ.match(a):
7            return f"firewall-cmd 只允許 --query-*、--list-*、--get-*、--info-*、--state、--permanent、--zone=，不允許 {a}"
8    return None

 1# -K（kill）、-D（寫檔）、-N（換 netns）、-F（讀過濾檔）、-r（反查 DNS，對外連線）、-E 都不在允許清單。
 2# 位置參數是過濾式（state listening、sport = :443 這類），ss 只讀它。
 3SS_OPTIONS = _spec(
 4    flags=(
 5        "-H --no-header -l --listening -n --numeric -t --tcp -u --udp -a --all -p --processes -4 --ipv4 -6 --ipv6 "
 6        "-x --unix -w --raw -e --extended -i --info -m --memory -s --summary -o --options"
 7    ),
 8    values="-f --family -A --query",
 9)
10
11
12def _rule_ss(argv: list[str]) -> str | None:
13    _, _, err = _parse_options(argv[1:], SS_OPTIONS)
14    return f"ss {err}" if err else None

 1# find 不是 getopt 語法，所以自己走一遍：選項只認下面兩張表（-delete、-exec*、-ok*、-fprint*、-fls、
 2# -D、-files0-from 都不在裡面）；值只吃下一個字；不以 - 開頭的字是路徑或運算子的括號、驚嘆號。
 3FIND_LEADING = frozenset({"-H", "-L", "-P"})
 4FIND_OPERATORS = frozenset({"(", ")", "!", ",", "-not", "-and", "-or", "-a", "-o"})
 5FIND_FLAGS = frozenset(
 6    "-print -print0 -prune -empty -xdev -mount -true -false -readable -writable -executable".split()
 7)
 8FIND_VALUES = frozenset(
 9    (
10        "-name -iname -path -ipath -regex -iregex -type -xtype -mtime -mmin -atime -amin -ctime -cmin "
11        "-newer -size -user -group -uid -gid -perm -links -inum -maxdepth -mindepth -printf"
12    ).split()
13)
14
15
16def _rule_find(argv: list[str]) -> str | None:
17    args = argv[1:]
18    i = 0
19    while i < len(args) and args[i] in FIND_LEADING:
20        i += 1
21    while i < len(args):
22        tok = args[i]
23        i += 1
24        if tok in FIND_FLAGS or tok in FIND_OPERATORS:
25            continue
26        if tok in FIND_VALUES:
27            if i >= len(args):
28                return f"find 的選項 {tok} 缺少值"
29            i += 1
30            continue
31        if tok.startswith("-") and tok != "-":
32            return f"find 的選項 {tok} 不在允許清單"
33    return None

 1# sqlite3 的長選項是單破折號整字（--x 等同 -x），不是 getopt 語法，所以自己走一遍：先把開頭的 -- 正規化成 -，再整字比對。
 2SQLITE_FLAGS = frozenset(
 3    "-readonly -safe -bail -batch -header -noheader -csv -json -line -list -box -table -markdown -column -quote -tabs -nofollow".split()
 4)
 5SQLITE_VALUES = frozenset("-separator -newline -nullvalue".split())
 6SQLITE_REQUIRED = frozenset({"-readonly", "-safe", "-init"})
 7
 8
 9def _rule_sqlite3(argv: list[str]) -> str | None:
10    # -readonly 只讓資料庫檔唯讀；SQL 函式 writefile()、edit() 與 ATTACH 仍能寫檔案系統。
11    # -safe 擋下這些函式與會寫檔的 dot-command（OL10 的 3.46.1 實測支援）。
12    # -init 檔與 ~/.sqliterc 在 safe mode 生效前就執行（.shell 照跑，實測會建檔），HOME 擋不住
13    # （sqlite3 用 getpwuid 找家目錄）；唯一的擋法是 -init /dev/null 取代 ~/.sqliterc。三個都要帶。
14    args = argv[1:]
15    seen: set[str] = set()
16    positionals: list[str] = []
17    i = 0
18    while i < len(args):
19        tok = args[i]
20        i += 1
21        if not tok.startswith("-"):
22            positionals.append(tok)
23            continue
24        name = tok[1:] if tok.startswith("--") else tok
25        if name in SQLITE_FLAGS:
26            seen.add(name)
27        elif name == "-init":
28            if i >= len(args) or args[i] != "/dev/null":
29                return "sqlite3 的 -init 的值必須是 /dev/null"
30            seen.add(name)
31            i += 1
32        elif name in SQLITE_VALUES:
33            if i >= len(args):
34                return f"sqlite3 的選項 {tok} 缺少值"
35            i += 1
36        else:
37            return f"sqlite3 的選項 {tok} 不在允許清單"
38    if not SQLITE_REQUIRED <= seen:
39        return "sqlite3 要同時帶 -init /dev/null、-readonly 與 -safe（~/.sqliterc 與 writefile() 都會寫檔）"
40    for sql in positionals[1:]:
41        if any(line.lstrip().startswith(".") for line in sql.splitlines()):
42            return "sqlite3 的 dot-command（.backup、.output、.shell 這類）不允許"
43    return None

 1OPENSSL_X509_OPTS = {"-noout", "-in", "-checkend", "-enddate", "-startdate", "-dates", "-subject", "-issuer", "-serial", "-fingerprint", "-sha256", "-ext"}
 2
 3
 4def _rule_openssl(argv: list[str]) -> str | None:
 5    if len(argv) < 2 or argv[1] != "x509" or "-noout" not in argv:
 6        return "openssl 只允許 x509 -noout"
 7    for a in argv[2:]:
 8        if a.startswith("-") and a not in OPENSSL_X509_OPTS:
 9            return f"openssl x509 不允許 {a}"
10    return None

1def _rule_sshd(argv: list[str]) -> str | None:
2    # -E 與 -f 會寫檔或換設定，所以只接受單獨的 -T。
3    return None if argv[1:] == ["-T"] else "sshd 只允許單獨的 -T（印出生效中的設定）"

 1NEEDS_RESTARTING_OPTIONS = _spec(flags="-r -s --reboothint --services")
 2
 3
 4def _rule_needs_restarting(argv: list[str]) -> str | None:
 5    positionals, _, err = _parse_options(argv[1:], NEEDS_RESTARTING_OPTIONS)
 6    if err or positionals:
 7        return "needs-restarting 只允許 -r、-s、--reboothint、--services"
 8    return None
 9
10
11DNF_OPTIONS = _spec(flags="-q --quiet -C --cacheonly")
12NO_OPTIONS = _spec()
13
14
15def _rule_dnf(argv: list[str]) -> str | None:
16    # -x、--setopt、-c、--installroot 這類吃值的選項不在允許清單；只允許子命令前的 -q、-C。
17    rest, _, err = _parse_options(argv[1:], DNF_OPTIONS, stop_at_first=True)
18    if err:
19        return f"dnf {err}"
20    if rest[:1] == ["needs-restarting"]:
21        return _rule_needs_restarting(rest)
22    if rest[:2] in (["history", "list"], ["history", "info"]):
23        _, _, err = _parse_options(rest[2:], NO_OPTIONS)
24        return f"dnf {err}" if err else None
25    return "dnf 只允許 needs-restarting 與 history list/info"

 1# --eval、-E、-D/--define 展開巨集，%(...) 會經 shell 執行；--pipe 直接執行指令；
 2# --macros、--rcfile、--dbpath、--root、-p 換設定或資料來源。都不在允許清單。
 3RPM_OPTIONS = _spec(
 4    flags=(
 5        "-q -a -f -i -l -c -d -s -R --query --all --file --info --list --configfiles --docfiles "
 6        "--state --requires --provides --changelog --last --whatprovides --whatrequires"
 7    ),
 8    values="--qf --queryformat",
 9)
10
11
12def _rule_rpm(argv: list[str]) -> str | None:
13    # 第一個選項必須是查詢模式（-q、-qa、-qi 這類或 --query），不讓 -i、-U 這類模式排在前面。
14    if len(argv) < 2 or not (argv[1].startswith("-q") or argv[1] == "--query"):
15        return "rpm 只允許查詢（第一個選項要是 -q 開頭或 --query）"
16    _, seen, err = _parse_options(argv[1:], RPM_OPTIONS)
17    if err:
18        return f"rpm {err}"
19    if not ({"-q", "--query"} & seen):
20        return "rpm 只允許查詢（要帶 -q）"
21    return None

 1# ausearch 是單破折號多字母選項，整字比對。--checkpoint、-if/--input 這類寫檔或換輸入來源的選項不在允許清單。
 2AUSEARCH_FLAGS = frozenset("-i --interpret -r --raw --just-one".split())
 3AUSEARCH_VALUES = frozenset(
 4    "-m --message -k --key -c --comm -p --pid -ui --uid -x --executable -sc --syscall -sv --success --format".split()
 5)
 6AUSEARCH_TIME = frozenset("-ts --start -te --end".split())
 7AUSEARCH_CLOCK = re.compile(r"^\d{1,2}:\d{2}(:\d{2})?$")
 8
 9
10def _rule_ausearch(argv: list[str]) -> str | None:
11    args = argv[1:]
12    i = 0
13    while i < len(args):
14        tok = args[i]
15        i += 1
16        name, eq, _ = tok.partition("=") if tok.startswith("--") else (tok, "", "")
17        if name in AUSEARCH_FLAGS and not eq:
18            continue
19        if name in AUSEARCH_VALUES or name in AUSEARCH_TIME:
20            if not eq:
21                if i >= len(args):
22                    return f"ausearch 的選項 {tok} 缺少值"
23                i += 1
24                # -ts 後面是日期時，再下一個字可能是時間（-ts 10/26/2026 10:30）
25                if name in AUSEARCH_TIME and i < len(args) and AUSEARCH_CLOCK.match(args[i]):
26                    i += 1
27            continue
28        return f"ausearch 的選項 {tok} 不在允許清單"
29    return None

 1def _allow(argv: list[str]) -> str | None:
 2    return None
 3
 4
 5READONLY_RULES: dict[str, Callable[[list[str]], str | None]] = {
 6    "ss": _rule_ss,
 7    "systemctl": _rule_systemctl,
 8    "journalctl": _rule_journalctl,
 9    "firewall-cmd": _rule_firewall_cmd,
10    "find": _rule_find,
11    "sqlite3": _rule_sqlite3,
12    "openssl": _rule_openssl,
13    "sshd": _rule_sshd,
14    "needs-restarting": _rule_needs_restarting,
15    "dnf": _rule_dnf,
16    "rpm": _rule_rpm,
17    "ausearch": _rule_ausearch,
18    "grep": _allow,
19    "wc": _allow,
20    "stat": _allow,
21    "du": _allow,
22    "df": _allow,
23    "test": _allow,
24    "id": _allow,
25    "uname": _allow,
26    "getenforce": _allow,
27    "sestatus": _allow,
28    "getsebool": _allow,
29}

 1# Checker 實際會拼出來的指令；validate 載入時用同一批函式先比對白名單，兩邊不會不一致。
 2LISTEN_COMMAND = "ss -H -l -n -t -u"
 3UPDATE_TIMER_VERBS = (("is-enabled", "enabled"), ("is-active", "active"))
 4# needs-restarting -r 判定要重開機時印的句子（dnf 4 與 dnf 5 相同，Runner 以 LC_ALL=C 執行）。
 5# exit 1 也是 dnf 自己出錯時的 exit（ProtectSystem=strict 下寫不了 /var/log/dnf.log），只看 exit 分不開。
 6REBOOT_REQUIRED_TEXT = "Reboot is required"
 7
 8
 9def timer_command(verb: str, timer: str) -> str:
10    return f"systemctl {verb} {shlex.quote(timer)}"
11
12
13def journal_command(unit: str) -> str:
14    return f"journalctl -u {shlex.quote(unit)} -n 1 -q --no-pager -o short-iso"
15
16
17TRUSTED_BIN_DIRS = frozenset({"/usr/bin", "/usr/sbin", "/bin", "/sbin"})
18
19
20def readonly_violation(command: str) -> str | None:
21    """command 不在只讀白名單時回傳理由。"""
22    try:
23        argv = shlex.split(command)
24    except ValueError as e:
25        return f"指令切詞失敗：{e}"
26    if not argv:
27        return "指令是空的"
28    if "/" in argv[0] and os.path.dirname(argv[0]) not in TRUSTED_BIN_DIRS:
29        return f"{argv[0]} 的路徑不在 {', '.join(sorted(TRUSTED_BIN_DIRS))} 之下（只接受裸名或系統目錄）"
30    name = os.path.basename(argv[0])
31    rule = READONLY_RULES.get(name)
32    if rule is None:
33        return f"{name} 不在只讀白名單（{', '.join(sorted(READONLY_RULES))}）"
34    return rule([name] + argv[1:])

  1# ── 清單解析與驗證 ─────────────────────────────────────────
  2
  3
  4@dataclass
  5class Probe:
  6    command: str
  7    expect_exit: int | None = None
  8    ok_exit: list[int] = field(default_factory=lambda: [0])
  9    expect_stdout: str | None = None
 10    expect_contains: str | None = None
 11    expect_not_contains: str | None = None
 12    measure: str | None = None
 13    warn_above: float | None = None
 14
 15
 16class _Validator:
 17    def __init__(self, variables: dict[str, str]):
 18        self.errors: list[str] = []
 19        self.vars = variables
 20
 21    def err(self, where: str, msg: str) -> None:
 22        self.errors.append(f"{where}: {msg}")
 23
 24    def subst(self, where: str, value: str) -> str:
 25        def repl(m: re.Match[str]) -> str:
 26            name = m.group(1)
 27            if name not in self.vars:
 28                self.err(where, f"{{{{{name}}}}} 沒有定義在 [vars]")
 29                return m.group(0)
 30            return self.vars[name]
 31
 32        return VAR_RE.sub(repl, value)
 33
 34    def get(self, tbl: dict[str, Any], key: str, typ: type | tuple[type, ...], where: str, required: bool = True, enum: tuple[str, ...] | None = None) -> Any:
 35        if key not in tbl:
 36            if required:
 37                self.err(where, f"缺少必填欄位 {key}")
 38            return None
 39        v = tbl[key]
 40        # bool 是 int 的子類別，數字欄位不收 true/false。
 41        if not isinstance(v, typ) or (isinstance(v, bool) and bool not in (typ if isinstance(typ, tuple) else (typ,))):
 42            self.err(where, f"{key} 的型別要是 {_typename(typ)}，實際是 {type(v).__name__}")
 43            return None
 44        if enum is not None and v not in enum:
 45            self.err(where, f"{key} 要是 {list(enum)} 之一，實際是 {v!r}")
 46            return None
 47        if isinstance(v, str):
 48            return self.subst(f"{where}.{key}", v)
 49        return v
 50
 51    def reject_old_command_name(self, tbl: dict[str, Any], where: str) -> None:
 52        # command 一律會被執行；只印不執行的欄位叫 manual_command，舊名出現在只印的位置判成格式錯。
 53        if "command" in tbl:
 54            self.err(where, "只印不執行的指令改用 manual_command（command 一律會被執行並過白名單）")
 55
 56    def str_list(self, tbl: dict[str, Any], key: str, where: str, required: bool = True) -> list[str] | None:
 57        v = self.get(tbl, key, list, where, required)
 58        if v is None:
 59            return None
 60        if not all(isinstance(x, str) for x in v):
 61            self.err(where, f"{key} 的每一項都要是字串")
 62            return None
 63        return [self.subst(f"{where}.{key}", x) for x in v]
 64
 65    def table_list(self, data: dict[str, Any], key: str, required: bool = True) -> list[dict[str, Any]]:
 66        v = data.get(key)
 67        if v is None:
 68            if required:
 69                self.err(key, f"缺少 [[{key}]]")
 70            return []
 71        if not isinstance(v, list) or not all(isinstance(x, dict) for x in v):
 72            self.err(key, f"{key} 要是 [[{key}]] 表格陣列")
 73            return []
 74        return v
 75
 76    def probe(self, tbl: dict[str, Any], where: str, allow_measure: bool = False) -> Probe | None:
 77        command = self.get(tbl, "command", str, where)
 78        if command is None:
 79            return None
 80        reason = readonly_violation(command)
 81        if reason:
 82            self.err(where, f"command 不是只讀指令：{reason}（{command}）")
 83        p = Probe(command=command)
 84        p.expect_exit = self.get(tbl, "expect_exit", int, where, required=False)
 85        ok_exit = tbl.get("ok_exit", [0])
 86        if not isinstance(ok_exit, list) or not all(isinstance(x, int) and not isinstance(x, bool) for x in ok_exit):
 87            self.err(where, "ok_exit 要是整數陣列")
 88        else:
 89            p.ok_exit = ok_exit
 90        p.expect_stdout = self.get(tbl, "expect_stdout", str, where, required=False)
 91        p.expect_contains = self.get(tbl, "expect_contains", str, where, required=False)
 92        p.expect_not_contains = self.get(tbl, "expect_not_contains", str, where, required=False)
 93        if allow_measure:
 94            p.measure = self.get(tbl, "measure", str, where, required=False, enum=MEASURE_VALUES)
 95            p.warn_above = self.get(tbl, "warn_above", (int, float), where, required=p.measure is not None)
 96        has_expect = any(x is not None for x in (p.expect_exit, p.expect_stdout, p.expect_contains, p.expect_not_contains, p.measure))
 97        if not has_expect and "ok_exit" not in tbl:
 98            self.err(where, "探針至少要有一個預期值（expect_exit、ok_exit、expect_stdout、expect_contains、expect_not_contains 或 measure）")
 99        return p
100
101
102def _typename(typ: type | tuple[type, ...]) -> str:
103    if isinstance(typ, tuple):
104        return " 或 ".join(t.__name__ for t in typ)
105    return typ.__name__

  1def load_manifest(path: str, overrides: dict[str, str] | None = None) -> dict[str, Any]:
  2    """讀取並驗證清單，回傳正規化後的結構；格式錯時丟 ManifestError。"""
  3    try:
  4        with open(path, "rb") as f:
  5            data = tomllib.load(f)
  6    except OSError as e:
  7        raise ManifestError([f"讀不到清單 {path}：{e}"]) from e
  8    except tomllib.TOMLDecodeError as e:
  9        raise ManifestError([f"{path} 不是合法的 TOML 1.0：{e}"]) from e
 10    return validate(data, overrides or {})
 11
 12
 13def validate(data: dict[str, Any], overrides: dict[str, str]) -> dict[str, Any]:
 14    raw_vars = data.get("vars", {})
 15    errors: list[str] = []
 16    if not isinstance(raw_vars, dict) or not all(isinstance(v, str) for v in raw_vars.values()):
 17        errors.append("vars: [vars] 的每個值都要是字串")
 18        raw_vars = {}
 19    variables = dict(raw_vars)
 20    for k in overrides:
 21        if k not in variables:
 22            errors.append(f"--set {k}: [vars] 沒有這個名稱")
 23    variables.update(overrides)
 24    v = _Validator(variables)
 25    v.errors.extend(errors)
 26    out: dict[str, Any] = {"vars": variables}
 27
 28    host = data.get("host")
 29    if not isinstance(host, dict):
 30        v.err("host", "缺少 [host]")
 31        host = {}
 32    out["host"] = {k: v.get(host, k, str, "host") for k in ("name", "provider", "region", "role")}
 33
 34    la = data.get("listen_audit", {"protocols": ["tcp"], "allow": []})
 35    if not isinstance(la, dict):
 36        v.err("listen_audit", "要是表格")
 37        la = {}
 38    protocols = v.str_list(la, "protocols", "listen_audit") or []
 39    for p in protocols:
 40        if p not in PROTOCOL_VALUES:
 41            v.err("listen_audit", f"protocols 只能是 {list(PROTOCOL_VALUES)}，實際有 {p!r}")
 42    allow_out = []
 43    allow = la.get("allow", [])
 44    if not isinstance(allow, list):
 45        v.err("listen_audit", "allow 要是陣列")
 46        allow = []
 47    for i, a in enumerate(allow):
 48        w = f"listen_audit.allow[{i}]"
 49        if not isinstance(a, dict):
 50            v.err(w, "要是 { port = ..., protocol = ..., reason = ... }")
 51            continue
 52        allow_out.append({"port": v.get(a, "port", int, w), "protocol": v.get(a, "protocol", str, w, enum=PROTOCOL_VALUES), "reason": v.get(a, "reason", str, w)})
 53    out["listen_audit"] = {"protocols": protocols, "allow": allow_out}
 54
 55    eps = []
 56    for i, ep in enumerate(v.table_list(data, "entrypoint")):
 57        w = f"entrypoint[{i}]"
 58        name = v.get(ep, "name", str, w)
 59        w = f"entrypoint[{name or i}]"
 60        e = {
 61            "name": name,
 62            "port": v.get(ep, "port", int, w),
 63            "protocol": v.get(ep, "protocol", str, w, enum=PROTOCOL_VALUES),
 64            "exposure": v.get(ep, "exposure", str, w, enum=EXPOSURE_VALUES),
 65            "where": v.get(ep, "where", str, w, enum=WHERE_VALUES),
 66            "enforced_by": v.str_list(ep, "enforced_by", w) or [],
 67            "verify": [],
 68        }
 69        verifies = ep.get("verify", [])
 70        if not isinstance(verifies, list) or not all(isinstance(x, dict) for x in verifies):
 71            v.err(w, "verify 要是 [[entrypoint.verify]] 表格陣列")
 72            verifies = []
 73        layers_seen = set()
 74        for j, vf in enumerate(verifies):
 75            vw = f"{w}.verify[{j}]"
 76            layer = v.get(vf, "layer", str, vw)
 77            where = v.get(vf, "where", str, vw, enum=WHERE_VALUES)
 78            layers_seen.add(layer)
 79            item: dict[str, Any] = {"layer": layer, "where": where}
 80            if where == "host":
 81                item["probe"] = v.probe(vf, vw)
 82            elif where is not None:
 83                item["console_path"] = v.get(vf, "console_path", str, vw, required=False)
 84                v.reject_old_command_name(vf, vw)
 85                item["manual_command"] = v.get(vf, "manual_command", str, vw, required=False)
 86                if item["console_path"] is None and item["manual_command"] is None:
 87                    v.err(vw, "主機外的確認要有 console_path 或 manual_command")
 88                item["expect"] = v.get(vf, "expect", str, vw)
 89            e["verify"].append(item)
 90        for layer in e["enforced_by"]:
 91            if layer not in layers_seen:
 92                v.err(w, f"enforced_by 的 {layer!r} 沒有對應的 [[entrypoint.verify]]（layer = {layer!r}）")
 93        for layer in layers_seen - set(e["enforced_by"]) - {None}:
 94            v.err(w, f"[[entrypoint.verify]] 的 layer {layer!r} 不在 enforced_by 裡")
 95        eps.append(e)
 96    out["entrypoint"] = eps
 97
 98    logs = []
 99    for i, lg in enumerate(v.table_list(data, "log")):
100        name = v.get(lg, "name", str, f"log[{i}]")
101        w = f"log[{name or i}]"
102        v.reject_old_command_name(lg, w)
103        source = v.get(lg, "source", str, w)
104        if source is not None and not re.match(r"^(file|journal):.+", source):
105            v.err(w, f"source 要寫成 file:<路徑> 或 journal:<unit>，實際是 {source!r}")
106        item = {
107            "name": name,
108            "source": source,
109            "look_for": v.get(lg, "look_for", str, w),
110            "interval_days": v.get(lg, "interval_days", int, w),
111            "manual_command": v.get(lg, "manual_command", str, w),
112            "alerts": [],
113        }
114        alerts = lg.get("alert", [])
115        if not isinstance(alerts, list) or not all(isinstance(x, dict) for x in alerts):
116            v.err(w, "alert 要是 [[log.alert]] 表格陣列")
117            alerts = []
118        for j, al in enumerate(alerts):
119            aw = f"{w}.alert[{j}]"
120            aname = v.get(al, "name", str, aw)
121            item["alerts"].append({"name": aname, "probe": v.probe(al, f"{w}.alert[{aname or j}]", allow_measure=True)})
122        logs.append(item)
123    out["log"] = logs
124
125    accs = []
126    for i, ac in enumerate(v.table_list(data, "accumulation")):
127        name = v.get(ac, "name", str, f"accumulation[{i}]")
128        w = f"accumulation[{name or i}]"
129        paths = ac.get("path")
130        if isinstance(paths, str):
131            paths = [paths]
132            ac = dict(ac, path=paths)
133        item = {
134            "name": name,
135            "paths": v.str_list(ac, "path", w) or [],
136            "missing_ok": v.get(ac, "missing_ok", bool, w, required=False) or False,
137            "limit_bytes": v.get(ac, "limit_bytes", int, w, required=False),
138            "retention_days": v.get(ac, "retention_days", int, w, required=False),
139            "purge_command": v.get(ac, "purge_command", str, w),
140            "oldest_command": None,
141        }
142        if item["limit_bytes"] is None and item["retention_days"] is None:
143            v.err(w, "limit_bytes 與 retention_days 至少要有一個")
144        oc = v.get(ac, "oldest_command", str, w, required=False)
145        if oc is not None:
146            if item["retention_days"] is None:
147                v.err(w, "有 oldest_command 時要有 retention_days")
148            reason = readonly_violation(oc)
149            if reason:
150                v.err(w, f"oldest_command 不是只讀指令：{reason}（{oc}）")
151            item["oldest_command"] = oc
152        accs.append(item)
153    out["accumulation"] = accs
154
155    fss = []
156    for i, fs in enumerate(v.table_list(data, "filesystem", required=False)):
157        name = v.get(fs, "name", str, f"filesystem[{i}]")
158        w = f"filesystem[{name or i}]"
159        fss.append({"name": name, "mount": v.get(fs, "mount", str, w), "min_free_bytes": v.get(fs, "min_free_bytes", int, w)})
160    out["filesystem"] = fss
161
162    up = data.get("updates")
163    if not isinstance(up, dict):
164        v.err("updates", "缺少 [updates]")
165        up = {}
166    expect = up.get("expect", {})
167    if not isinstance(expect, dict) or not all(isinstance(x, str) for x in expect.values()):
168        v.err("updates", "expect 要是值為字串的 inline table")
169        expect = {}
170    reboot = v.get(up, "reboot_check", str, "updates", required=False)
171    if reboot is not None:
172        reason = readonly_violation(reboot)
173        if reason:
174            v.err("updates", f"reboot_check 不是只讀指令：{reason}")
175    out["updates"] = {
176        "mode": v.get(up, "mode", str, "updates"),
177        "timer": v.get(up, "timer", str, "updates"),
178        "config_files": v.str_list(up, "config_files", "updates", required=False) or [],
179        "expect": expect,
180        "reboot_check": reboot,
181    }
182
183    for phase in ("suspend", "resume"):
184        steps = []
185        for i, st in enumerate(v.table_list(data, phase)):
186            w = f"{phase}[{i}]"
187            item = {
188                "step": v.get(st, "step", int, w),
189                "resource": v.get(st, "resource", str, w),
190                "billed_when_stopped": v.get(st, "billed_when_stopped", str, w, enum=BILLED_VALUES),
191                "billed_when_stopped_beyond_free": v.get(st, "billed_when_stopped_beyond_free", str, w, enum=BILLED_VALUES),
192                "billing_basis": v.get(st, "billing_basis", str, w),
193                "where": v.get(st, "where", str, w, enum=WHERE_VALUES),
194                "manual_command": v.get(st, "manual_command", str, w, required=False),
195                "console_path": v.get(st, "console_path", str, w, required=False),
196                "verify": None,
197            }
198            v.reject_old_command_name(st, w)
199            if item["manual_command"] is None and item["console_path"] is None:
200                v.err(w, "manual_command 或 console_path 至少要有一個")
201            vf = st.get("verify")
202            if item["where"] == "host":
203                if not isinstance(vf, dict):
204                    v.err(w, "where = \"host\" 的 verify 要是只讀探針的 inline table")
205                else:
206                    item["verify"] = v.probe(vf, f"{w}.verify")
207            elif item["where"] is not None:
208                if not isinstance(vf, str):
209                    v.err(w, "主機外的 verify 要是一句要看到的結果（字串）")
210                else:
211                    item["verify"] = v.subst(f"{w}.verify", vf)
212            steps.append(item)
213        nums = [s["step"] for s in steps if s["step"] is not None]
214        if len(nums) != len(set(nums)):
215            v.err(phase, "step 有重複的編號")
216        out[phase] = sorted(steps, key=lambda s: s["step"] or 0)
217
218    _check_builtin_commands(v, out)
219    if v.errors:
220        raise ManifestError(v.errors)
221    return out
222
223
224def _check_builtin_commands(v: "_Validator", out: dict[str, Any]) -> None:
225    # Checker 自己拼的指令也含清單的值（timer 名稱、journal unit），載入時一併比對白名單，不合就判成格式錯。
226    cmds = [("listen_audit", LISTEN_COMMAND)]
227    timer = out["updates"]["timer"]
228    if isinstance(timer, str):
229        cmds += [("updates.timer", timer_command(verb, timer)) for verb, _ in UPDATE_TIMER_VERBS]
230    for lg in out["log"]:
231        source = lg["source"]
232        if isinstance(source, str) and source.startswith("journal:"):
233            cmds.append((f"log[{lg['name']}].source", journal_command(source.partition(":")[2])))
234    for where, cmd in cmds:
235        reason = readonly_violation(cmd)
236        if reason:
237            v.err(where, f"自動執行的指令不是只讀指令：{reason}（{cmd}）")