第176回では、CLIへ渡す環境変数を許可リストで絞りました。今回は、実行ファイルを開いてSHA-256を照合し、そのファイル記述子から直接起動します。
絶対パスを指定しても、検査後に同じパスを開き直せば、途中で別のファイルへ差し替えられる余地があります。開いたファイル記述子を検査と実行の両方に使えば、このパス差し替えを避けられます。
ただし、ファイルの内容そのものを変更できる利用者がいる場合、検査後から実行までの変更は防げません。SHA-256の一致も発行元を証明する電子署名ではありません。今回は、この限界を含めて扱います。
今回防ぐもの
| 方式 | 確認できること | 残る問題 |
|---|---|---|
PATHから検索する |
検索順で見つかったコマンドを実行する | PATHや配置順によって別の実体が選ばれる |
| 絶対パスを実行する | 指定したパスを開く | 内容が承認時と同じか確認しない |
| パスを検査してから開き直す | 検査時点の内容を確認する | 検査と再オープンの間にパスを差し替えられる |
| 開いたファイルを検査し、その記述子から実行する | 検査したファイルを実行対象にする | 同じファイルの直接変更や動的依存物までは固定しない |
Pythonのos.execve()は、対応するUnix環境ではパスの代わりに開いたファイル記述子を受け取れます。Linuxのfexecve(3)にも、パスの交換は避けられる一方、検査後の内容変更は別途権限で防ぐ必要があると記載されています。
実行契約
| 項目 | 仕様 |
|---|---|
| 対象 | Python 3.10以上のLinux。os.execveのファイル記述子指定に対応する環境 |
| 実行ファイル | 4バイト以上256MiB以下の通常ファイル。先頭4バイトがELFマジックであること |
| パス | 正規化済みの絶対パス。先頭の//と末尾要素のシンボリックリンクを拒否する |
| 権限 | 読み取り可能で、いずれかの実行権限ビットがあること。setuid、setgid、グループ書き込み、その他書き込みを拒否する。最終的な実行可否はカーネルが判定する |
| ポリシー | BOMなしUTF-8のJSON。最大65,536バイト。未知項目と重複キーを拒否する |
| 引数 | --より後ろをそのまま渡す。最大64個、合計32,768バイト |
| 環境変数 | ポリシー内の文字列だけを渡す。最大32個、合計32,768バイト。親環境は継承しない |
| 標準入出力 | ラッパーでは付け替えない |
| 成功時 | ラッパーのプロセスを対象CLIへ置き換える。成功JSONは出力しない |
| 対象外 | スクリプト、ELFインタープリタ、共有ライブラリ、設定、プラグイン、引数の意味、ネットワークアクセスの検証 |
O_NOFOLLOWが拒否するのはパスの最後の要素です。途中のディレクトリにあるシンボリックリンクは解決されます。ただし、オープン後の検査と実行は同じファイル記述子を使うため、途中のパスが後から変わっても実行対象は切り替わりません。
ファイル記述子から実行するラッパー
次のコードをverified_exec.pyとして保存します。
from __future__ import annotations
import hashlib
import hmac
import json
import os
import re
import stat
import sys
from typing import Any
MAX_POLICY_BYTES = 65_536
MAX_EXECUTABLE_BYTES = 268_435_456
MAX_PATH_BYTES = 4_095
MAX_ARGUMENTS = 64
MAX_ARGUMENT_BYTES = 32_768
MAX_ENVIRONMENT_ITEMS = 32
MAX_ENVIRONMENT_BYTES = 32_768
CHUNK_BYTES = 1_048_576
SHA256_PATTERN = re.compile(r"[0-9a-f]{64}")
POLICY_KEYS = {"schema_version", "executable", "sha256", "environment"}
class AppError(Exception):
def __init__(
self,
exit_code: int,
kind: str,
message: str,
details: dict[str, Any] | None = None,
) -> None:
super().__init__(message)
self.exit_code = exit_code
self.kind = kind
self.message = message
self.details = details or {}
def emit(stream: Any, payload: dict[str, Any]) -> None:
print(
json.dumps(
payload,
ensure_ascii=False,
separators=(",", ":"),
sort_keys=True,
),
file=stream,
flush=True,
)
def object_without_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
result: dict[str, Any] = {}
for key, value in pairs:
if key in result:
raise AppError(2, "invalid_policy", "JSONに重複キーがあります")
result[key] = value
return result
def reject_nonstandard_number(value: str) -> None:
raise AppError(2, "invalid_policy", f"JSONの数値 {value} は使用できません")
def validate_executable_path(value: Any, error_kind: str) -> None:
if not isinstance(value, str):
raise AppError(2, error_kind, "実行パスは文字列にしてください")
if "\x00" in value:
raise AppError(2, error_kind, "実行パスにNUL文字は使用できません")
if (
not value.startswith("/")
or value.startswith("//")
or os.path.normpath(value) != value
):
raise AppError(
2,
error_kind,
"実行パスは正規化済みの絶対パスにしてください",
)
try:
encoded = os.fsencode(value)
except UnicodeEncodeError as exc:
raise AppError(
2,
error_kind,
"実行パスをファイルシステムの文字コードへ変換できません",
) from exc
if len(encoded) > MAX_PATH_BYTES:
raise AppError(2, error_kind, "実行パスが長すぎます")
def validate_environment(value: Any) -> None:
if not isinstance(value, dict):
raise AppError(2, "invalid_policy", "environmentはJSONオブジェクトにしてください")
if len(value) > MAX_ENVIRONMENT_ITEMS:
raise AppError(2, "invalid_policy", "環境変数は32個以内にしてください")
total = 0
for name, item in value.items():
if not isinstance(name, str) or not isinstance(item, str):
raise AppError(2, "invalid_policy", "環境変数の名前と値は文字列にしてください")
if not name or "=" in name or "\x00" in name or "\x00" in item:
raise AppError(2, "invalid_policy", "環境変数の名前または値が不正です")
try:
name_bytes = os.fsencode(name)
item_bytes = os.fsencode(item)
except UnicodeEncodeError as exc:
raise AppError(
2,
"invalid_policy",
"環境変数をファイルシステムの文字コードへ変換できません",
) from exc
total += len(name_bytes) + len(item_bytes) + 2
if total > MAX_ENVIRONMENT_BYTES:
raise AppError(2, "invalid_policy", "環境変数が合計32768バイトを超えています")
def read_policy(path: str) -> dict[str, Any]:
try:
with open(path, "rb") as source:
raw = source.read(MAX_POLICY_BYTES + 1)
except OSError as exc:
raise AppError(
66,
"policy_read_failed",
"ポリシーファイルを読み取れません",
{"errno": exc.errno},
) from exc
if len(raw) > MAX_POLICY_BYTES:
raise AppError(2, "invalid_policy", "ポリシーが65536バイトを超えています")
if raw.startswith(b"\xef\xbb\xbf"):
raise AppError(2, "invalid_policy", "UTF-8のBOMは使用できません")
try:
text = raw.decode("utf-8")
except UnicodeDecodeError as exc:
raise AppError(2, "invalid_policy", "ポリシーはUTF-8ではありません") from exc
try:
data = json.loads(
text,
object_pairs_hook=object_without_duplicates,
parse_constant=reject_nonstandard_number,
)
except json.JSONDecodeError as exc:
raise AppError(
2,
"invalid_policy",
"ポリシーは正しいJSONではありません",
{"line": exc.lineno, "column": exc.colno},
) from exc
except (RecursionError, ValueError) as exc:
raise AppError(2, "invalid_policy", "ポリシーを解析できません") from exc
if not isinstance(data, dict):
raise AppError(2, "invalid_policy", "JSONの最上位はオブジェクトにしてください")
if set(data) != POLICY_KEYS:
raise AppError(2, "invalid_policy", "ポリシーの項目が不足または過剰です")
if type(data["schema_version"]) is not int or data["schema_version"] != 1:
raise AppError(2, "invalid_policy", "schema_versionは整数の1にしてください")
validate_executable_path(data["executable"], "invalid_policy")
if not isinstance(data["sha256"], str) or not SHA256_PATTERN.fullmatch(
data["sha256"]
):
raise AppError(2, "invalid_policy", "sha256は小文字16進数64桁にしてください")
validate_environment(data["environment"])
return data
def validate_arguments(values: list[str]) -> None:
if len(values) > MAX_ARGUMENTS:
raise AppError(2, "invalid_arguments", "実行引数は64個以内にしてください")
total = 0
for value in values:
if "\x00" in value:
raise AppError(2, "invalid_arguments", "実行引数にNUL文字は使用できません")
try:
encoded = os.fsencode(value)
except UnicodeEncodeError as exc:
raise AppError(
2,
"invalid_arguments",
"実行引数をファイルシステムの文字コードへ変換できません",
) from exc
total += len(encoded) + 1
if total > MAX_ARGUMENT_BYTES:
raise AppError(2, "invalid_arguments", "実行引数が合計32768バイトを超えています")
def check_platform() -> None:
required_flags = ("O_CLOEXEC", "O_NOFOLLOW")
if (
not sys.platform.startswith("linux")
or os.execve not in os.supports_fd
or any(not hasattr(os, name) for name in required_flags)
):
raise AppError(
126,
"unsupported_platform",
"この環境ではファイル記述子からexecveを実行できません",
)
def open_executable(path: str) -> int:
check_platform()
flags = os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW
try:
return os.open(path, flags)
except OSError as exc:
raise AppError(
126,
"executable_open_failed",
"実行ファイルを開けません",
{"errno": exc.errno},
) from exc
def metadata_tuple(info: os.stat_result) -> tuple[int, ...]:
return (
info.st_dev,
info.st_ino,
info.st_mode,
info.st_uid,
info.st_gid,
info.st_size,
info.st_mtime_ns,
info.st_ctime_ns,
)
def examine_open_file(fd: int) -> str:
try:
before = os.fstat(fd)
except OSError as exc:
raise AppError(
126,
"stat_failed",
"実行ファイルを確認できません",
{"errno": exc.errno},
) from exc
if not stat.S_ISREG(before.st_mode):
raise AppError(65, "executable_rejected", "実行対象は通常ファイルではありません")
if before.st_size < 4 or before.st_size > MAX_EXECUTABLE_BYTES:
raise AppError(65, "executable_rejected", "実行ファイルのサイズが許容範囲外です")
if before.st_mode & 0o111 == 0:
raise AppError(65, "executable_rejected", "実行権限ビットがありません")
if before.st_mode & (stat.S_ISUID | stat.S_ISGID):
raise AppError(65, "executable_rejected", "setuidまたはsetgidファイルは実行しません")
if before.st_mode & (stat.S_IWGRP | stat.S_IWOTH):
raise AppError(65, "executable_rejected", "グループまたはその他から書き込み可能です")
digest = hashlib.sha256()
prefix = b""
size = 0
try:
os.lseek(fd, 0, os.SEEK_SET)
while True:
chunk = os.read(fd, CHUNK_BYTES)
if not chunk:
break
if len(prefix) < 4:
prefix = (prefix + chunk)[:4]
size += len(chunk)
if size > MAX_EXECUTABLE_BYTES:
raise AppError(
65,
"executable_changed",
"検査中に実行ファイルのサイズが上限を超えました",
)
digest.update(chunk)
after = os.fstat(fd)
os.lseek(fd, 0, os.SEEK_SET)
except OSError as exc:
raise AppError(
126,
"executable_read_failed",
"実行ファイルを検査できません",
{"errno": exc.errno},
) from exc
if prefix != b"\x7fELF":
raise AppError(65, "executable_rejected", "先頭4バイトがELFマジックではありません")
if size != before.st_size or metadata_tuple(before) != metadata_tuple(after):
raise AppError(65, "executable_changed", "検査中に実行ファイルが変化しました")
return digest.hexdigest()
def inspect_executable(path: str) -> None:
validate_executable_path(path, "invalid_arguments")
fd = open_executable(path)
try:
digest = examine_open_file(fd)
finally:
os.close(fd)
emit(
sys.stdout,
{
"environment": {"LANG": "C"},
"executable": path,
"schema_version": 1,
"sha256": digest,
},
)
def run_executable(policy_path: str, arguments: list[str]) -> None:
policy = read_policy(policy_path)
validate_arguments(arguments)
fd = open_executable(policy["executable"])
try:
digest = examine_open_file(fd)
if not hmac.compare_digest(digest, policy["sha256"]):
raise AppError(
65,
"sha256_mismatch",
"実行ファイルのSHA-256がポリシーと一致しません",
{"observed_sha256": digest},
)
argv = [policy["executable"], *arguments]
try:
os.execve(fd, argv, policy["environment"])
except (OSError, NotImplementedError) as exc:
errno_value = exc.errno if isinstance(exc, OSError) else None
raise AppError(
126,
"exec_failed",
"検査済みファイルを実行できません",
{"errno": errno_value},
) from exc
finally:
os.close(fd)
def print_help() -> None:
print("usage:")
print(" python3 verified_exec.py inspect /absolute/path/to/elf")
print(" python3 verified_exec.py run POLICY.json -- [ARG ...]")
def main(argv: list[str]) -> int:
if argv in (["-h"], ["--help"]):
print_help()
return 0
if len(argv) == 2 and argv[0] == "inspect":
inspect_executable(argv[1])
return 0
if len(argv) >= 3 and argv[0] == "run" and argv[2] == "--":
run_executable(argv[1], argv[3:])
return 0
raise AppError(2, "usage_error", "引数の形式が正しくありません")
if __name__ == "__main__":
try:
raise SystemExit(main(sys.argv[1:]))
except AppError as exc:
emit(
sys.stderr,
{
"status": "error",
"kind": exc.kind,
"message": exc.message,
**exc.details,
},
)
raise SystemExit(exc.exit_code)
except KeyboardInterrupt:
emit(
sys.stderr,
{"status": "error", "kind": "interrupted", "message": "中断されました"},
)
raise SystemExit(130)
except Exception:
emit(
sys.stderr,
{
"status": "error",
"kind": "unexpected_error",
"message": "予期しないエラーが発生しました",
},
)
raise SystemExit(70)
os.supports_fdで、現在のPythonがファイル記述子形式のexecveに対応しているかを確認しています。ファイルの種類、権限、サイズには、パスではなくos.fstat()の結果を使います。
SHA-256は1MiBずつ計算します。検査前後のデバイス番号、inode、モード、所有者、サイズ、更新時刻が変わった場合も実行しません。
1．候補ポリシーを作る
新しいディレクトリで試します。次の例はBash用です。
mkdir verified-exec-demo || exit $?
cd verified-exec-demo || exit $?
umask 077
set -o noclobber
このディレクトリへverified_exec.pyを保存します。例では/usr/bin/printfを使います。環境によって配置や形式が異なるため、通常のELF実行ファイルであることを先に確認してください。
measured_policy=$(python3 verified_exec.py inspect /usr/bin/printf) || exit $?
printf '%s\n' "$measured_policy" > printf-policy.json
unset measured_policy
python3 -m json.tool printf-policy.json
inspectは、現在開けるファイルを測定した結果を出すだけです。この結果だけでは、そのファイルが承認済みであることを証明できません。本番用のSHA-256は、管理されたビルド成果物や別の検証済み環境と照合して決めます。
2．検査したファイルを実行する
python3 verified_exec.py run printf-policy.json -- 'verified: %s\n' ok
run_status=$?
printf 'exit=%s\n' "$run_status"
標準出力は次のとおりです。
verified: ok
exit=0
成功したos.execve()は戻りません。プロセスIDとカレントディレクトリは変わらず、ラッパーは標準入出力を付け替えません。シェルが受け取る終了状態は、実行されたprintfのものです。
このポリシーが渡す環境変数はLANG=Cだけです。固定したPATHなどが必要なら、ポリシーのenvironmentへ明示します。秘密値はポリシーファイルへ保存しないでください。
3．SHA-256不一致を確認する
テスト専用のポリシーを作り、SHA-256を64個の0へ置き換えます。
python3 -c 'import json,sys; data=json.load(sys.stdin); data["sha256"]="0"*64; json.dump(data,sys.stdout,separators=(",",":")); sys.stdout.write("\n")' < printf-policy.json > bad-policy.json
python3 verified_exec.py run bad-policy.json -- 'must not run\n' 2> error.json
run_status=$?
printf '%s\n' "$run_status"
python3 -c 'import json,pathlib; print(json.loads(pathlib.Path("error.json").read_text(encoding="utf-8"))["kind"])'
結果は次のとおりです。
65
sha256_mismatch
must not runは出力されません。エラーJSONには観測したSHA-256も入りますが、その値を自動で承認値へ反映してはいけません。計画した更新か、取り違えかを確認します。
4．対象CLIの終了状態を引き継ぐ
/usr/bin/falseが同じ検査条件を満たす環境では、次のように確認できます。
false_policy=$(python3 verified_exec.py inspect /usr/bin/false) || exit $?
printf '%s\n' "$false_policy" > false-policy.json
unset false_policy
python3 verified_exec.py run false-policy.json --
run_status=$?
printf '%s\n' "$run_status"
出力は1です。これはラッパーの失敗ではなく、実行されたfalseの終了状態です。
終了コードと出力
| 終了コード | 意味 | 出力 |
|---|---|---|
| 対象CLIが決める値 | 対象CLIへのプロセス置換に成功した | 対象CLIが決める |
| 2 | 使用法、JSON、ポリシー、引数の違反 | 標準エラー出力へ1行JSON |
| 65 | 形式、権限、サイズ、変更検知、SHA-256の不一致 | 標準エラー出力へ1行JSON |
| 66 | ポリシーファイルを読み取れない | 標準エラー出力へ1行JSON |
| 70 | 想定外のPython例外 | 標準エラー出力へ1行JSON |
| 126 | 非対応環境、オープン、読み取り、実行の失敗 | 標準エラー出力へ1行JSON |
| 130 | 対象へ置き換わる前にCtrl+Cで中断した | 標準エラー出力へ1行JSON |
対象CLI自身が65や126を返すこともあります。外側の監督処理では、ラッパーが標準エラー出力へ出したJSONの有無も合わせて判定します。
固定できないもの
- 検査後の直接変更：同じファイルを変更できる利用者がいる場合、SHA-256計算後の競合は残ります。実行物の所有者と書き込み権限を分離します。
- ポリシー：実行パスとSHA-256を一緒に書き換えられれば検査は通ります。通常の実行担当者が変更できない場所へ置きます。
- 動的依存物：ELFインタープリタ、共有ライブラリ、設定、プラグイン、証明書は検査対象ではありません。
- 引数の意味：個数、合計サイズ、NUL文字だけを検査します。危険なオプションや出力先は、CLIごとに別の許可規則が必要です。
- スクリプト：先頭がELFではないため拒否します。Pythonインタープリタだけを固定しても、ソースや依存パッケージは固定できません。
- 隔離：ファイル、プロセス、ネットワークへのアクセスを制限するサンドボックスではありません。
運用へ入れる順番
- 信頼できる配布物から実行パスとSHA-256を特定する
- 実行ファイルとポリシーの所有者、書き込み権限を確認する
- 副作用のないCLIで、引数、入出力、終了状態を試す
- 不一致時は自動更新せず、予定した更新か意図しない変更かを調べる
- 必要なら、第173回の排他制御や第175回のタイムアウト監督を外側へ置く
このラッパーが避けるのは、検査したパスを開き直すときの取り違えです。実行ファイルを変更できる権限まで消えるわけではありません。SHA-256の決め方と変更権限を先に固めてから使います。