PythonでmTLS（相互TLS認証）を実装する方法｜自作CAから証明書発行まで

最近、ゼロトラストアーキテクチャの標準手法として「Python mTLS 実装」が注目されている。cryptographyライブラリを使えば、自作CAから証明書発行・Flask認証まで純PythonだけでmTLS環境が構築できる。

今回は、ローカル開発環境（Python 3.11 + Flask）で「相互TLS認証（mTLS）」のフルスタック実装を行った。コードはコピーしてそのまま動く構成だ。

この記事で分かること

通常TLSとmTLSの構造的な違い（ASCIIアート図解付き）
Pythonのcryptographyライブラリで自作CA・サーバ証明書・クライアント証明書を発行する方法
FlaskサーバでmTLS認証を有効にし、requestsクライアントから接続確認する方法

mTLSとは（通常TLSとの違い）

mTLS（Mutual TLS）は、クライアントとサーバがお互いの証明書を検証する認証方式だ。通常のHTTPS（TLS）はサーバ証明書だけをクライアントが確認するが、mTLSではサーバ側もクライアント証明書を要求・検証する。

【通常 TLS（片方向認証）】

  Client ──────────────────────────────► Server
          ① ClientHello
                 ◄──────────────────────
          ② ServerHello + サーバ証明書
          ③ クライアントがサーバ証明書を検証
          ④ 暗号化通信開始
  ※ サーバはクライアントを「誰でもOK」として扱う

【mTLS（双方向認証）】

  Client ◄────────────────────────────► Server
          ① ClientHello
                 ◄──────────────────────
          ② ServerHello + サーバ証明書 + CertificateRequest
          ③ クライアントがサーバ証明書を検証
          ④ クライアント証明書を送信
                 ──────────────────────►
          ⑤ サーバがクライアント証明書を検証
          ⑥ 相互認証完了 → 暗号化通信開始
  ※ 証明書を持たないクライアントは接続を拒否される

mTLSを使う主な場面は以下の通りだ。

マイクロサービス間通信：Kubernetes内のサービスメッシュ（Istio等）で標準採用
内部API認証：APIキーより強固なクライアント識別が必要な場合
ゼロトラストネットワーク：「ネットワーク内にいる＝信頼」を廃止し、接続ごとに認証する設計
IoTデバイス管理：デバイスに埋め込んだ証明書でサーバへの接続を制御

通常TLSとmTLSの主な違いをまとめると以下の通りだ。

項目	通常TLS	mTLS（相互TLS）
サーバ証明書の検証	クライアントが行う	クライアントが行う
クライアント証明書の検証	なし	サーバが行う
認証失敗時の挙動	なし（全員接続可）	TLSハンドシェイク段階で切断
主な用途	一般公開Webサイト	マイクロサービス・内部API
証明書の管理コスト	サーバ側のみ	サーバ＋クライアント両方

全体のファイル構成と実装ステップ

今回作成するファイルは以下の通りだ。すべて同じディレクトリ（mtls_demo/）に配置する。

mtls_demo/
├── certs/
│   ├── ca.key          # CA秘密鍵（手順①で生成）
│   ├── ca.crt          # CA証明書（手順①で生成）
│   ├── server.key      # サーバ秘密鍵（手順②で生成）
│   ├── server.crt      # サーバ証明書（手順②で生成）
│   ├── client.key      # クライアント秘密鍵（手順③で生成）
│   └── client.crt      # クライアント証明書（手順③で生成）
├── 01_create_ca.py
├── 02_create_server_cert.py
├── 03_create_client_cert.py
├── 04_flask_server.py
└── 05_client_request.py

事前にcryptographyとflaskをインストールしておく。

pip install cryptography flask

手順①〜⑤：証明書発行からFlask・クライアント実装まで

手順①：カスタムCA（認証局）の作成

すべての証明書の信頼の起点となるCAを作成する。実行後、certs/ca.keyとcerts/ca.crtの2ファイルが生成される。

# ファイル名: 01_create_ca.py
import os
import datetime
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa

os.makedirs("certs", exist_ok=True)

# CA秘密鍵を生成（4096ビットRSA）
ca_key = rsa.generate_private_key(public_exponent=65537, key_size=4096)

# CA証明書のSubject/Issuer情報（自己署名なので同一）
ca_name = x509.Name([
    x509.NameAttribute(NameOID.COUNTRY_NAME, "JP"),
    x509.NameAttribute(NameOID.ORGANIZATION_NAME, "MyOrg CA"),
    x509.NameAttribute(NameOID.COMMON_NAME, "MyOrg Root CA"),
])

# CA証明書を構築して自己署名
ca_cert = (
    x509.CertificateBuilder()
    .subject_name(ca_name)
    .issuer_name(ca_name)
    .public_key(ca_key.public_key())
    .serial_number(x509.random_serial_number())
    .not_valid_before(datetime.datetime.utcnow())
    .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=3650))
    .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
    .add_extension(
        x509.KeyUsage(
            digital_signature=True, key_cert_sign=True, crl_sign=True,
            content_commitment=False, key_encipherment=False,
            data_encipherment=False, key_agreement=False,
            encipher_only=False, decipher_only=False,
        ),
        critical=True,
    )
    .sign(ca_key, hashes.SHA256())
)

