第179回 承認ハッシュを署名付きマニフェストで渡す:Pythonで検証するEd25519

第178回では、検査したELF実行ファイルをmemfdへコピーし、内容を固定してから起動しました。ただし、照合に使うSHA-256が正しい承認値かは別の問題です。

成果物と期待ハッシュを同じ権限で変更できれば、両方を差し替えられます。

今回は、成果物ID、リリースID、SHA-256をJSONマニフェストへ記録し、承認環境のEd25519秘密鍵で署名します。実行環境には公開鍵だけを置き、署名、マニフェストの形式、期待するリリース、成果物の実際のハッシュを順に検証します。

署名が示すのは、対応する秘密鍵によってマニフェストのバイト列が署名されたことです。成果物の安全性や署名者の業務上の権限までは証明しません。公開鍵を信頼できる方法で配置し、秘密鍵を実行環境へ置かないことが前提です。

署名と封印の役割

仕組み 確認すること 残る問題
SHA-256照合 読み取った成果物と期待するバイト列の一致 期待ハッシュの入手元
Ed25519署名 マニフェストが対応する秘密鍵で署名され、署名後に変わっていないこと 秘密鍵の漏えい、公開鍵の置換、署名権限の管理
リリースID照合 今回指定したリリースとの一致 期待値を渡す設定の改変
第178回の封印済みmemfd 検査後から起動までの主実行ファイルの固定 共有ライブラリ、設定、カーネルなどの固定

今回のCLIは検証までを担当し、成果物を起動しません。検証後に別のプロセスで成果物を開き直すなら、再度の照合が必要です。

実行条件

項目 条件
承認環境 Linux、Python 3.10以上、OpenSSL 3系
検証環境 Linux、Python 3.10以上、cryptography 50.0.2
マニフェスト 本稿で定義するASCII JSON。末尾改行を含め4096バイト以下
署名 Ed25519の64バイト署名
公開鍵 PEM形式、4096バイト以下。Ed25519以外は拒否
成果物 通常ファイル。最終パス要素のシンボリックリンクは拒否
成果物の上限 既定値268,435,456バイト。指定できる最大値は1,073,741,824バイト
成功 標準出力へ1行JSON、終了コード0
失敗 引数解析エラーを除き、標準エラー出力へ1行JSON

1.検証環境を用意する

2026年10月28日時点で公開されているcryptography 50.0.2へ固定します。実運用では、組織の手順に従って配布ファイルのハッシュも固定してください。

python3 -m venv .venv
.venv/bin/python -m pip install 'cryptography==50.0.2'
.venv/bin/python -c 'import cryptography; print(cryptography.__version__)'

Ed25519の検証APIは、署名が一致しないとInvalidSignatureを送出します。PEMには別の種類の公開鍵も格納できるため、読み込んだ鍵の型も確認します。

2.マニフェストを作成・検証するCLI

次のコードをmanifest_gate.pyとして保存します。

buildは成果物をハッシュ化し、新しいマニフェストを排他的に作成します。既存ファイルは上書きしません。verifyは署名を検証した後でJSONを解釈し、項目、型、正規形式、期待するID、成果物のSHA-256を確認します。

Pythonのjson.loads()は、通常は重複した名前の最後の値を残します。解釈を一つにするため、重複キーを拒否します。

import argparse
import hashlib
import hmac
import json
import os
import re
import stat
import sys

MANIFEST_LIMIT = 4096
PUBLIC_KEY_LIMIT = 4096
SIGNATURE_SIZE = 64
MAX_ARTIFACT_LIMIT = 1024 * 1024 * 1024
DEFAULT_ARTIFACT_LIMIT = 256 * 1024 * 1024
ID_PATTERN = re.compile(r'[a-z0-9][a-z0-9._-]{0,63}')
RELEASE_PATTERN = re.compile(r'[A-Za-z0-9][A-Za-z0-9._-]{0,63}')
SHA256_PATTERN = re.compile(r'[0-9a-f]{64}')
MANIFEST_KEYS = {'schema_version', 'artifact_id', 'release_id', 'sha256'}


