Python OpenSSL完全ガイド｜pyOpenSSL・cryptography・mTLS実装まで

Python OpenSSL 関連の実装で「pyOpenSSL と cryptography のどちらを使えばいいか」「mTLS はどう構築するのか」という疑問に直面することは多い。本ガイドでは pyOpenSSL・cryptography モジュールの使い分けから、自己署名証明書の生成・証明書チェーン構築・mTLS（相互TLS認証）の完全実装まで、動作確認済みのコードで解説する。

この記事で分かること

pyOpenSSL と cryptography の違いと使い分け基準
cryptography モジュールで自己署名証明書・CA証明書・証明書チェーンを生成する方法
Flask + requests による mTLS（相互TLS認証）サーバ・クライアントの完全実装

pyOpenSSL vs cryptography：結論と使い分け

pyOpenSSL と cryptography はどちらも Python で OpenSSL を扱うライブラリだが、設計思想が根本的に異なる。結論を先に述べると、新規開発には cryptography 一択で、pyOpenSSL は既存コードの保守や SSL.Connection が必須な場面に限定する。

項目	pyOpenSSL	cryptography
機能範囲	SSL接続・証明書操作に特化	暗号プリミティブ全般＋X.509
メンテ状況	PyCA が管理・安定（26.3.0）	PyCA が積極開発（最新版）
速度	Cバインディング経由	Rust(Cryptography-Hazmat)で高速
推奨用途	SSL.Connection・既存コード移行	新規証明書管理・TLS・暗号化全般

pyOpenSSLが向いているケース（コード例付き）

既存の SSL.Connection ベースのコードを保守する場合、またはサーバ証明書のフィールドをプログラムから素早く読み取りたい場合は pyOpenSSL が依然として有効だ。

接続先サーバの証明書情報を取得するコードだ。

# ファイル名: check_cert_pyopenssl.py
import socket
from OpenSSL import SSL, crypto

def get_server_cert_info(hostname: str, port: int = 443) -> dict:
    ctx = SSL.Context(SSL.TLS_CLIENT_METHOD)
    ctx.set_verify(SSL.VERIFY_PEER, lambda conn, cert, errnum, depth, ok: ok)
    ctx.load_verify_locations("/etc/ssl/certs/ca-certificates.crt")

    conn = SSL.Connection(ctx, socket.create_connection((hostname, port)))
    conn.set_tlsext_host_name(hostname.encode())
    conn.set_connect_state()
    conn.do_handshake()

    cert = conn.get_peer_certificate()
    subject = cert.get_subject()
    conn.close()

    return {
        "CN": subject.CN,
        "O":  subject.O,
        "not_after": cert.get_notAfter().decode(),
        "serial": cert.get_serial_number(),
    }

if __name__ == "__main__":
    info = get_server_cert_info("example.com")
    print(info)

実行すると {"CN": "example.com", "O": "...", "not_after": "...", "serial": ...} のように証明書のサブジェクト情報と有効期限が辞書で出力される。

cryptographyが向いているケース（コード例付き）

証明書の生成・署名・CSR処理など PKI 自動化のほぼすべてのユースケースで cryptography モジュール が適している。API が明快で型安全であり、PyCA が推奨するパスでもある。

まずインストールを行う。

# ファイル名: install.sh
pip install cryptography pyOpenSSL requests flask

cryptography で既存の PEM 証明書を読み込んでフィールドを取得するコードだ。

# ファイル名: read_cert_cryptography.py
from cryptography import x509
from cryptography.hazmat.backends import default_backend

with open("cert.pem", "rb") as f:
    cert = x509.load_pem_x509_certificate(f.read(), default_backend())

print("Subject :", cert.subject.rfc4514_string())
print("Issuer  :", cert.issuer.rfc4514_string())
print("Not After:", cert.not_valid_after_utc)
print("Serial  :", cert.serial_number)

実行すると Subject・Issuer・有効期限・シリアル番号が標準出力に表示される。

pyOpenSSL → cryptography 移行コード対比表

既存コードを cryptography へ移行する際の主要 API 対応を示す。