# ca.key を保存（本番環境では必ず暗号化すること）
with open("certs/ca.key", "wb") as f:
    f.write(ca_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.TraditionalOpenSSL,
        encryption_algorithm=serialization.NoEncryption(),
    ))

# ca.crt を保存
with open("certs/ca.crt", "wb") as f:
    f.write(ca_cert.public_bytes(serialization.Encoding.PEM))

print("[OK] certs/ca.key と certs/ca.crt を生成しました")

手順②：サーバ証明書の発行

SANs（SubjectAlternativeName）にlocalhostを設定しないと、Pythonのsslモジュールがホスト検証で失敗するため必須だ。実行後、certs/server.keyとcerts/server.crtが生成される。

# ファイル名: 02_create_server_cert.py
import datetime
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa

# 手順①で作ったCA証明書と秘密鍵を読み込む
with open("certs/ca.key", "rb") as f:
    ca_key = serialization.load_pem_private_key(f.read(), password=None)
with open("certs/ca.crt", "rb") as f:
    ca_cert = x509.load_pem_x509_certificate(f.read())

# サーバ秘密鍵を生成
server_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)

server_name = x509.Name([
    x509.NameAttribute(NameOID.COUNTRY_NAME, "JP"),
    x509.NameAttribute(NameOID.ORGANIZATION_NAME, "MyOrg"),
    x509.NameAttribute(NameOID.COMMON_NAME, "localhost"),
])

# サーバ証明書をCAで署名
server_cert = (
    x509.CertificateBuilder()
    .subject_name(server_name)
    .issuer_name(ca_cert.subject)
    .public_key(server_key.public_key())
    .serial_number(x509.random_serial_number())
    .not_valid_before(datetime.datetime.utcnow())
    .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=365))
    .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
    .add_extension(
        x509.SubjectAlternativeName([x509.DNSName("localhost")]),
        critical=False,
    )
    .sign(ca_key, hashes.SHA256())
)

with open("certs/server.key", "wb") as f:
    f.write(server_key.private_bytes(
        serialization.Encoding.PEM,
        serialization.PrivateFormat.TraditionalOpenSSL,
        serialization.NoEncryption(),
    ))

with open("certs/server.crt", "wb") as f:
    f.write(server_cert.public_bytes(serialization.Encoding.PEM))

print("[OK] certs/server.key と certs/server.crt を生成しました")

手順③：クライアント証明書の発行

ExtendedKeyUsageにCLIENT_AUTHを設定することで、サーバ側がTLSハンドシェイク時にこの証明書をクライアント認証用として正しく識別する。実行後、certs/client.keyとcerts/client.crtが生成される。

# ファイル名: 03_create_client_cert.py
import datetime
from cryptography import x509
from cryptography.x509.oid import NameOID, ExtendedKeyUsageOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa

# 手順①のCA証明書と秘密鍵を読み込む
with open("certs/ca.key", "rb") as f:
    ca_key = serialization.load_pem_private_key(f.read(), password=None)
with open("certs/ca.crt", "rb") as f:
    ca_cert = x509.load_pem_x509_certificate(f.read())

# クライアント秘密鍵を生成
client_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)

client_name = x509.Name([
    x509.NameAttribute(NameOID.COUNTRY_NAME, "JP"),
    x509.NameAttribute(NameOID.ORGANIZATION_NAME, "MyOrg"),
    x509.NameAttribute(NameOID.COMMON_NAME, "mtls-client-01"),
])

# クライアント証明書をCAで署名
client_cert = (
    x509.CertificateBuilder()
    .subject_name(client_name)
    .issuer_name(ca_cert.subject)
    .public_key(client_key.public_key())
    .serial_number(x509.random_serial_number())
    .not_valid_before(datetime.datetime.utcnow())
    .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=365))
    .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
    .add_extension(
        x509.ExtendedKeyUsage([ExtendedKeyUsageOID.CLIENT_AUTH]),
        critical=False,
    )
    .sign(ca_key, hashes.SHA256())
)

with open("certs/client.key", "wb") as f:
    f.write(client_key.private_bytes(
        serialization.Encoding.PEM,
        serialization.PrivateFormat.TraditionalOpenSSL,
        serialization.NoEncryption(),
    ))

with open("certs/client.crt", "wb") as f:
    f.write(client_cert.public_bytes(serialization.Encoding.PEM))

print("[OK] certs/client.key と certs/client.crt を生成しました")

手順④：FlaskサーバでmTLS認証を実装する

ssl.SSLContextのverify_modeをssl.CERT_REQUIREDに設定し、load_verify_locationsに手順①のca.crtを渡す。これにより「このCAが署名した証明書だけを信頼する」ルールが設定される。証明書なしで接続してきた場合は、TLSハンドシェイク段階でSSLエラーが発生して接続が切られる。