class GateError(Exception):
    def __init__(self, code, exit_code=3):
        super().__init__(code)
        self.code = code
        self.exit_code = exit_code


def emit(payload, stream=None):
    if stream is None:
        stream = sys.stdout
    print(
        json.dumps(
            payload,
            ensure_ascii=True,
            sort_keys=True,
            separators=(',', ':'),
        ),
        file=stream,
    )


def same_snapshot(left, right):
    return (
        left.st_dev,
        left.st_ino,
        left.st_size,
        left.st_mtime_ns,
        left.st_ctime_ns,
    ) == (
        right.st_dev,
        right.st_ino,
        right.st_size,
        right.st_mtime_ns,
        right.st_ctime_ns,
    )


def open_regular(path):
    flags = os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW
    fd = os.open(path, flags)
    try:
        info = os.fstat(fd)
    except BaseException:
        os.close(fd)
        raise
    if not stat.S_ISREG(info.st_mode):
        os.close(fd)
        raise GateError('not_regular_file')
    return fd, info


def read_regular_bounded(path, limit):
    fd, before = open_regular(path)
    try:
        if before.st_size > limit:
            raise GateError('input_too_large')
        handle = os.fdopen(fd, 'rb', closefd=True)
        fd = -1
        with handle:
            data = handle.read(limit + 1)
            after = os.fstat(handle.fileno())
    finally:
        if fd >= 0:
            os.close(fd)
    if len(data) > limit:
        raise GateError('input_too_large')
    if not same_snapshot(before, after):
        raise GateError('input_changed_during_read')
    return data


def hash_regular(path, limit):
    fd, before = open_regular(path)
    digest = hashlib.sha256()
    total = 0
    try:
        if before.st_size > limit:
            raise GateError('artifact_too_large')
        handle = os.fdopen(fd, 'rb', closefd=True)
        fd = -1
        with handle:
            while True:
                chunk = handle.read(1024 * 1024)
                if not chunk:
                    break
                total += len(chunk)
                if total > limit:
                    raise GateError('artifact_too_large')
                digest.update(chunk)
            after = os.fstat(handle.fileno())
    finally:
        if fd >= 0:
            os.close(fd)
    if total != after.st_size or not same_snapshot(before, after):
        raise GateError('artifact_changed_during_read')
    return digest.hexdigest(), total


def validate_artifact_id(value):
    if not isinstance(value, str) or ID_PATTERN.fullmatch(value) is None:
        raise GateError('invalid_artifact_id')


def validate_release_id(value):
    if not isinstance(value, str) or RELEASE_PATTERN.fullmatch(value) is None:
        raise GateError('invalid_release_id')


def canonical_bytes(document):
    text = json.dumps(
        document,
        ensure_ascii=True,
        sort_keys=True,
        separators=(',', ':'),
    )
    return (text + '\n').encode('ascii')