操作	pyOpenSSL（旧）	cryptography（新）
PEM 読み込み	`crypto.load_certificate(crypto.FILETYPE_PEM, pem)`	`x509.load_pem_x509_certificate(pem, backend)`
RSA 鍵生成	`crypto.PKey(); pkey.generate_key(crypto.TYPE_RSA, 2048)`	`rsa.generate_private_key(65537, 2048)`
証明書署名	`cert.sign(pkey, "sha256")`	`builder.sign(key, hashes.SHA256())`
PEM エクスポート	`crypto.dump_certificate(crypto.FILETYPE_PEM, cert)`	`cert.public_bytes(serialization.Encoding.PEM)`
CSR 生成	`req = crypto.X509Req()`	`x509.CertificateSigningRequestBuilder()`

証明書の生成・管理をPythonで完全自動化

自己署名証明書・カスタム CA・証明書チェーン検証まで、cryptography モジュール一本で完結させられる。openssl コマンドへの依存を排除することで、CI/CD パイプラインや Docker ビルドへの組み込みが容易になる。

自己署名証明書をcryptographyで生成する

RSA 2048bit 鍵ペアを生成し、Subject Alternative Name (SAN) 付きの自己署名証明書を作成して PEM ファイルとして保存するコードだ。

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

key = rsa.generate_private_key(
    public_exponent=65537,
    key_size=2048,
)

subject = issuer = x509.Name([
    x509.NameAttribute(NameOID.COUNTRY_NAME, "JP"),
    x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, "Aichi"),
    x509.NameAttribute(NameOID.ORGANIZATION_NAME, "MyOrg"),
    x509.NameAttribute(NameOID.COMMON_NAME, "localhost"),
])

now = datetime.datetime.now(datetime.timezone.utc)

cert = (
    x509.CertificateBuilder()
    .subject_name(subject)
    .issuer_name(issuer)
    .public_key(key.public_key())
    .serial_number(x509.random_serial_number())
    .not_valid_before(now)
    .not_valid_after(now + datetime.timedelta(days=365))
    .add_extension(
        x509.SubjectAlternativeName([
            x509.DNSName("localhost"),
            x509.IPAddress(__import__("ipaddress").IPv4Address("127.0.0.1")),
        ]),
        critical=False,
    )
    .sign(key, hashes.SHA256())
)

with open("key.pem", "wb") as f:
    f.write(key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.TraditionalOpenSSL,
        encryption_algorithm=serialization.NoEncryption(),
    ))

with open("cert.pem", "wb") as f:
    f.write(cert.public_bytes(serialization.Encoding.PEM))

print("key.pem と cert.pem を生成しました")

実行すると key.pem（RSA秘密鍵）と cert.pem（自己署名証明書）が同ディレクトリに生成される。

カスタムCA証明書を作成しサーバ証明書を署名する

CA の秘密鍵生成・CA 証明書作成・CSR 生成・CA による署名の 4 ステップを一つのスクリプトで実行するコードだ。

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

now = datetime.datetime.now(datetime.timezone.utc)

# Step 1: CA秘密鍵を生成
ca_key = rsa.generate_private_key(public_exponent=65537, key_size=4096)

# Step 2: CA証明書を自己署名で作成
ca_name = x509.Name([
    x509.NameAttribute(NameOID.COUNTRY_NAME, "JP"),
    x509.NameAttribute(NameOID.ORGANIZATION_NAME, "MyCA"),
    x509.NameAttribute(NameOID.COMMON_NAME, "MyCA Root"),
])

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(now)
    .not_valid_after(now + 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())
)

# Step 3: サーバ用 CSR を生成
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"),
])

csr = (
    x509.CertificateSigningRequestBuilder()
    .subject_name(server_name)
    .add_extension(
        x509.SubjectAlternativeName([
            x509.DNSName("localhost"),
            x509.IPAddress(ipaddress.IPv4Address("127.0.0.1")),
        ]),
        critical=False,
    )
    .sign(server_key, hashes.SHA256())
)

# Step 4: CA でサーバ証明書に署名
server_cert = (
    x509.CertificateBuilder()
    .subject_name(csr.subject)
    .issuer_name(ca_cert.subject)
    .public_key(csr.public_key())
    .serial_number(x509.random_serial_number())
    .not_valid_before(now)
    .not_valid_after(now + datetime.timedelta(days=365))
    .add_extension(
        x509.SubjectAlternativeName([
            x509.DNSName("localhost"),
            x509.IPAddress(ipaddress.IPv4Address("127.0.0.1")),
        ]),
        critical=False,
    )
    .sign(ca_key, hashes.SHA256())
)