# ファイル名: 04_flask_server.py
import ssl
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/api/hello", methods=["GET"])
def hello():
    # TLSハンドシェイクを通過した時点でクライアント証明書は検証済み
    client_dn = request.environ.get("SSL_CLIENT_S_DN", "不明")
    return jsonify({
        "status": "ok",
        "message": "mTLS認証成功",
        "client_dn": client_dn,
    })

@app.errorhandler(403)
def forbidden(e):
    return jsonify({"status": "error", "message": "クライアント証明書が無効です"}), 403

if __name__ == "__main__":
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)

    # サーバ証明書と秘密鍵を設定
    context.load_cert_chain(certfile="certs/server.crt", keyfile="certs/server.key")

    # クライアント証明書の検証を必須にする
    context.verify_mode = ssl.CERT_REQUIRED

    # 信頼するCAとして手順①の ca.crt を指定
    context.load_verify_locations("certs/ca.crt")

    print("[起動] https://localhost:5000 でmTLSサーバを開始します")
    app.run(host="0.0.0.0", port=5000, ssl_context=context, debug=False)

手順⑤：requestsクライアントで接続確認

cert=("certs/client.crt", "certs/client.key")でクライアント証明書ペアを渡し、verify="certs/ca.crt"でサーバ証明書の検証先CAを指定する。証明書なしでアクセスした場合はrequests.exceptions.SSLErrorがraiseされる。

# ファイル名: 05_client_request.py
import requests

BASE_URL = "https://localhost:5000"
CA_CERT  = "certs/ca.crt"
CLI_CERT = ("certs/client.crt", "certs/client.key")

print("=" * 50)
print("【テスト1】クライアント証明書あり → 成功するはず")
print("=" * 50)
try:
    resp = requests.get(
        f"{BASE_URL}/api/hello",
        cert=CLI_CERT,
        verify=CA_CERT,
    )
    print(f"ステータス: {resp.status_code}")
    print(f"レスポンス: {resp.json()}")
except requests.exceptions.SSLError as e:
    print(f"SSLエラー: {e}")

print()
print("=" * 50)
print("【テスト2】クライアント証明書なし → SSLエラーになるはず")
print("=" * 50)
try:
    resp = requests.get(
        f"{BASE_URL}/api/hello",
        verify=CA_CERT,
    )
    print(f"ステータス: {resp.status_code}")
except requests.exceptions.SSLError as e:
    print(f"[期待通り] SSLエラー発生: {type(e).__name__}")
    print(f"詳細: {str(e)[:120]}...")

テスト1とテスト2の期待出力は以下の通りだ。

==================================================
【テスト1】クライアント証明書あり → 成功するはず
==================================================
ステータス: 200
レスポンス: {'status': 'ok', 'message': 'mTLS認証成功', 'client_dn': '...'}

==================================================
【テスト2】クライアント証明書なし → SSLエラーになるはず
==================================================
[期待通り] SSLError発生: SSLError
詳細: HTTPSConnectionPool(host='localhost', port=5000): ...CERTIFICATE_REQUIRED...

本番環境での注意点

本番環境でmTLSを運用する場合、証明書の有効期限管理と失効処理が最重要だ。以下のスクリプトで証明書の残り日数を確認できる。

# ファイル名: check_cert_expiry.py
import datetime
from cryptography import x509

with open("certs/client.crt", "rb") as f:
    cert = x509.load_pem_x509_certificate(f.read())

expiry = cert.not_valid_after_utc
days_left = (expiry - datetime.datetime.now(datetime.timezone.utc)).days

print(f"証明書CN     : {cert.subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME)[0].value}")
print(f"有効期限     : {expiry.strftime('%Y-%m-%d %H:%M:%S UTC')}")
print(f"残り日数     : {days_left} 日")

if days_left < 30:
    print("[警告] 証明書の有効期限が30日以内です。ローテーションを実施してください。")

その他、本番運用で押さえるべき主な注意点は以下だ。

CRL（証明書失効リスト）とOCSP：証明書が漏洩・不正利用された場合に即時無効化するために、CAがCRLまたはOCSPレスポンダを公開する仕組みを用意する。Pythonのcryptographyライブラリでx509.CRLDistributionPoints拡張を証明書に埋め込むことで、クライアントが失効確認を自動的に行えるようになる
CAの秘密鍵保護：ca.keyはすべての証明書の信頼の根拠になるため、HSMまたはHashiCorp Vault等で厳重に管理する。本番CAの秘密鍵をディスクにプレーンテキストで置くのは絶対に避ける
証明書の有効期間設定：マイクロサービス向けのクライアント証明書は有効期間を短く（90日以内）設定し、自動ローテーションスクリプトをCI/CDに組み込むのが標準的なアプローチだ
本番FlaskはGunicorn+nginx構成：本番環境ではapp.run()の代わりに、GunicornをWSGIサーバとして使い、nginx側でTLS終端を行う構成が推奨される。nginxのssl_verify_client onとssl_client_certificateディレクティブでmTLSを設定できる

関連記事：Python OpenSSL完全ガイド

関連記事：pyOpenSSL vs cryptography ライブラリ比較