def reject_duplicate_keys(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise GateError('duplicate_json_key')
        result[key] = value
    return result


def reject_constant(_value):
    raise GateError('invalid_json_number')


def parse_manifest(raw):
    if raw.startswith(b'\xef\xbb\xbf'):
        raise GateError('manifest_has_bom')
    try:
        text = raw.decode('ascii')
        document = json.loads(
            text,
            object_pairs_hook=reject_duplicate_keys,
            parse_constant=reject_constant,
        )
    except UnicodeDecodeError as exc:
        raise GateError('manifest_not_ascii') from exc
    except json.JSONDecodeError as exc:
        raise GateError('manifest_invalid_json') from exc
    if not isinstance(document, dict) or set(document) != MANIFEST_KEYS:
        raise GateError('manifest_schema_mismatch')
    if type(document['schema_version']) is not int:
        raise GateError('manifest_schema_version_mismatch')
    if document['schema_version'] != 1:
        raise GateError('manifest_schema_version_mismatch')
    validate_artifact_id(document['artifact_id'])
    validate_release_id(document['release_id'])
    digest = document['sha256']
    if not isinstance(digest, str) or SHA256_PATTERN.fullmatch(digest) is None:
        raise GateError('invalid_manifest_sha256')
    if raw != canonical_bytes(document):
        raise GateError('manifest_not_canonical')
    return document


def write_new_file(path, data):
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC
    fd = os.open(path, flags, 0o644)
    identity = None
    try:
        identity = os.fstat(fd)
        offset = 0
        while offset < len(data):
            written = os.write(fd, data[offset:])
            if written == 0:
                raise OSError('zero-byte write')
            offset += written
        os.fsync(fd)
    except BaseException:
        try:
            os.close(fd)
        except OSError:
            pass
        if identity is not None:
            try:
                current = os.stat(path, follow_symlinks=False)
                if (
                    current.st_dev == identity.st_dev
                    and current.st_ino == identity.st_ino
                ):
                    os.unlink(path)
            except OSError:
                pass
        raise
    else:
        os.close(fd)


def parse_size(value):
    try:
        number = int(value, 10)
    except ValueError as exc:
        raise argparse.ArgumentTypeError('整数で指定してください') from exc
    if not 1 <= number <= MAX_ARTIFACT_LIMIT:
        raise argparse.ArgumentTypeError(
            f'1から{MAX_ARTIFACT_LIMIT}までで指定してください'
        )
    return number


def command_build(args):
    validate_artifact_id(args.artifact_id)
    validate_release_id(args.release_id)
    digest, size = hash_regular(args.artifact, args.max_bytes)
    document = {
        'schema_version': 1,
        'artifact_id': args.artifact_id,
        'release_id': args.release_id,
        'sha256': digest,
    }
    write_new_file(args.output, canonical_bytes(document))
    emit(
        {
            'artifact_id': args.artifact_id,
            'release_id': args.release_id,
            'sha256': digest,
            'size': size,
            'status': 'manifest_created',
        }
    )
    return 0


def command_verify(args):
    try:
        from cryptography.exceptions import InvalidSignature, UnsupportedAlgorithm
        from cryptography.hazmat.primitives import serialization
        from cryptography.hazmat.primitives.asymmetric.ed25519 import (
            Ed25519PublicKey,
        )
    except ImportError as exc:
        raise GateError('cryptography_not_available') from exc

    validate_artifact_id(args.expected_artifact_id)
    validate_release_id(args.expected_release_id)
    manifest_raw = read_regular_bounded(args.manifest, MANIFEST_LIMIT)
    signature = read_regular_bounded(args.signature, SIGNATURE_SIZE)
    if len(signature) != SIGNATURE_SIZE:
        raise GateError('invalid_signature_size')
    public_pem = read_regular_bounded(args.public_key, PUBLIC_KEY_LIMIT)
    try:
        public_key = serialization.load_pem_public_key(public_pem)
    except (ValueError, UnsupportedAlgorithm) as exc:
        raise GateError('invalid_public_key') from exc
    if not isinstance(public_key, Ed25519PublicKey):
        raise GateError('public_key_is_not_ed25519')
    try:
        public_key.verify(signature, manifest_raw)
    except InvalidSignature as exc:
        raise GateError('invalid_signature', 4) from exc

    document = parse_manifest(manifest_raw)
    if not hmac.compare_digest(
        document['artifact_id'], args.expected_artifact_id
    ):
        raise GateError('artifact_id_mismatch', 5)
    if not hmac.compare_digest(
        document['release_id'], args.expected_release_id
    ):
        raise GateError('release_id_mismatch', 5)
    actual_digest, size = hash_regular(args.artifact, args.max_bytes)
    if not hmac.compare_digest(document['sha256'], actual_digest):
        raise GateError('artifact_digest_mismatch', 5)

    public_der = public_key.public_bytes(
        encoding=serialization.Encoding.DER,
        format=serialization.PublicFormat.SubjectPublicKeyInfo,
    )
    emit(
        {
            'approved': True,
            'artifact_id': document['artifact_id'],
            'public_key_sha256': hashlib.sha256(public_der).hexdigest(),
            'release_id': document['release_id'],
            'sha256': actual_digest,
            'size': size,
        }
    )
    return 0


def build_parser():
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest='command', required=True)

    build = subparsers.add_parser('build')
    build.add_argument('--artifact', required=True)
    build.add_argument('--artifact-id', required=True)
    build.add_argument('--release-id', required=True)
    build.add_argument('--output', required=True)
    build.add_argument(
        '--max-bytes',
        type=parse_size,
        default=DEFAULT_ARTIFACT_LIMIT,
    )
    build.set_defaults(handler=command_build)

    verify = subparsers.add_parser('verify')
    verify.add_argument('--artifact', required=True)
    verify.add_argument('--manifest', required=True)
    verify.add_argument('--signature', required=True)
    verify.add_argument('--public-key', required=True)
    verify.add_argument('--expected-artifact-id', required=True)
    verify.add_argument('--expected-release-id', required=True)
    verify.add_argument(
        '--max-bytes',
        type=parse_size,
        default=DEFAULT_ARTIFACT_LIMIT,
    )
    verify.set_defaults(handler=command_verify)
    return parser