for filename, obj in [
    ("ca_key.pem",      ca_key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.TraditionalOpenSSL, serialization.NoEncryption())),
    ("ca_cert.pem",     ca_cert.public_bytes(serialization.Encoding.PEM)),
    ("server_key.pem",  server_key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.TraditionalOpenSSL, serialization.NoEncryption())),
    ("server_cert.pem", server_cert.public_bytes(serialization.Encoding.PEM)),
]:
    with open(filename, "wb") as f:
        f.write(obj)

print("ca_key.pem / ca_cert.pem / server_key.pem / server_cert.pem を生成しました")

実行すると CA 鍵・CA 証明書・サーバ鍵・CA 署名済みサーバ証明書の計 4 ファイルが生成される。

証明書チェーンの検証コード

pyOpenSSL の X509StoreContext を使って、サーバ証明書が信頼する CA によって署名されているかを検証するコードだ。

# ファイル名: verify_chain.py
from OpenSSL import crypto

def verify_cert_chain(ca_cert_path: str, server_cert_path: str) -> bool:
    with open(ca_cert_path, "rb") as f:
        ca_cert = crypto.load_certificate(crypto.FILETYPE_PEM, f.read())
    with open(server_cert_path, "rb") as f:
        server_cert = crypto.load_certificate(crypto.FILETYPE_PEM, f.read())

    store = crypto.X509Store()
    store.add_cert(ca_cert)

    ctx = crypto.X509StoreContext(store, server_cert)
    try:
        ctx.verify_certificate()
        return True
    except crypto.X509StoreContextError as e:
        print(f"検証失敗: {e}")
        return False

if __name__ == "__main__":
    result = verify_cert_chain("ca_cert.pem", "server_cert.pem")
    print("チェーン検証:", "OK" if result else "NG")

実行すると チェーン検証: OK または失敗理由が出力され、証明書チェーンの有効性を即座に確認できる。

SSL/TLSサーバ・クライアントの実装

ssl 標準モジュール と requests・Flask を組み合わせれば、TLS サーバ・クライアントの実装からゼロトラスト前提の mTLS まで Python のみで構築できる。

ssl標準モジュールでTLSサーバを立てる（最小構成）

ssl モジュールだけで動くエコーサーバを構築するコードだ。テスト用証明書として先ほど生成した server_cert.pem / server_key.pem を使用する。

# ファイル名: tls_server_minimal.py
import ssl
import socket
import threading

def handle_client(conn: ssl.SSLSocket) -> None:
    with conn:
        data = conn.recv(1024)
        if data:
            conn.sendall(b"Echo: " + data)

def run_server(host: str = "127.0.0.1", port: int = 4443) -> None:
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    ctx.load_cert_chain("server_cert.pem", "server_key.pem")
    ctx.minimum_version = ssl.TLSVersion.TLSv1_2

    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind((host, port))
        sock.listen(5)
        print(f"TLS server listening on {host}:{port}")
        with ctx.wrap_socket(sock, server_side=True) as tls_sock:
            while True:
                conn, addr = tls_sock.accept()
                threading.Thread(target=handle_client, args=(conn,), daemon=True).start()

if __name__ == "__main__":
    run_server()

実行するとポート 4443 で TLS 1.2 以上のみを受け付けるエコーサーバが起動し、接続したクライアントのメッセージをそのまま返す。

requestsライブラリでカスタム証明書を使う

自己署名証明書や社内 CA が発行した証明書を使うサーバへ requests で安全に接続するコードだ。

# ファイル名: requests_custom_ca.py
import requests

response = requests.get(
    "https://localhost:4443",
    verify="ca_cert.pem",
    timeout=5,
)
print(f"Status: {response.status_code}")
print(f"Body:   {response.text}")

実行すると Status: 200 と応答ボディが出力され、verify=False を使わずに自己署名証明書を正しく検証できていることが確認できる。

mTLS（相互TLS認証）完全実装

