このページの内容
gNOI証明書管理サービス
gNOI CertificateManagement サービスを使用して、ターゲットネットワーク要素上の証明書を管理します。
概要
gnoi.certificateパッケージ内のgNOICertificateManagementサービスは、ターゲットネットワーク要素での証明書管理を処理します。プロト定義ファイルは https://github.com/openconfig/gnoi/blob/master/cert/cert.proto にあります。
公開鍵基盤(PKI)は、公開暗号化鍵の配布と識別をサポートします。これにより、ユーザーはインターネットなどのネットワーク上で安全にデータを交換したり、相手の身元を確認することができます。Junos PKIを使用すると、証明書のダウンロード、生成、検証など、Junosデバイス上の公開キー証明書を管理できます。gNOI CertificateManagement サービスは、Junos PKI を介して行われる証明書管理の操作を定義します。主な操作は以下の2つです。
-
インストール—ターゲットネットワークデバイスに新しい証明書IDを使用して新しい証明書をインストールします。証明書IDが既に存在する場合、操作はエラーを返します。
-
ローテーション—ターゲットネットワークデバイス上で、すでに既存の証明書IDを持っている既存の証明書を置き換えます。プロセス中にストリームが中断したり、ステップが失敗したりした場合、デバイスは元の証明書にロールバックします。
図1は、Install()とRotate()の操作のワークフローの概要を示しています。どちらの操作でも、クライアントは証明書署名要求(CSR)自体を生成するか、ターゲットにCSRの生成をリクエストできます。いずれの場合も、クライアントはCSRを認証機関(CA)に転送して、デジタル証明書をリクエストします。次に、クライアントは、Install()操作用の新しい証明書 ID、またはRotate()操作用の既存の証明書IDを使用して、ターゲットに証明書をロードします。Rotate()操作の場合、クライアントは交換証明書も検証し、検証の成功または失敗に基づいてRotate()操作を確定またはキャンセルする必要があります。クライアントが操作をキャンセルすると、サーバーは証明書、キーペア、およびCAバンドル(リクエストに存在する場合)をロールバックします。
Junos OS Evolvedリリース23.1R1以降、 Install()、 Rotate()、または LoadCertificate() 操作中に、gNOIサーバーは対応するCA証明書を使用して新しいエンドエンティティ証明書を検証します。したがって、gNOIサーバーのPKIには、新しい証明書を検証するルートCA証明書が含まれている必要があります。必要なCA証明書をgNOI CAバンドルの一部として読み込むことも、個別に読み込むこともできます。検証に失敗した場合、デバイスは新しい証明書をインストールしません。
gNOIサーバーは、gNOIサービス用に1つのグローバルCA証明書バンドルのみをサポートします。gNOI CertificateManagement サービスを使用してCAバンドルを読み込む場合、以下のステートメントが適用されます。
-
CertificateManagementサービスは、常にca-profile-group予約済み識別子gnoi-ca-bundleを使用して、CA証明書バンドルを読み込みます。 -
CertificateManagementサービスを使用してCA証明書バンドルを読み込む場合、デバイスは暗黙的に相互認証を使用します。 -
CertificateManagementサービスが新しいCA証明書バンドルを読み込むリクエストを送信すると、サーバーはデバイスから以前のCAバンドルの証明書をクリアし、新しい証明書を読み込みます。 -
CertificateManagementサービスを使用してCA証明書バンドルを読み込み、デバイス構成で相互認証を明示的に設定する場合は、設定されたステートメントが優先されます。
したがって、最初にgNOIサーバーでサーバー専用認証を設定してから、 Install() RPCを使用してCA証明書を読み込むことができます。gNOI を使用して初期 CA 証明書バンドルを読み込むと、デバイスは以下の手順を実行します。
- Junos PKIにCA証明書を追加します。
ca-profile-group識別子gnoi-ca-bundleを使用して、[edit security pki]階層レベルでgNOI CA証明書バンドルを自動的に設定します。- サーバーのみの認証から相互認証にスイッチします。
[edit]
security {
pki {
ca-profile gnoi-ca-bundle_1 {
ca-identity gnoi-ca-bundle_1;
}
ca-profile gnoi-ca-bundle_2 {
ca-identity gnoi-ca-bundle_2;
}
ca-profile-group gnoi-ca-bundle {
cert-base-count 2;
}
}
}
Rotate() RPCは、回転操作中の認証モード間の切り替えをサポートしていません。したがって、Rotate()は、操作中にデバイスがサーバーのみの認証から相互認証に切り替わる原因となるため、gNOIサーバーへのCA証明書バンドルの初めての読み込みをサポートしていません。認証モードが変更されると、ネットワークデバイスはgRPCスタックを再起動する必要があり、接続は失われます。ストリームが中断した場合、クライアントはローテーション要求を完了できず、デバイスはRotate()要求が開始される前に配置されていた証明書にロールバックします。
[edit system services extension-service request-response grpc ssl]階層レベルのhot-reloadingステートメントは、証明書更新時に、操作中に認証モードが変更されていない場合にのみ、gRPCセッションを維持します。たとえば、認証モードがサーバーのみから相互認証に切り替わる、またはその逆の場合、クライアントは切断されます。
サポートされているRPC
表1は、JunosデバイスでサポートされるCertificateManagementサービスRPCの概要を示しています。
| RPC | 説明 | リリースで導入 |
|---|---|---|
CanGenerateCSR() |
ターゲットデバイスにクエリして、指定されたキータイプ、キーサイズ、証明書タイプで証明書署名要求(CSR)を生成できるかどうかを判断します。サポートされている値:
gNOIサーバーが特定のキータイプ、キーサイズ、および証明書タイプをサポートしているかどうか |
Junos OS Evolved 23.1R1 |
GenerateCSR() |
証明書署名要求(CSR)を生成して返します。 |
Junos OS Evolved 22.2R1 |
GetCertificates() |
ターゲットデバイスにロードされているローカル証明書を返します。 |
Junos OS Evolved 22.2R1 |
Install() |
CSRリクエストを作成し、CSRに基づいて証明書を生成し、新しい証明書IDを使用して証明書をロードすることで、ターゲットデバイスに新しい証明書をロードします。 |
Junos OS Evolved 22.2R1 |
LoadCertificate() |
認証機関(CA)によって署名された証明書をターゲットデバイスにロードします。 |
Junos OS Evolved 22.2R1 |
LoadCertificateAuthorityBundle() |
ターゲットデバイスにCA証明書バンドルをロードします。 |
Junos OS Evolved 22.2R1 |
RevokeCertificates() |
ターゲットデバイスで指定された証明書IDを持つ証明書を取り消します。 |
Junos OS Evolved 23.1R1 |
Rotate() |
CSRリクエストを作成し、CSRに基づいて証明書を生成し、既存の証明書IDを使用して証明書を読み込むことで、ターゲットデバイス上の既存の証明書を置き換えます。 |
Junos OS Evolved 22.2R1 |
ネットワークデバイスの設定
始める前に:
- gRPCサービスを設定するの説明に従って、ネットワークデバイス上でgRPCサービスを設定します。
- gNOIサービスを設定するで説明されているように、gNOI操作をサポートするようにネットワーク管理システムを構成します。
[edit system services http servers]階層レベルで設定されたgNOIサーバーの場合、追加の設定は必要ありません。
[edit system services extension-service request-response grpc ssl]階層レベルで設定されたgNOIサーバーの場合、同じ階層レベルでuse-pkiおよびhot-reloadingステートメントを設定する必要があります。hot-reloadingステートメントは、セッションに影響を与える証明書を更新する際にgRPCセッションを維持するために必要です。ステートメントを設定するには:
証明書をインストールする
RPCInstall()CertificateManagementサービスを使用して、ターゲットデバイスに新しい証明書をロードできます。Install()操作を使用して新しい証明書をインストールする場合、ターゲットデバイスにまだ存在しない新しい証明書IDを指定する必要があります。オプションで、Install()操作の一部としてCA証明書バンドルを読み込むこともできます。
Install()操作の一環として、デバイスは新しい証明書を検証します。そのため、Junos PKIには、新しい証明書を検証するルートCA証明書が必要です。必要なCA証明書は、Install()操作の一部として読み込むことができます。PKI にまだない場合は、操作の前に個別に読み込むこともできます。
gRPCセッション認証に使用する新しいローカル証明書をインストールする場合は、新しい証明書IDを使用するように、デバイス上のgRPCサーバー設定も更新する必要があります。
例:証明書をインストールする
この例では、gNOIサーバーは最初はローカル証明書のみで構成されており、相互認証を使用するようには構成されていません。gNOIクライアントは、 Install() RPCを使用して、新しいローカル証明書とCA証明書バンドルをデバイスにロードします。CAバンドルがgNOIサーバーにロードされると、サーバーはデフォルトで相互認証を使用します。CAバンドルには、クライアント証明書のルートCA証明書と、新しいサーバー証明書のルートCA証明書が含まれています。
クライアントは、以下の操作を実行する gnoi_cert_install_certificate_csr.py Pythonアプリケーションを実行します。
- ターゲットにCSRを生成するようリクエストします。
- CSRに基づいて署名された証明書を取得します。
- 新しいサーバー証明書、サーバーの新しいルート CA 証明書、およびクライアントのルート CA 証明書をターゲットネットワークデバイスにロードします。
アプリケーションは、適切なパラメータとともに InstallCertificateRequest メッセージを使用して、CSRの生成と証明書の読み込みのためのリクエストを定義します。各要求に対して、アプリケーションは Install() RPCを使用して要求をネットワークデバイスに送信します。
gnoi_cert_install_certificate_csr.pyアプリケーションは、チャネルを確立するためのgrpc_channelモジュールをインポートします。grpc_channelモジュールについては、「gNOIサービスを設定する」で説明しています。アプリケーションの引数はargs_cert_install_csr.txtファイルに格納されます。アプリケーションファイルと引数ファイルがここに示されています。
gnoi_cert_install_certificate_csr.py
"""gNOI Install Certificate utility."""
from __future__ import print_function
from __future__ import unicode_literals
import argparse
import logging
import re
from getpass import getpass
from subprocess import call
import cert_pb2
import cert_pb2_grpc
from grpc_channel import grpc_authenticate_channel_server_only
def get_args(parser):
parser.add_argument('--server',
dest='server',
type=str,
default='localhost',
help='Server IP or name. Default is localhost.')
parser.add_argument('--port',
dest='port',
nargs='?',
type=int,
default=32767,
help='Server port. Default is 32767')
parser.add_argument('--client_key',
dest='client_key',
type=str,
default='',
help='Full path of the client private key. Default is "".')
parser.add_argument('--client_cert',
dest='client_cert',
type=str,
default='',
help='Full path of the client certificate. Default is "".')
parser.add_argument('--root_ca_cert',
dest='root_ca_cert',
required=True,
type=str,
help='Full path of the Root CA certificate.')
parser.add_argument('--user_id',
dest='user_id',
required=True,
type=str,
help='User ID for RPC call credentials.')
parser.add_argument('--type',
dest='type',
type=int,
default='1',
help='Certificate Type. Default is 1. Valid value is 1 (1 is CT_X509); Invalid value is 0 (0 is CT_UNKNOWN).')
parser.add_argument('--min_key_size',
dest='min_key_size',
type=int,
default='2048',
help='Minimum key size. Default is 2048.')
parser.add_argument('--key_type',
dest='key_type',
type=int,
default='1',
help='Key Type. Default is 1 (KT_RSA); 0 is KT_UNKNOWN.')
parser.add_argument('--common_name',
dest='common_name',
type=str,
default='',
help='CN of the certificate')
parser.add_argument('--country',
dest='country',
type=str,
default='US',
help='Country name')
parser.add_argument('--state',
dest='state',
type=str,
default='CA',
help='State name')
parser.add_argument('--city',
dest='city',
type=str,
default='Sunnyvale',
help='City name')
parser.add_argument('--organization',
dest='organization',
type=str,
default='Acme',
help='Organization name')
parser.add_argument('--organizational_unit',
dest='organizational_unit',
type=str,
default='Test',
help='Organization unit name')
parser.add_argument('--ip_address',
dest='ip_address',
type=str,
default='',
help='IP address on the certificate')
parser.add_argument('--email_id',
dest='email_id',
type=str,
default='',
help='Email id')
parser.add_argument('--certificate_id',
dest='certificate_id',
required=True,
type=str,
help='Certificate id.')
parser.add_argument('--server_cert_private_key',
dest='server_cert_private_key',
type=str,
default='',
help='Server certificate private key')
parser.add_argument('--server_cert_public_key',
dest='server_cert_public_key',
type=str,
default='',
help='Server certificate public key')
parser.add_argument('--server_cert',
dest='server_cert',
type=str,
default='server_cert',
help='Server certificate')
parser.add_argument('--server_root_ca1',
dest='server_root_ca1',
type=str,
default='server_root_ca1',
help='Server Root CA')
parser.add_argument('--server_root_ca2',
dest='server_root_ca2',
type=str,
default='server_root_ca2',
help='Server Root CA')
parser.add_argument('--client_root_ca1',
dest='client_root_ca1',
type=str,
default='client_root_ca1',
help='Client Root CA')
parser.add_argument('--client_root_ca2',
dest='client_root_ca2',
type=str,
default='client_root_ca2',
help='Client Root CA')
parser.add_argument('--client_root_ca3',
dest='client_root_ca3',
type=str,
default='client_root_ca3',
help='Client Root CA')
parser.add_argument('--client_root_ca4',
dest='client_root_ca4',
type=str,
default='client_root_ca4',
help='Client Root CA')
args = parser.parse_args()
return args
def install_cert(channel, metadata, args):
try:
stub = cert_pb2_grpc.CertificateManagementStub(channel)
print("Executing GNOI::CertificateManagement::Install")
# Create request to generate certificate signing request (CSR)
it = []
req = cert_pb2.InstallCertificateRequest()
req.generate_csr.csr_params.type = args.type
req.generate_csr.csr_params.min_key_size = args.min_key_size
req.generate_csr.csr_params.key_type = args.key_type
req.generate_csr.csr_params.common_name = args.common_name
req.generate_csr.csr_params.country = args.country
req.generate_csr.csr_params.state = args.state
req.generate_csr.csr_params.city = args.city
req.generate_csr.csr_params.organization = args.organization
req.generate_csr.csr_params.organizational_unit = args.organizational_unit
req.generate_csr.csr_params.ip_address = args.ip_address
req.generate_csr.csr_params.email_id = args.email_id
req.generate_csr.certificate_id = args.certificate_id
it.append(req)
# Send request to generate CSR
for csr_rsp in stub.Install(iter(it), metadata=metadata, timeout=180):
logging.info(csr_rsp)
# Write CSR to a file
with open('/home/lab/certs/server_temp.csr', "wb") as file:
file.write(csr_rsp.generated_csr.csr.csr)
# If client connects to server IP address
# update openssl.cnf template to include subjectAltName IP extension
with open('/etc/pki/certs/openssl.cnf', 'r') as fd:
data = fd.read()
data1 = re.sub(r'(subjectAltName=IP:).*',
r'\g<1>'+args.ip_address, data)
with open('/home/lab/certs/openssl_temp.cnf', 'w') as fd:
fd.write(data1)
# Generate certificate with v3 extensions
cmd = "openssl x509 -req -days 365 -in /home/lab/certs/server_temp.csr -CA /etc/pki/certs/serverRootCA.crt -CAkey /etc/pki/certs/serverRootCA.key -CAcreateserial -out /home/lab/certs/server_temp.crt -extensions v3_sign -extfile /home/lab/certs/openssl_temp.cnf -sha384"
decrypted = call(cmd, shell=True)
# Create request to install node certificate and CA certificates
print("\nExecuting GNOI::CertificateManagement::Install")
it = []
req = cert_pb2.InstallCertificateRequest()
# Import certificate and add to request
cert_data = bytearray(b'')
with open("/home/lab/certs/server_temp.crt", "rb") as file:
cert_data = file.read()
req.load_certificate.certificate.type = args.type
req.load_certificate.certificate_id = args.certificate_id
req.load_certificate.certificate.certificate = cert_data
# Add client and server CA certificates to request
ca1 = req.load_certificate.ca_certificates.add()
ca1.type = args.type
ca1.certificate = open(args.client_root_ca1, 'rb').read()
ca2 = req.load_certificate.ca_certificates.add()
ca2.type = args.type
ca2.certificate = open(args.server_root_ca1, 'rb').read()
it.append(req)
# Send request to install node certificate and CA bundle
for rsp in stub.Install(iter(it), metadata=metadata, timeout=180):
logging.info("Installing certificates: %s", rsp)
print("Install complete.")
except Exception as e:
logging.error('Certificate install error: %s', e)
print(e)
def main():
parser = argparse.ArgumentParser(fromfile_prefix_chars='@')
args = get_args(parser)
grpc_server_password = getpass("gRPC server password for executing RPCs: ")
metadata = [('username', args.user_id),
('password', grpc_server_password)]
try:
# Establish grpc channel to network device
channel = grpc_authenticate_channel_server_only(
args.server, args.port, args.root_ca_cert)
install_cert(channel, metadata, args)
except Exception as e:
logging.error('Received error: %s', e)
print(e)
if __name__ == '__main__':
logging.basicConfig(filename='gnoi-testing.log',
format='%(asctime)s %(levelname)-8s %(message)s',
level=logging.INFO,
datefmt='%Y-%m-%d %H:%M:%S')
main()
args_cert_install_csr.txt
--server=10.53.52.169 --port=32767 --root_ca_cert=/etc/pki/certs/serverRootCA.crt --user_id=gnoi-user --type=1 --min_key_size=2048 --key_type=1 --common_name=gnoi-server.example.com --country=US --state=CA --city=Sunnyvale --organization=Acme --organizational_unit=testing --ip_address=10.53.52.169 --email_id=test@example.com --certificate_id=gnoi-server1 --client_root_ca1=/etc/pki/certs/clientRootCA.crt --server_root_ca1=/etc/pki/certs/serverRootCA1.crt
アプリケーションの実行
クライアントがアプリケーションを実行すると、アプリケーションはCSRを要求し、署名された証明書を取得して、新しいサーバー証明書とCA証明書をターゲットネットワークデバイスにロードします。
lab@gnoi-client:~/src/gnoi/proto$ python3 gnoi_cert_install_certificate_csr.py @args_cert_install_csr.txt gRPC server password for executing RPCs: Creating channel Executing GNOI::CertificateManagement::Install Signature ok subject=CN = gnoi-server.example.com, C = US, ST = CA, O = Acme, OU = testing Getting CA Private Key Executing GNOI::CertificateManagement::Install Install complete.
新しいサーバー証明書をインストールした後、gRPCセッション認証にその証明書IDを使用するようにサーバーを設定する必要があります。特定のサーバー設定用に local-certificate ステートメントを更新します。この設定は、以下のいずれかの階層レベルで設定できます。
-
[システムサービス拡張サービスリクエスト応答gRPC SSLの編集]
-
[TLS name システムサービスHTTPサーバーサーバーの編集]
次に例を示します。
user@gnoi-server> show configuration system services extension-service request-response grpc ssl port 32767; local-certificate gnoi-server1; hot-reloading; use-pki;
さらに、 Install() 操作は新しいCA証明書を読み込むため、デバイスは暗黙的に相互認証を使用します。その結果、チャネルを確立する際に、後続のすべてのgRPCセッションにクライアントの証明書とキーを含める必要があります。
アプリケーションを実行し、サーバー上に既に存在する証明書IDを指定すると、Install()操作には新しい証明書IDが必要になるため、アプリケーションはALREADY_EXISTSエラーを返します。
lab@gnoi-client:~/src/gnoi/proto$ python3 gnoi_cert_install_certificate_csr.py @args_cert_install_csr.txt
gRPC server password for executing RPCs:
Creating channel
Executing GNOI::CertificateManagement::Install
Signature ok
subject=CN = gnoi-server.example.com, C = US, ST = CA, O = Acme, OU = testing
Getting CA Private Key
Executing GNOI::CertificateManagement::Install
<_MultiThreadedRendezvous of RPC that terminated with:
status = StatusCode.ALREADY_EXISTS
details = ""
debug_error_string = "{"created":"@1652241881.676147097","description":"Error received from peer ipv4:10.53.52.169:32767","file":"src/core/lib/surface/call.cc","file_line":903,"grpc_message":"","grpc_status":6}"
証明書のローテーション
RPCRotate()CertificateManagementサービスを使用して、ターゲットデバイス上の既存の証明書を置き換えることができます。Rotate()操作を使用して既存の証明書を置き換える場合は、ターゲットデバイスにすでに存在する証明書IDを使用して証明書を読み込む必要があります。オプションで、Rotate()操作の一環として既存のgNOI CA証明書バンドルを置き換えることもできます。
Rotate()操作はInstall()操作と似ていますが、Rotate()操作は新しい証明書をインストールするのではなく、既存の証明書を置き換える点が異なります。さらに、クライアントは、更新された証明書が機能することを検証し、証明書検証の成功または失敗に基づいてRotate()要求を確定またはキャンセルする必要があります。
Rotate()操作の一環として、デバイスは新しい証明書を検証します。そのため、Junos PKIには、新しい証明書を検証するルートCA証明書が必要です。必要なCA証明書は、Rotate()操作の一部として読み込むことができます。PKI にまだない場合は、操作の前に個別に読み込むこともできます。
例:証明書のローテーション
この例では、クライアントは gnoi_cert_rotate_certificate_csr.py Pythonアプリケーションを実行し、以下の操作を実行します。
- ターゲットにCSRを生成するようリクエストします。
- CSRに基づいて署名された証明書を取得します。
- ターゲットネットワークデバイス上のノード証明書とgNOI CAバンドルを置き換えます。
- 新しい証明書を検証します。
Rotate操作を終了します。
アプリケーションは、 RotateCertificateRequest メッセージを適切なパラメータとともに使用して、CSRの生成、証明書およびCAバンドルのロードのためのリクエストを定義します。各要求に対して、アプリケーションは Rotate() RPCを使用して要求をネットワークデバイスに送信します。ターゲットデバイスが新しいノード証明書を検証できるように、アプリケーションは既存のCAバンドルを新しいCAバンドルに置き換えます。バンドルには、クライアントCA証明書と、ノード証明書の検証に必要なCA証明書の両方が含まれています。
アプリケーションは、ネットワークデバイスとの新しいgRPCセッションを作成し、単純な Time() RPCを実行することで、新しい証明書が動作することを検証しますが、任意のRPCでセッション認証テストできます。セッションが正常に確立されると、アプリケーションはローテーション要求を確定し、セッション認証に失敗するとローテーション要求を取り消します。
gnoi_cert_rotate_certificate_csr.pyアプリケーションは、チャネルを確立するためのgrpc_channelモジュールをインポートします。grpc_channelモジュールについては、gNOIサービスを設定するで説明しています。アプリケーションの引数はargs_cert_rotate_csr.txtファイルに格納されます。アプリケーションファイルと引数ファイルがここに示されています。
gnoi_cert_rotate_certificate_csr.py
"""gNOI Rotate Certificate utility."""
from __future__ import print_function
from __future__ import unicode_literals
import argparse
import logging
import time
import re
import grpc
from getpass import getpass
from subprocess import call
import cert_pb2
import cert_pb2_grpc
import system_pb2
import system_pb2_grpc
from grpc_channel import grpc_authenticate_channel_mutual
def get_args(parser):
parser.add_argument('--server',
dest='server',
type=str,
default='localhost',
help='Server IP or name. Default is localhost.')
parser.add_argument('--port',
dest='port',
nargs='?',
type=int,
default=32767,
help='Server port. Default is 32767')
parser.add_argument('--client_key',
dest='client_key',
type=str,
default='',
help='Full path of the client private key. Default is "".')
parser.add_argument('--client_cert',
dest='client_cert',
type=str,
default='',
help='Full path of the client certificate. Default is "".')
parser.add_argument('--root_ca_cert',
dest='root_ca_cert',
required=True,
type=str,
help='Full path of the Root CA certificate.')
parser.add_argument('--user_id',
dest='user_id',
required=True,
type=str,
help='User ID for RPC call credentials.')
parser.add_argument('--type',
dest='type',
type=int,
default='1',
help='Certificate Type. Default is 1. Valid value is 1 (1 is CT_X509); Invalid value is 0 (0 is CT_UNKNOWN).')
parser.add_argument('--min_key_size',
dest='min_key_size',
type=int,
default='2048',
help='Minimum key size. Default is 2048.')
parser.add_argument('--key_type',
dest='key_type',
type=int,
default='1',
help='Key Type. Default is 1 (KT_RSA); 0 is KT_UNKNOWN.')
parser.add_argument('--common_name',
dest='common_name',
type=str,
default='',
help='CN of the certificate')
parser.add_argument('--country',
dest='country',
type=str,
default='US',
help='Country name')
parser.add_argument('--state',
dest='state',
type=str,
default='CA',
help='State name')
parser.add_argument('--city',
dest='city',
type=str,
default='Sunnyvale',
help='City name')
parser.add_argument('--organization',
dest='organization',
type=str,
default='Acme',
help='Organization name')
parser.add_argument('--organizational_unit',
dest='organizational_unit',
type=str,
default='Test',
help='Organization unit name')
parser.add_argument('--ip_address',
dest='ip_address',
type=str,
default='',
help='IP address on the certificate')
parser.add_argument('--email_id',
dest='email_id',
type=str,
default='',
help='Email id')
parser.add_argument('--certificate_id',
dest='certificate_id',
required=True,
type=str,
help='Certificate id.')
parser.add_argument('--server_cert_private_key',
dest='server_cert_private_key',
type=str,
default='',
help='Server certificate private key')
parser.add_argument('--server_cert_public_key',
dest='server_cert_public_key',
type=str,
default='',
help='Server certificate public key')
parser.add_argument('--server_cert',
dest='server_cert',
type=str,
default='server_cert',
help='Server certificate')
parser.add_argument('--server_root_ca1',
dest='server_root_ca1',
type=str,
default='server_root_ca1',
help='Server Root CA')
parser.add_argument('--server_root_ca2',
dest='server_root_ca2',
type=str,
default='server_root_ca2',
help='Server Root CA')
parser.add_argument('--client_root_ca1',
dest='client_root_ca1',
type=str,
default='client_root_ca1',
help='Client Root CA')
parser.add_argument('--client_root_ca2',
dest='client_root_ca2',
type=str,
default='client_root_ca2',
help='Client Root CA')
parser.add_argument('--client_root_ca3',
dest='client_root_ca3',
type=str,
default='client_root_ca3',
help='Client Root CA')
parser.add_argument('--client_root_ca4',
dest='client_root_ca4',
type=str,
default='client_root_ca4',
help='Client Root CA')
parser.add_argument("--client_key_test",
dest='client_key_test',
type=str,
default='',
help='Full path of the test client private key. Default ""')
parser.add_argument("--client_cert_test",
dest='client_cert_test',
type=str,
default='',
help='Full path of the test client certificate. Default ""')
args = parser.parse_args()
return args
def rotate_cert(channel, metadata, args):
try:
result = ''
stub = cert_pb2_grpc.CertificateManagementStub(channel)
print("Executing GNOI::CertificateManagement::Rotate")
# Create request to generate certificate signing request (CSR)
it = []
req = cert_pb2.RotateCertificateRequest()
req.generate_csr.csr_params.type = args.type
req.generate_csr.csr_params.min_key_size = args.min_key_size
req.generate_csr.csr_params.key_type = args.key_type
req.generate_csr.csr_params.common_name = args.common_name
req.generate_csr.csr_params.country = args.country
req.generate_csr.csr_params.state = args.state
req.generate_csr.csr_params.city = args.city
req.generate_csr.csr_params.organization = args.organization
req.generate_csr.csr_params.organizational_unit = args.organizational_unit
req.generate_csr.csr_params.ip_address = args.ip_address
req.generate_csr.csr_params.email_id = args.email_id
req.generate_csr.certificate_id = args.certificate_id
it.append(req)
# Send request to generate CSR
print('Sending request for CSR')
for csr_rsp in stub.Rotate(iter(it), metadata=metadata, timeout=30):
logging.info(csr_rsp)
# Write CSR to a file
with open('/home/lab/certs/server_temp.csr', "wb") as file:
file.write(csr_rsp.generated_csr.csr.csr)
# If client connects to server IP address
# update openssl.cnf template to include subjectAltName IP extension
with open('/etc/pki/certs/openssl.cnf', 'r') as fd:
data = fd.read()
data1 = re.sub(r'(subjectAltName=IP:).*',
r'\g<1>'+args.ip_address, data)
with open('/home/lab/certs/openssl_temp.cnf', 'w') as fd:
fd.write(data1)
# Generate certificate with v3 extensions
cmd = "openssl x509 -req -days 365 -in /home/lab/certs/server_temp.csr -CA /etc/pki/certs/serverRootCA.crt -CAkey /etc/pki/certs/serverRootCA.key -CAcreateserial -out /home/lab/certs/server_temp.crt -extensions v3_sign -extfile /home/lab/certs/openssl_temp.cnf -sha384"
decrypted = call(cmd, shell=True)
# Create request to rotate node certificate and CA certificates
print("\nExecuting GNOI::CertificateManagement::Rotate")
it = []
req = cert_pb2.RotateCertificateRequest()
# Import certificate and add to request
with open("/home/lab/certs/server_temp.crt", "rb") as file:
cert_data = file.read()
req.load_certificate.certificate.type = args.type
req.load_certificate.certificate.certificate = cert_data
req.load_certificate.certificate_id = args.certificate_id
# Add client and server CA certificates to request
ca1 = req.load_certificate.ca_certificates.add()
ca1.type = args.type
ca1.certificate = open(args.client_root_ca1, 'rb').read()
ca2 = req.load_certificate.ca_certificates.add()
ca2.type = args.type
ca2.certificate = open(args.server_root_ca1, 'rb').read()
it.append(req)
# Send request to replace node certificate and CA bundle
for rsp in stub.Rotate(iter(it), metadata=metadata, timeout=60):
logging.info("Rotating certificates. %s", rsp)
# Validate certificates
print("Validating certificates")
time.sleep(5)
validate_rc = True
try:
validate_channel = grpc_authenticate_channel_mutual(
args.server, args.port, args.server_root_ca1, args.client_key_test,
args.client_cert_test)
validate_stub = system_pb2_grpc.SystemStub(validate_channel)
validate_rsp = validate_stub.Time(
request=system_pb2.TimeRequest(), metadata=metadata, timeout=60)
except grpc.RpcError as e:
print("Validation failed with error:", e)
validate_rc = False
pass
if validate_rc:
print("Finalizing certificate rotation.")
it = []
req = cert_pb2.RotateCertificateRequest()
req.finalize_rotation.SetInParent()
it.append(req)
for rsp in stub.Rotate(iter(it), metadata=metadata, timeout=30):
logging.info("Finalizing rotate. %s", rsp)
logging.info(
"Certificate validation succeeded. Certificate rotation finalized.")
result = "Certificate rotation finalized."
else:
print("Rolling back certificates.")
logging.info(
"Certificate validation failed. Rolling back to original certificates.")
rsp.cancel()
except Exception as e:
logging.error('Certificate rotate error: %s', e)
print(e)
else:
return result
def main():
parser = argparse.ArgumentParser(fromfile_prefix_chars='@')
args = get_args(parser)
grpc_server_password = getpass("gRPC server password for executing RPCs: ")
metadata = [('username', args.user_id),
('password', grpc_server_password)]
try:
# Establish grpc channel to network device
channel = grpc_authenticate_channel_mutual(
args.server, args.port, args.root_ca_cert, args.client_key, args.client_cert)
response = rotate_cert(channel, metadata, args)
print(response)
except Exception as e:
logging.error('Error: %s', e)
print(e)
if __name__ == '__main__':
logging.basicConfig(filename='gnoi-testing.log',
format='%(asctime)s %(levelname)-8s %(message)s',
level=logging.INFO,
datefmt='%Y-%m-%d %H:%M:%S')
main()
args_cert_rotate_csr.txt
--server=10.53.52.169 --port=32767 --root_ca_cert=/etc/pki/certs/serverRootCA.crt --client_key=/home/lab/certs/client.key --client_cert=/home/lab/certs/client.crt --user_id=gnoi-user --type=1 --min_key_size=2048 --key_type=1 --common_name=gnoi-server.example.com --country=US --state=CA --city=Sunnyvale --organization=Acme --organizational_unit=testing --ip_address=10.53.52.169 --email_id=test@example.com --certificate_id=gnoi-server --client_root_ca1=/etc/pki/certs/clientRootCA.crt --server_root_ca1=/etc/pki/certs/serverRootCA.crt --client_key_test=/home/lab/certs/client.key --client_cert_test=/home/lab/certs/client.crt
root_ca_cert引数は、初期チャネル資格情報に必要なサーバーのルートCA証明書であることに注意することが重要です。server_root_ca1引数は、サーバーの新しい証明書に対応するルートCA証明書です。Rotate()操作中に新しいローカル証明書を検証するには、Junos PKIに新しいルートCA証明書が必要です。さらに、新しい証明書を検証するgRPCセッションのチャネル資格情報は、このルートCA証明書を使用します。この例では、新しいサーバー証明書と古いサーバー証明書に同じルート CA 証明書を使用していますが、別のケースでは異なる場合があります。
アプリケーションの実行
クライアントがアプリケーションを実行すると、アプリケーションはCSRを要求し、署名された証明書を取得して、交換証明書とCAバンドルをターゲットネットワークデバイスにロードします。その後、アプリケーションは、単純な Time() RPCを実行する新しいgRPCセッションで交換証明書を検証します。検証に成功すると、クライアントはローテーション要求を確定します。
lab@gnoi-client:~/src/gnoi/proto$ python3 gnoi_cert_rotate_certificate_csr.py @args_cert_rotate_csr.txt gRPC server password for executing RPCs: Creating channel Executing GNOI::CertificateManagement::Rotate Sending request for CSR Signature ok subject=CN = gnoi-server.example.com, C = US, ST = CA, O = Acme, OU = testing Getting CA Private Key Executing GNOI::CertificateManagement::Rotate Validating certificates Creating channel Finalizing certificate rotation. Certificate rotation finalized.
証明書の失効
gNOIクライアントは、 RevokeCertificates() RPCを使用して、ターゲットデバイスから1つ以上の証明書を削除できます。クライアントには、取り消す証明書IDのリストが記載された RevokeCertificatesRequest メッセージが含まれます。
gNOIサーバーは、 RevokeCertificates() リクエストを受信すると、リスト内の各証明書IDを次のように処理します。
-
証明書が存在し、失効が成功した場合、デバイスはファイル システムと Junos PKI から証明書を削除し、正常に失効した証明書のリストに証明書 ID を追加します。
-
証明書が存在する場合、失効に失敗した場合、デバイスは証明書IDと失敗の理由を証明書失効エラーリストに含めます。
-
証明書が存在しない場合、デバイスは失効操作が成功したとみなし、正常に失効した証明書のリストに証明書IDを追加します。
リクエストによって現在のセッションに使用されている証明書が取り消された場合、セッションは影響を受けません。
リクエストを処理した後、gNOIサーバーは以下を含む RevokeCertificatesResponse メッセージを返します。
-
正常に取り消された証明書IDのリスト。
-
証明書IDと失敗の理由を含む失効エラーのリスト。
例:証明書の失効
この例では、クライアントが gnoi_cert_revoke_certificates.py Pythonアプリケーションを実行し、サーバー上の2つの証明書を取り消します。最初の証明書IDは、デバイス上の有効な識別子です。2 番目の証明書 ID は、デバイスに存在しない識別子です。
アプリケーションは、 RevokeCertificatesRequest メッセージを適切なパラメーターとともに使用して要求を定義します。アプリケーションは、 RevokeCertificates() RPCをネットワークデバイスに送信して操作を実行します。
gnoi_cert_revoke_certificates.pyアプリケーションは、チャネルを確立するためのgrpc_channelモジュールをインポートします。grpc_channelモジュールについては、gNOIサービスを設定するで説明しています。アプリケーションの引数はargs_cert_revoke_certificates.txtファイルに格納されます。アプリケーションファイルと引数ファイルがここに示されています。
gnoi_cert_revoke_certificates.py
"""gNOI Revoke Certificates utility."""
from __future__ import print_function
from __future__ import unicode_literals
import argparse
import logging
from getpass import getpass
import cert_pb2
import cert_pb2_grpc
from grpc_channel import grpc_authenticate_channel_mutual
def get_args(parser):
parser.add_argument('--server',
dest='server',
type=str,
default='localhost',
help='Server IP or name. Default is localhost.')
parser.add_argument('--port',
dest='port',
nargs='?',
type=int,
default=32767,
help='Server port. Default is 32767')
parser.add_argument('--client_key',
dest='client_key',
type=str,
default='',
help='Full path of the client private key. Default is "".')
parser.add_argument('--client_cert',
dest='client_cert',
type=str,
default='',
help='Full path of the client certificate. Default is "".')
parser.add_argument('--root_ca_cert',
dest='root_ca_cert',
required=True,
type=str,
help='Full path of the Root CA certificate.')
parser.add_argument('--user_id',
dest='user_id',
required=True,
type=str,
help='User ID for RPC call credentials.')
parser.add_argument('--certificate_id1',
dest='certificate_id1',
required=True,
type=str,
help='Certificate id.')
parser.add_argument('--certificate_id2',
dest='certificate_id2',
type=str,
help='Certificate id.')
args = parser.parse_args()
return args
def revoke_cert(channel, metadata, args):
try:
stub = cert_pb2_grpc.CertificateManagementStub(channel)
print("Executing GNOI::CertificateManagement::RevokeCertificates")
# Create request to revoke certificates
req = cert_pb2.RevokeCertificatesRequest()
req.certificate_id.append(args.certificate_id1)
req.certificate_id.append(args.certificate_id2)
# Send request to revoke certificates
logging.info("Sending RevokeCertificates request.")
rsp = stub.RevokeCertificates(req, metadata=metadata, timeout=60)
logging.info(rsp)
print("rsp:\n%s" %rsp)
except Exception as e:
logging.error('Error: %s', e)
print(e)
def main():
parser = argparse.ArgumentParser(fromfile_prefix_chars='@')
args = get_args(parser)
grpc_server_password = getpass("gRPC server password for executing RPCs: ")
metadata = [('username', args.user_id),
('password', grpc_server_password)]
try:
# Establish grpc channel to network device
channel = grpc_authenticate_channel_mutual(
args.server, args.port, args.root_ca_cert, args.client_key, args.client_cert)
revoke_cert(channel, metadata, args)
except Exception as e:
logging.error('Received error: %s', e)
print(e)
if __name__ == '__main__':
logging.basicConfig(filename='gnoi-testing.log',
format='%(asctime)s %(levelname)-8s %(message)s',
level=logging.INFO,
datefmt='%Y-%m-%d %H:%M:%S')
main()
args_cert_revoke_certificates.txt
--server=10.53.52.169 --port=32767 --root_ca_cert=/etc/pki/certs/serverRootCA.crt --client_key=/home/lab/certs/client.key --client_cert=/home/lab/certs/client.crt --user_id=gnoi-user --certificate_id1=gnoi-server --certificate_id2=id-does-not-exist
アプリケーションの実行
クライアントがアプリケーションを実行すると、指定された証明書を取り消すようにターゲットデバイスに指示します。デバイスは、正常に取り消された証明書とエラーのリストを返します。デバイスは、有効な証明書IDと、デバイスに現在存在しない証明書IDの両方について、操作が成功したとみなします。
lab@gnoi-client:~/src/gnoi/proto$ python3 gnoi_cert_revoke_certificates.py @args_cert_revoke_certificates.txt gRPC server password for executing RPCs: Creating channel Executing GNOI::CertificateManagement::RevokeCertificates rsp: revoked_certificate_id: "gnoi-server" revoked_certificate_id: "id-does-not-exist"
変更履歴テーブル
サポートされる機能は、使用しているプラットフォームとリリースによって決まります。 機能エクスプローラー を使用して、機能がお使いのプラットフォームでサポートされているかどうかを確認します。
Install()、
Rotate()、および
LoadCertificate() の操作は、操作の一環として新しい証明書を検証します。