def main():
    parser = build_parser()
    args = parser.parse_args()
    try:
        return args.handler(args)
    except GateError as exc:
        emit({'error': exc.code, 'ok': False}, sys.stderr)
        return exc.exit_code
    except OSError as exc:
        emit(
            {
                'detail': exc.strerror or type(exc).__name__,
                'error': 'os_error',
                'ok': False,
            },
            sys.stderr,
        )
        return 3
    except KeyboardInterrupt:
        emit({'error': 'interrupted', 'ok': False}, sys.stderr)
        return 130


if __name__ == '__main__':
    raise SystemExit(main())

3.承認環境で鍵と署名を作る

鍵生成は初回だけ行います。既存鍵を誤って上書きしないよう、新しいディレクトリの作成に成功した場合だけ生成します。秘密鍵はパスフレーズで暗号化し、実行環境へコピーしません。

mkdir -- approval-keys && (
  umask 077
  openssl genpkey -algorithm Ed25519 -aes-256-cbc \
    -out approval-keys/approval-private.pem &&
  openssl pkey -in approval-keys/approval-private.pem -pubout \
    -out approval-keys/approval-public.pem
)

動作確認用にLinuxの/bin/trueをコピーし、マニフェストを作ります。既存のapproved-toolがある場合は処理を止めます。

test ! -e ./approved-tool &&
cp -- /bin/true ./approved-tool &&
.venv/bin/python manifest_gate.py build \
  --artifact ./approved-tool \
  --artifact-id batch-worker \
  --release-id 2026-10-28.1 \
  --output manifest.json

artifact_idは小文字英数字で始まる1~64文字、release_idは英数字で始まる1~64文字です。どちらも、2文字目以降にはピリオド、アンダースコア、ハイフンを使えます。

manifest.jsonを編集せず、そのバイト列へ署名します。OpenSSL 3.0のpkeyutl資料では、Ed25519に-rawinが必要です。パスフレーズは対話入力し、引数へ書きません。

test ! -e manifest.sig &&
openssl pkeyutl -sign -rawin \
  -inkey approval-keys/approval-private.pem \
  -in manifest.json \
  -out manifest.sig

署名処理が失敗すると、不完全なmanifest.sigが残る場合があります。そのファイルを署名として使用してはいけません。原因を確認し、別の新しい出力名でやり直します。

実行環境へ渡すのは、成果物、マニフェスト、署名、公開鍵です。秘密鍵とパスフレーズは渡しません。

4.実行環境で検証する