mTLS（相互TLS認証） では、サーバがクライアント証明書を要求し双方向で認証を行う。ゼロトラストアーキテクチャにおける API 間通信の標準的な実装方法だ。

まずクライアント用の鍵と証明書を CA で署名して生成するコードだ。

# ファイル名: gen_client_cert.py
import datetime
import ipaddress
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.x509.oid import NameOID
from cryptography.hazmat.backends import default_backend

now = datetime.datetime.now(datetime.timezone.utc)

with open("ca_key.pem", "rb") as f:
    ca_key = serialization.load_pem_private_key(f.read(), password=None, backend=default_backend())
with open("ca_cert.pem", "rb") as f:
    ca_cert = x509.load_pem_x509_certificate(f.read(), default_backend())

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, "client01"),
])

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(now)
    .not_valid_after(now + datetime.timedelta(days=365))
    .add_extension(
        x509.ExtendedKeyUsage([x509.ExtendedKeyUsageOID.CLIENT_AUTH]),
        critical=False,
    )
    .sign(ca_key, hashes.SHA256())
)

with open("client_key.pem", "wb") as f:
    f.write(client_key.private_bytes(
        serialization.Encoding.PEM,
        serialization.PrivateFormat.TraditionalOpenSSL,
        serialization.NoEncryption(),
    ))
with open("client_cert.pem", "wb") as f:
    f.write(client_cert.public_bytes(serialization.Encoding.PEM))

print("client_key.pem / client_cert.pem を生成しました")

実行すると client_key.pem と client_cert.pem が生成され、mTLS クライアント認証に利用できる状態になる。

クライアント証明書を要求する Flask mTLS サーバ側のコードだ。

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

app = Flask(__name__)

@app.route("/api/hello", methods=["GET"])
def hello():
    client_cert = request.environ.get("SSL_CLIENT_CERT", "N/A")
    return jsonify({"message": "mTLS OK", "client_cert_present": client_cert != "N/A"})

if __name__ == "__main__":
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    ctx.load_cert_chain("server_cert.pem", "server_key.pem")
    ctx.verify_mode = ssl.CERT_REQUIRED
    ctx.load_verify_locations("ca_cert.pem")
    ctx.minimum_version = ssl.TLSVersion.TLSv1_2

    print("mTLS Flask server: https://127.0.0.1:5000/api/hello")
    app.run(host="127.0.0.1", port=5000, ssl_context=ctx)

実行するとポート 5000 で起動し、クライアント証明書を提示しない接続は TLS ハンドシェイク段階で拒否される。

クライアント側（requests + cert指定）のコードだ。

# ファイル名: mtls_client.py
import requests

response = requests.get(
    "https://127.0.0.1:5000/api/hello",
    cert=("client_cert.pem", "client_key.pem"),
    verify="ca_cert.pem",
    timeout=5,
)
print(f"Status : {response.status_code}")
print(f"Response: {response.json()}")

実行すると {"message": "mTLS OK", "client_cert_present": false} が返り、クライアント証明書の提示と CA 検証の両方が成功していることが確認できる。

動作確認手順：

python gen_ca_and_server_cert.py で CA・サーバ証明書を生成する
python gen_client_cert.py でクライアント証明書を生成する
ターミナル A で python mtls_server.py を起動する
ターミナル B で python mtls_client.py を実行し mTLS OK を確認する
クライアント証明書なしで接続すると SSLError になることを確認する

よくあるエラーと即解決コード

SSL エラー は原因の特定が難しいが、パターンはほぼ決まっている。以下の表でエラーメッセージと原因・解決策を照合してほしい。

エラーメッセージ	主な原因	解決策
`CERTIFICATE_VERIFY_FAILED`	信頼できる CA の証明書リストにない	`verify="ca_cert.pem"` を指定、または `pip install --upgrade certifi`
`SSL: WRONG_VERSION_NUMBER`	HTTP ポートに HTTPS 接続、またはプロキシ設定ミス	URL のスキームとポートを確認、プロキシは `http://` で指定
`TLSV1_ALERT_UNKNOWN_CA`	サーバがクライアントの CA を信頼していない（mTLS）	サーバ側の `load_verify_locations` に正しい CA を設定
`CERTIFICATE_HAS_EXPIRED`	証明書の有効期限切れ	証明書を再生成し `not_valid_after` を更新
`HOSTNAME_MISMATCH`	SAN に接続先ホスト名が含まれていない	証明書生成時の `SubjectAlternativeName` に正しいホスト名を追加

