第172回から第175回では、複数CLIの順次実行、二重起動の防止、再試行、タイムアウト時の停止を扱いました。実行方法を整えても、子へ渡す環境変数が未整理では、不要な設定や秘密値まで引き継ぐ可能性があります。
今回は、親から引き継ぐ変数と固定値をJSONで指定し、それ以外を渡さずにCLIを起動します。対象はPython 3.10以上を実行するPOSIX環境です。外部ライブラリは使いません。
この仕組みが制限するのは、実行対象へ渡す環境変数です。秘密管理、利用者認証、ファイルやネットワークの隔離までは行いません。
現在のプロセスを実行対象へ置き換える
os.execve()は、現在のプロセスを指定したプログラムへ置き換えます。渡した環境マッピングは現在の環境への追加ではなく、新しいプログラムの環境として使われます。execve()自身はPATHから実行ファイルを検索しません。
実行に成功するとPythonラッパーへは戻りません。POSIX環境ではPIDとカレントディレクトリが維持されます。開いているファイル記述子も、close-on-execが設定されたものを除いて引き継がれます。このラッパーは標準入力、標準出力、標準エラー出力を付け替えません。
| 方式 | 向いている用途 | 今回の扱い |
|---|---|---|
subprocess.Popen(..., env=...) |
親が子の完了を待ち、その後に記録や後処理を行う | 使わない |
os.execve(path, argv, env) |
環境を確定した後、その場で実行対象へ置き換わる | 採用する |
execve()後に監査レポートを書く場合は、ラッパーの外側に監督プロセスが必要です。第175回のタイムアウトラッパーなどから今回のラッパーを起動すれば、置換後も監督対象のPIDは変わりません。
今回の実行契約
| 項目 | 仕様 |
|---|---|
| ポリシー | BOMなしUTF-8のJSON。最大65,536バイトで、最上位項目はinherit、required、setの三つだけ |
inherit |
親に存在する場合だけ値を引き継ぐ変数名の配列 |
required |
親に存在しなければ実行を止める変数名の配列。すべてinheritにも含める |
set |
ポリシーで固定する文字列値。inheritとの同名指定は拒否する |
| 変数名 | 英字またはアンダースコアで始まり、以後は英数字またはアンダースコア。重複は不可 |
| 上限 | 子へ渡す変数は最大64件、値は1件8,192バイト。環境全体は各名前=値と終端を数えて65,536バイト |
| 実行ファイル | --の後ろに絶対パスで指定する。シェルとPATH検索は使わない |
| 入出力 | 標準入力、標準出力、標準エラー出力を付け替えない |
| 成功時 | ラッパーを対象CLIへ置き換える。独自の成功メッセージは出さない |
| 失敗時 | 環境変数の値を含まない1行JSONを標準エラー出力へ出す |
65,536バイトは、このサンプルで決めた上限です。OSには引数と環境を合わせた別の上限があります。長い引数列などのために起動を拒否された場合は終了コード126にします。
1．環境ポリシーを作る
次をenv-policy.jsonとして保存します。
{
"inherit": ["JOB_BATCH_ID"],
"required": ["JOB_BATCH_ID"],
"set": {
"APP_MODE": "batch",
"PYTHONUTF8": "1"
}
}
JOB_BATCH_IDは実行元から受け取り、欠けていれば止めます。APP_MODEとPYTHONUTF8は固定値です。
ポリシーファイルへ秘密値を直接書かないでください。対象CLIが秘密値を必要とする場合は、その変数名をinheritとrequiredへ追加し、別の方法で実行元へ設定します。
PATH、HOME、ロケール、プロキシー関連変数は自動では渡りません。必要性と値を確認してから追加します。LD_PRELOAD、LD_LIBRARY_PATH、PYTHONPATHなどは読み込み動作を変えるため、普段から設定されているという理由だけで許可しません。
2．確認用CLIを用意する
次をinspect_env.pyとして保存します。値を表示するのは動作確認用のバッチIDだけです。実運用で秘密値を表示してはいけません。
import json
import os
result = {
"job_batch_id": os.environ.get("JOB_BATCH_ID"),
"app_mode": os.environ.get("APP_MODE"),
"api_token_present": "API_TOKEN" in os.environ,
"path_present": "PATH" in os.environ,
}
print(json.dumps(result, ensure_ascii=False, separators=(",", ":")))
未許可のAPI_TOKENとPATHについては、値ではなく存在の有無だけを返します。
3．許可リスト実行ラッパーを作る
次をenv_exec.pyとして保存します。
from __future__ import annotations
import json
import os
import re
import sys
from pathlib import Path
from typing import Any
MAX_POLICY_BYTES = 65_536
MAX_VARIABLES = 64
MAX_VALUE_BYTES = 8_192
MAX_ENV_BYTES = 65_536
NAME_PATTERN = re.compile(r"[A-Za-z_][A-Za-z0-9_]*\Z")
POLICY_KEYS = {"inherit", "required", "set"}
USAGE = "usage: env_exec.py POLICY.json -- /absolute/executable [arg ...]"
class ContractError(Exception):
pass
def reject_number(_value: str) -> None:
raise ContractError("このポリシーでJSON数値は使用できません")
def reject_constant(value: str) -> None:
raise ContractError(f"JSON定数 {value} は使用できません")
def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
result: dict[str, Any] = {}
for key, value in pairs:
if key in result:
raise ContractError(f"JSONキーが重複しています: {key!r}")
result[key] = value
return result
def validate_name(name: Any, label: str) -> str:
if type(name) is not str or NAME_PATTERN.fullmatch(name) is None:
raise ContractError(
f"{label}の変数名は英字または_で始まる英数字と_だけにしてください"
)
return name
def validate_name_list(value: Any, label: str) -> list[str]:
if type(value) is not list:
raise ContractError(f"{label}は配列にしてください")
if len(value) > MAX_VARIABLES:
raise ContractError(f"{label}は{MAX_VARIABLES}件以下にしてください")
result: list[str] = []
seen: set[str] = set()
for item in value:
name = validate_name(item, label)
if name in seen:
raise ContractError(f"{label}内で変数名が重複しています: {name}")
seen.add(name)
result.append(name)
return result
def encoded_size(value: str, label: str) -> int:
if "\0" in value:
raise ContractError(f"{label}にNUL文字は使用できません")
try:
return len(os.fsencode(value))
except UnicodeEncodeError as exc:
raise ContractError(f"{label}をOS向けに符号化できません") from exc
def validate_value(name: str, value: Any) -> str:
if type(value) is not str:
raise ContractError(f"set.{name}は文字列にしてください")
size = encoded_size(value, name)
if size > MAX_VALUE_BYTES:
raise ContractError(
f"環境変数{name}の値は{MAX_VALUE_BYTES}バイト以下にしてください"
)
return value
def load_policy(path: Path) -> tuple[list[str], list[str], dict[str, str]]:
try:
with path.open("rb") as handle:
raw = handle.read(MAX_POLICY_BYTES + 1)
except OSError as exc:
detail = exc.strerror or type(exc).__name__
raise ContractError(f"ポリシーを読み取れません: {detail}") from exc
if len(raw) > MAX_POLICY_BYTES:
raise ContractError(
f"ポリシーは{MAX_POLICY_BYTES}バイト以下にしてください"
)
if raw.startswith(b"\xef\xbb\xbf"):
raise ContractError("ポリシーはBOMなしUTF-8にしてください")
try:
text = raw.decode("utf-8")
except UnicodeDecodeError as exc:
raise ContractError("ポリシーはBOMなしUTF-8にしてください") from exc
try:
data = json.loads(
text,
object_pairs_hook=unique_object,
parse_int=reject_number,
parse_float=reject_number,
parse_constant=reject_constant,
)
except json.JSONDecodeError as exc:
raise ContractError(
f"JSON構文エラーです: line={exc.lineno}, column={exc.colno}"
) from exc
except RecursionError as exc:
raise ContractError("JSONの入れ子が深すぎます") from exc
if type(data) is not dict:
raise ContractError("ポリシーの最上位はオブジェクトにしてください")
actual_keys = set(data)
if actual_keys != POLICY_KEYS:
missing = sorted(POLICY_KEYS - actual_keys)
unknown = sorted(actual_keys - POLICY_KEYS)
raise ContractError(
f"ポリシー項目が一致しません: missing={missing}, unknown={unknown}"
)
inherit = validate_name_list(data["inherit"], "inherit")
required = validate_name_list(data["required"], "required")
fixed_raw = data["set"]
if type(fixed_raw) is not dict:
raise ContractError("setはオブジェクトにしてください")
if len(fixed_raw) > MAX_VARIABLES:
raise ContractError(f"setは{MAX_VARIABLES}件以下にしてください")
fixed: dict[str, str] = {}
for raw_name, raw_value in fixed_raw.items():
name = validate_name(raw_name, "set")
fixed[name] = validate_value(name, raw_value)
inherit_set = set(inherit)
required_set = set(required)
fixed_set = set(fixed)
if not required_set <= inherit_set:
names = sorted(required_set - inherit_set)
raise ContractError(f"requiredはinheritにも含めてください: {names}")
if inherit_set & fixed_set:
names = sorted(inherit_set & fixed_set)
raise ContractError(f"inheritとsetで変数名が重複しています: {names}")
if len(inherit_set | fixed_set) > MAX_VARIABLES:
raise ContractError(
f"子へ渡す環境変数は合計{MAX_VARIABLES}件以下にしてください"
)
return inherit, required, fixed
def build_environment(
inherit: list[str], required: list[str], fixed: dict[str, str]
) -> dict[str, str]:
missing = [name for name in required if name not in os.environ]
if missing:
raise ContractError(f"必須の環境変数がありません: {missing}")
child_env: dict[str, str] = {}
for name in inherit:
if name in os.environ:
value = os.environ[name]
validate_value(name, value)
child_env[name] = value
child_env.update(fixed)
total = sum(
encoded_size(name, name) + 1 + encoded_size(value, name) + 1
for name, value in child_env.items()
)
if total > MAX_ENV_BYTES:
raise ContractError(
f"子へ渡す環境変数は合計{MAX_ENV_BYTES}バイト以下にしてください"
)
return child_env
def parse_cli(argv: list[str]) -> tuple[Path, list[str]] | None:
if argv in (["-h"], ["--help"]):
return None
if len(argv) < 3 or argv[1] != "--":
raise ContractError(USAGE)
command = argv[2:]
executable = command[0]
if not os.path.isabs(executable):
raise ContractError("実行ファイルは絶対パスで指定してください")
return Path(argv[0]), command
def emit_error(kind: str, message: str, **extra: Any) -> None:
payload: dict[str, Any] = {
"status": "error",
"kind": kind,
"message": message,
}
payload.update(extra)
raw = (
json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n"
).encode("utf-8", errors="backslashreplace")
sys.stderr.buffer.write(raw)
sys.stderr.buffer.flush()
def main(argv: list[str]) -> int:
try:
parsed = parse_cli(argv)
if parsed is None:
print(USAGE)
return 0
policy_path, command = parsed
inherit, required, fixed = load_policy(policy_path)
child_env = build_environment(inherit, required, fixed)
try:
os.execve(command[0], command, child_env)
except FileNotFoundError:
emit_error(
"exec_not_found",
"実行ファイルまたは指定されたインタープリターが見つかりません",
executable=command[0],
)
return 127
except PermissionError:
emit_error(
"exec_not_permitted",
"実行ファイルを起動する権限がありません",
executable=command[0],
)
return 126
except OSError as exc:
emit_error(
"exec_failed",
"実行ファイルを起動できません",
executable=command[0],
errno=exc.errno,
)
return 126
except ContractError as exc:
emit_error("contract_error", str(exc))
return 2
except KeyboardInterrupt:
emit_error("interrupted", "実行前に中断されました")
return 130
if __name__ == "__main__":
raise SystemExit(main(sys.argv[1:]))
object_pairs_hook、parse_int、parse_float、parse_constantを使っています。通常は後の値が残る重複キーを拒否し、このポリシーで使わないJSON数値と、標準JSON外のNaNやInfinityも受け付けません。
ポリシーは上限より1バイトだけ多く読みます。上限超過を判定するために、ファイル全体を読み込む必要はありません。
os.fsencode()で数えるのは、POSIX上でPythonが環境変数をOSへ渡す際の符号化に合わせるためです。ただし、この検査はOSによる起動成功を保証しません。
4．許可した変数だけが見えることを確認する
現在使っているPythonインタープリターのパスを取得します。sys.executableを取得できない場合や、絶対パスでない場合は止めます。
PYTHON_BIN="$(python3 -c 'import sys; print(sys.executable or "")' )"
case "$PYTHON_BIN" in
/*) ;;
*) exit 1 ;;
esac
動作確認用の値を親へ設定して実行します。API_TOKENの値は例示用であり、実在する認証情報ではありません。
JOB_BATCH_ID=batch-20261023 API_TOKEN=sample-not-passed \
"$PYTHON_BIN" env_exec.py env-policy.json -- \
"$PYTHON_BIN" inspect_env.py
標準出力は次の1行です。
{"job_batch_id":"batch-20261023","app_mode":"batch","api_token_present":false,"path_present":false}
JOB_BATCH_IDと固定値は渡り、未許可のAPI_TOKENとPATHは渡っていません。終了コードは確認用CLIの0です。
execve()が渡すのは許可した環境ですが、起動後のプログラムが環境変数を追加または変更することはあります。対象CLI内の環境が将来も同じ件数である、といった判定には使いません。
5．必須変数の不足と対象CLIの失敗を確認する
必須のJOB_BATCH_IDを外したサブシェルで実行します。
(
unset JOB_BATCH_ID
"$PYTHON_BIN" env_exec.py env-policy.json -- \
"$PYTHON_BIN" inspect_env.py
)
echo $?
標準エラー出力と終了コードは次のとおりです。環境変数の値は出力しません。
{"status":"error","kind":"contract_error","message":"必須の環境変数がありません: ['JOB_BATCH_ID']"}
2
実行対象が終了コード7を返す場合は、その値が呼び出し元へ返ります。
JOB_BATCH_ID=batch-20261023 \
"$PYTHON_BIN" env_exec.py env-policy.json -- \
"$PYTHON_BIN" -c 'import sys; sys.exit(7)'
echo $?
7
| 終了コード | 意味 |
|---|---|
| 対象CLIの終了コード | execve()成功後は対象CLIが直接返す。0に限定しない |
| 2 | 引数、ポリシー、必須環境変数の契約違反。ただし対象CLI自身も2を返し得る |
| 126 | 権限不足、形式不正、引数と環境の過大などにより対象を起動できなかった |
| 127 | 実行ファイル、またはスクリプトのshebangが指すインタープリターが見つからなかった |
| 130 | execve()より前にラッパーがCtrl+Cで中断された |
execve()後のCtrl+CやSIGTERMは、置き換わった対象プログラムが直接受けます。最終的な状態は、そのプログラムのシグナル処理に従います。
ラッパー由来の終了コード2と対象CLI由来の2は、終了コードだけでは区別できません。必要なら、標準エラー出力のJSONと組み合わせて判定します。
運用前に確認する制約
- ラッパー自身は、環境を絞る前に親の環境で起動します。
PYTHONPATHやLD_PRELOADなどを含む親環境そのものを信頼できない場合、このラッパーだけでは起動時の影響を防げません。管理された起動元を使います。 - ラッパーは選別のために親の環境を読み取れます。コードとポリシーは、信頼できる場所で管理します。
- 絶対パスは
PATHによる取り違えを防ぎますが、実行ファイルの改変、シンボリックリンクの差し替え、ハッシュ固定までは扱いません。 - shebangが
/usr/bin/env python3のスクリプトを直接指定すると、最小化した環境では起動できない場合があります。例のように、絶対パスのPythonとスクリプトを別々の引数にします。 - 対象CLIが別のコマンドを起動する場合も、絶対パスを使うか、内容を確認した固定
PATHを渡します。PATHがない場合の検索動作を一律に仮定しません。 - カレントディレクトリ、umask、ユーザーID、リソース上限、ネットワーク、標準入出力、継承可能なファイル記述子は隔離しません。
- 秘密をコマンドライン引数へ移すと、プロセス一覧などから見える可能性があります。環境変数の削減は、秘密の安全な受け渡し方法そのものではありません。
- 実行後の記録、タイムアウト、再試行は外側の監督処理で行います。副作用のあるCLIを再試行する場合は、第174回と同様に冪等性を確認します。
今回のまとめ
親の環境を丸ごと継承せず、引き継ぐ名前と固定値を明示すると、対象CLIへ不要な設定や秘密値が渡る範囲を小さくできます。
まず確認用CLIで、必要な変数が存在し、不要な変数が存在しないことを確かめます。その後、対象CLIが実際に必要とする変数だけを追加してください。ファイルやネットワークも制限する場合は、環境変数の許可リストとは別の隔離が必要です。