.venv/bin/python manifest_gate.py verify \
  --artifact ./approved-tool \
  --manifest ./manifest.json \
  --signature ./manifest.sig \
  --public-key ./approval-keys/approval-public.pem \
  --expected-artifact-id batch-worker \
  --expected-release-id 2026-10-28.1

成功時は、成果物ID、リリースID、成果物のSHA-256、サイズ、公開鍵のフィンガープリントを1行JSONで返します。public_key_sha256は、PEMファイルのバイト列ではなく、読み込んだEd25519公開鍵をDER SubjectPublicKeyInfo形式へ変換したバイト列のSHA-256です。そのため、sha256sum approval-public.pemで得る値とは一致しません。この値は使用した鍵の記録であり、公開鍵を信頼できる経路で配置する代わりにはなりません。

検査の順序は次のとおりです。

  1. マニフェスト、署名、公開鍵を上限付きで読み取る
  2. 公開鍵がEd25519であることを確認する
  3. マニフェストの生バイト列に対する署名を検証する
  4. JSONの項目、型、重複キー、正規形式を検査する
  5. 成果物IDとリリースIDをコマンド側の期待値と照合する
  6. 成果物を一つのファイル記述子から読み、SHA-256を照合する

署名検証を先に行うため、未署名または改変済みのマニフェストを業務上の指示として扱いません。

5.改変時の停止を確認する

成果物のコピーへ1バイト追加して検証します。

test ! -e changed-tool &&
cp -- approved-tool changed-tool &&
printf 'x' >> changed-tool &&
.venv/bin/python manifest_gate.py verify \
  --artifact ./changed-tool \
  --manifest ./manifest.json \
  --signature ./manifest.sig \
  --public-key ./approval-keys/approval-public.pem \
  --expected-artifact-id batch-worker \
  --expected-release-id 2026-10-28.1
echo $?

標準エラー出力は{"error":"artifact_digest_mismatch","ok":false}となり、終了コードは5です。

署名の改変は、長さを変えずに1ビットだけ反転して確認します。

test ! -e changed.sig &&
cp -- manifest.sig changed.sig &&
.venv/bin/python -c "from pathlib import Path; p = Path('changed.sig'); data = bytearray(p.read_bytes()); data[0] ^= 1; p.write_bytes(data)"

verifyの--signatureをchanged.sigへ変えると、invalid_signatureで終了コード4になります。

終了コード

終了コード 意味
0 マニフェスト作成、またはすべての検証に成功
2 argparseによる引数エラー
3 入出力、形式、サイズ、鍵種別などの検査失敗
4 Ed25519署名が一致しない
5 成果物ID、リリースID、または成果物のSHA-256が一致しない
130 処理中に割り込まれた

この署名だけでは解決しないこと

  • 公開鍵を置き換えられる利用者は、自分の鍵、マニフェスト、署名を一式配置できます。公開鍵と検証スクリプトは、成果物とは別の権限で保護します。
  • 秘密鍵が漏えいすれば、不正なマニフェストにも有効な署名を作れます。保管、利用記録、失効、ローテーションは別に設計します。
  • 古い正規版の署名も有効です。信頼できる設定から--expected-release-idを渡します。
  • O_NOFOLLOWが拒否するのは最終パス要素です。親ディレクトリとその権限も保護します。
  • 読み取り前後のファイル情報を比較していますが、ファイルをロックする仕組みではありません。
  • 署名は、成果物が安全、正しい、無害であることを判定しません。コードレビューと動作試験は別に行います。
  • 共有ライブラリ、設定ファイル、ネットワーク先などは今回のマニフェストに含めていません。

次の一歩

まず、公開鍵を成果物とは別の経路で実行環境へ配置し、所有者と書き込み権限を確認します。次に、秘密鍵のない実行環境で正常系と改変時の停止を試します。

ELFを起動する場合は、検証済みのSHA-256を第178回の処理へ渡します。memfdへのコピーを封印した後、同じ値を再照合します。これで、署名された承認値と実際に起動する固定済みのバイト列を接続できます。