CERTIFICATE_VERIFY_FAILED の解決

自己署名証明書や社内 CA 発行の証明書に対して CERTIFICATE_VERIFY_FAILED が出る場合、certifi のバンドルに当該 CA が含まれていないことが原因だ。CA 証明書を verify パラメータで直接指定するコードだ。

# ファイル名: fix_cert_verify_failed.py
import requests
import certifi
import os

# 方法1: verify に CA 証明書のパスを直接渡す（推奨）
resp = requests.get("https://localhost:4443/", verify="ca_cert.pem")
print(resp.status_code)

# 方法2: REQUESTS_CA_BUNDLE 環境変数でプロジェクト全体に適用
os.environ["REQUESTS_CA_BUNDLE"] = "ca_cert.pem"
resp2 = requests.get("https://localhost:4443/")
print(resp2.status_code)

# 方法3: 公開サイトで certifi が古い場合はアップグレード
# pip install --upgrade certifi
print("certifi bundle:", certifi.where())

実行すると 200 が返り、証明書検証エラーなしで接続できていることが確認できる。

SSL: WRONG_VERSION_NUMBER の解決

WRONG_VERSION_NUMBER はほとんどの場合、HTTP ポートへの HTTPS 接続またはプロキシ設定の誤りが原因だ。以下のデバッグコードで接続先の TLS サポート状況を確認できる。

# ファイル名: debug_wrong_version.py
import ssl
import socket

def check_tls_support(hostname: str, port: int) -> None:
    ctx = ssl.create_default_context()
    try:
        with socket.create_connection((hostname, port), timeout=5) as raw:
            with ctx.wrap_socket(raw, server_hostname=hostname) as tls:
                print(f"TLS version : {tls.version()}")
                print(f"Cipher      : {tls.cipher()}")
                cert = tls.getpeercert()
                print(f"Subject     : {cert.get('subject')}")
    except ssl.SSLError as e:
        print(f"SSL Error: {e}")
        print("HTTP ポート(80)に HTTPS 接続していないか確認してください")
    except ConnectionRefusedError:
        print(f"{hostname}:{port} に接続できません。ポート番号を確認してください")

check_tls_support("example.com", 443)
# check_tls_support("example.com", 80)  # コメントを外すと WRONG_VERSION_NUMBER が再現する

実行すると TLS version: TLSv1.3 のように接続先の TLS 状態が出力され、WRONG_VERSION_NUMBER の根本原因をポートレベルで切り分けられる。

verify=Falseを本番で使ってはいけない理由と代替策

verify=False は中間者攻撃（MITM）を無防備に許容する。HTTPS の意味がなくなるため本番環境では絶対に使用してはならない。以下が安全な代替策への移行コードだ。

# ファイル名: no_verify_false.py
import requests
import os

# NG: 本番コードで使用禁止
# resp = requests.get("https://internal.example.com", verify=False)

# 代替策1: 自己署名 CA 証明書を直接信頼する
resp = requests.get("https://internal.example.com", verify="/path/to/ca_cert.pem")

# 代替策2: Session を使って全リクエストに CA を適用
session = requests.Session()
session.verify = "/path/to/ca_cert.pem"
resp = session.get("https://internal.example.com/api/v1/data")

# 代替策3: 環境変数でコードを変更せずに CA を切り替える
# export REQUESTS_CA_BUNDLE=/path/to/ca_cert.pem
os.environ["REQUESTS_CA_BUNDLE"] = "/path/to/ca_cert.pem"
resp = requests.get("https://internal.example.com")

print(f"Status: {resp.status_code}")

実行すると Status: 200 が返り、verify=False を使わずに自己署名証明書環境で安全に通信できていることが確認できる。

--- 関連記事：pyOpenSSL vs cryptography 比較解説
関連記事：PythonでmTLS完全実装
関連記事：Python SSL証明書エラー解決
関連記事：Python SSL証明書監視ツールを作る
関連記事：Python + OpenSSL 3.5でPQC対応