gNOI证书管理服务
使用 gNOI CertificateManagement 服务管理目标网元上的证书。
概述
包中的 gnoi.certificate gNOI CertificateManagement 服务处理目标网络元素上的证书管理。原型定义文件位于 https://github.com/openconfig/gnoi/blob/master/cert/cert.proto。
公钥基础架构 （PKI） 支持公共加密密钥的分发和识别，使用户既能通过互联网等网络安全地交换数据，又能验证对方的身份。您可以通过 Junos PKI 管理 Junos 设备上的公钥证书，包括下载、生成和验证证书。gNOI CertificateManagement 服务可定义通过 Junos PKI 进行的证书管理操作。两个主要操作是：
-
安装 — 在目标网络设备上使用新证书 ID 安装新证书。如果证书 ID 已存在，则该操作将返回错误。
-
轮换 - 替换目标网络设备上已具有现有证书 ID 的现有证书。如果流中断或在此过程中任何步骤失败，则设备将回滚到原始证书。
图 1 概述了 和 Install() Rotate() 运维的工作流程。对于这两种操作，客户端可以自己生成证书签名请求 （CSR） 或请求目标生成 CSR。在任何一种情况下，客户端都会将 CSR 转发到证书颁发机构 （证书颁发机构） 以请求数字证书。然后，客户端在目标上加载证书，要么使用新的证书 ID 进行操作，Install()要么使用现有证书 ID 进行Rotate()操作。对于Rotate()操作，客户端还应验证任何替换证书，并根据验证的成功或失败完成或取消Rotate()操作。如果客户机取消该操作，那么服务器将回滚证书、密钥对和任何 证书颁发机构包（如果请求中存在）。
从 Junos OS 演化版 23.1R1 开始，在 、 Rotate()或 LoadCertificate() 操作期间Install()，gNOI 服务器会使用相应的 CA 证书验证新的证书颁发机构证书。因此，gNOI 服务器的 PKI 必须包含验证新证书的根证书颁发机构证书。您可以将所需的证书颁发机构证书作为 gNOI 证书颁发机构捆绑包的一部分加载，也可以单独加载。如果验证失败，设备不会安装新证书。
gNOI 服务器仅支持一个用于 gNOI 服务的全局证书颁发机构。使用 gNOI CertificateManagement 服务加载 证书颁发机构 捆绑包时，以下语句适用：
-
服务
CertificateManagement始终使用ca-profile-group保留标识符gnoi-ca-bundle加载证书颁发机构证书捆绑包。 -
如果使用该
CertificateManagement服务加载证书颁发机构证书包，则设备将隐式使用相互身份验证。 -
如果服务发送加载新 证书颁发机构证书捆绑包的请求，则
CertificateManagement服务器将从设备中清除先前 CA 证书颁发机构捆绑包的证书并加载新证书。 -
如果使用该
CertificateManagement服务加载 证书颁发机构证书包，并且还在设备配置中显式配置了相互身份验证，则配置的语句优先。
因此，您最初可以在 gNOI 服务器上设置仅限服务器的身份验证，然后使用 Install() RPC 加载证书颁发机构证书。使用 gNOI 加载初始 证书颁发机构证书包时，设备将执行以下步骤：
- 在 Junos PKI 中添加 证书颁发机构证书。
- 使用
ca-profile-group标识gnoi-ca-bundle符在层次结构级别自动[edit security pki]配置 gNOI 证书颁发机构证书包。 - 交换机从纯服务器身份验证切换到相互身份验证。
[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服务器上加载证书颁发机构证书包，因为这会导致设备在操作过程中从纯服务器身份验证切换到相互身份验证。当身份验证模式发生变化时，网络设备必须重新启动 gRPC 堆栈，并且连接将丢失。如果流中断，客户端将无法完成轮换请求，并且设备将回滚到启动请求之前Rotate()的证书。
hot-reloading当身份验证模式在操作期间保持不变时，层次结构级别的[edit system services extension-service request-response grpc ssl]语句仅会在证书更新期间维护 gRPC 会话。例如，如果身份验证模式从纯服务器切换到相互身份验证，反之亦然，则客户端将断开连接。
支持的 RPC
表 1 概述了 Junos 设备上支持的CertificateManagement服务 RPC。
| RPC | 描述 | 在版本中引入 |
|---|---|---|
CanGenerateCSR() |
查询目标设备，以确定其是否可以生成具有指定密钥类型、密钥大小和证书类型的证书签名请求 （CSR）。支持的值：
如果 gNOI 服务器支持特定的密钥类型、密钥大小和证书类型，则返回 |
Junos OS 演化版 23.1R1 |
GenerateCSR() |
生成并返回证书签名请求 （CSR）。 |
Junos OS 演化版 22.2R1 |
GetCertificates() |
返回目标设备上加载的本地证书。 |
Junos OS 演化版 22.2R1 |
Install() |
通过创建 CSR 请求，基于 CSR 生成证书并使用新证书 ID 加载证书，在目标设备上加载新证书。 |
Junos OS 演化版 22.2R1 |
LoadCertificate() |
在目标设备上加载由证书颁发机构 （证书颁发机构） 签名的证书。 |
Junos OS 演化版 22.2R1 |
LoadCertificateAuthorityBundle() |
在目标设备上加载 证书颁发机构证书包。 |
Junos OS 演化版 22.2R1 |
RevokeCertificates() |
撤销目标设备上具有指定证书 ID 的证书。 |
Junos OS 演化版 23.1R1 |
Rotate() |
创建 CSR 请求，基于 CSR 生成证书，并使用现有证书 ID 加载证书，以替换目标设备上的现有证书。 |
Junos OS 演化版 22.2R1 |
网络设备配置
开始之前：
- 在网络设备上配置 gRPC 服务，如配置 gRPC 服务中所述。
- 配置网络管理系统以支持 gNOI 操作，如配置 gNOI 服务中所述。
对于在层次结构级别配置的 [edit system services http servers] gNOI 服务器，无需额外配置。
对于在层次结构级别配置的 [edit system services extension-service request-response grpc ssl] gNOI 服务器，您必须 use-pki 在同一层次结构级别配置 and hot-reloading 语句。在更新影响会话的证书时，需要该 hot-reloading 语句来维护 gRPC 会话。要配置语句，请执行以下操作：
安装证书
可以使用 CertificateManagement 服务 Install() RPC 在目标设备上加载新证书。使用该 Install() 操作安装新证书时，必须指定目标设备上尚不存在的新证书 ID。您还可以选择性地加载 证书颁发机构证书包，作为操作的一部分 Install() 。
作为操作的一部分 Install() ，设备将验证新证书。因此，Junos PKI 必须具有用于验证新证书的根 证书颁发机构证书。您可以将所需的 证书颁发机构证书作为操作的一部分 Install() 进行加载，如果 PKI 中尚未加载，则可以在操作之前单独加载该证书。
如果安装将用于 gRPC 会话身份验证的新本地证书，还必须更新设备上的 gRPC 服务器配置以使用新的证书 ID。
示例：安装证书
在此示例中，gNOI 服务器最初仅配置了本地证书，尚未配置为使用相互身份验证。gNOI 客户端使用 Install() RPC 在设备上加载新的本地证书和 证书颁发机构证书包。在 gNOI 服务器上加载证书颁发机构包后，服务器默认使用相互身份验证。证书颁发机构捆绑包包括客户机证书的根 证书颁发机构证书以及新服务器证书的根 证书颁发机构证书。
客户端执行 gnoi_cert_install_certificate_csr.py Python 应用程序，该应用程序执行以下操作：
- 请求目标生成 CSR。
- 根据 CSR 获取已签名的证书。
- 在目标网络设备上加载新的服务器证书、服务器的新根证书颁发机构证书和客户端的根证书颁发机构证书。
应用程序使用带有适当参数的消息来 InstallCertificateRequest 定义生成 CSR 和加载证书的请求。对于每个请求，应用程序使用 RPC 将 Install() 请求发送到网络设备。
gnoi_cert_install_certificate_csr.py应用程序导入grpc_channel模块以建立通道。配置 gNOI 服务中介绍了该grpc_channel模块。应用程序的参数存储在文件中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，获取已签名的证书，并在目标网络设备上加载新的服务器证书和证书颁发机构证书。
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.
安装新的服务器证书后，您必须将服务器配置为将该证书 ID 用于 gRPC 会话身份验证。更新 local-certificate 特定服务器配置的语句，该配置可能在以下层级之一进行配置：
-
[编辑系统服务扩展-服务请求-响应 gRPC SSL]
-
[编辑系统服务 http 服务器 name 服务器 tls]
例如：
user@gnoi-server> show configuration system services extension-service request-response grpc ssl port 32767; local-certificate gnoi-server1; hot-reloading; use-pki;
此外，由于操作 Install() 加载了新的证书颁发机构证书，因此设备隐式使用相互身份验证。因此，在建立通道时，所有后续 gRPC 会话都必须包含客户端的证书和密钥。
如果执行应用程序并提供服务器上已存在的证书 ID，则应用程序将返回 ALREADY_EXISTS 错误，因为该 Install() 操作需要新的证书 ID。
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}"
轮换证书
可以使用 CertificateManagement 服务 Rotate() RPC 替换目标设备上的现有证书。使用该 Rotate() 操作替换现有证书时，必须使用目标设备上已存在的证书 ID 加载证书。作为操作的一部分 Rotate() ，您还可以选择性地替换现有的 gNOI 证书颁发机构证书包。
该 Rotate() 操作与 Install() 操作类似，不同之处在于该 Rotate() 操作会替换现有证书，而不是安装新证书。此外，客户端必须验证更新后的证书是否有效，然后根据证书验证的成功或失败完成或取消 Rotate() 请求。
作为操作的一部分 Rotate() ，设备将验证新证书。因此，Junos PKI 必须具有用于验证新证书的根 证书颁发机构证书。您可以将所需的 证书颁发机构证书作为操作的一部分 Rotate() 进行加载，如果 PKI 中尚未加载，则可以在操作之前单独加载该证书。
示例：轮换证书
在此示例中，客户端执行 gnoi_cert_rotate_certificate_csr.py Python 应用程序，该应用程序执行以下操作：
- 请求目标生成 CSR。
- 根据 CSR 获取已签名的证书
- 替换目标网络设备上的节点证书和 gNOI 证书颁发机构包。
- 验证新证书。
- 完成操作
Rotate。
应用程序使用带有适当参数的消息来 RotateCertificateRequest 定义生成 CSR 和加载证书和证书颁发机构束的请求。对于每个请求，应用程序使用 Rotate() RPC 将请求发送到网络设备。为了使目标设备能够验证新的节点证书，应用程序会将现有证书颁发机构捆绑包替换为新的证书颁发机构捆绑包。该捆绑包包括验证节点证书所需的客户端证书颁发机构证书和证书颁发机构证书。
应用程序通过与网络设备创建新的 gRPC 会话并执行简单的 Time() RPC 来验证新证书是否有效，但您可以使用任何 RPC 测试会话身份验证。如果会话成功建立，应用程序将完成轮换请求，如果会话身份验证失败，则取消轮换请求。
gnoi_cert_rotate_certificate_csr.py应用程序导入grpc_channel模块以建立通道。配置 gNOI 服务中介绍了该grpc_channel模块。应用程序的参数存储在文件中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 参数是初始通道凭据所需的服务器的根证书颁发机构证书。参数 server_root_ca1 是与服务器的新证书相对应的根证书颁发机构证书。Junos PKI 必须具有新的根 证书颁发机构证书，才能在操作期间 Rotate() 验证新的本地证书。此外，验证新证书的 gRPC 会话的通道凭据使用此根证书颁发机构证书。尽管此示例对新旧证书颁发机构证书使用相同的根 CA 证书，但对于其他情况，这些证书可能有所不同。
执行应用
当客户端执行应用时，应用会请求 CSR，获取已签名的证书，并将替换证书和 证书颁发机构束加载到目标网络设备上。然后，应用程序使用执行简单 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 从目标设备中删除一个或多个证书。客户端包含一条 RevokeCertificatesRequest 消息，其中包含要撤销的证书 ID 列表。
当 gNOI 服务器收到 RevokeCertificates() 请求时，它会按如下方式处理列表中的每个证书 ID：
-
如果证书存在且吊销成功，设备将从文件系统和 Junos PKI 中移除证书，并将证书 ID 添加到成功吊销的证书列表中。
-
如果证书存在且吊销失败，则设备会在证书吊销错误列表中包含证书 ID 和失败原因。
-
如果证书不存在，设备会认为吊销操作成功，并将证书 ID 添加到成功吊销的证书列表中。
如果请求吊销用于当前会话的证书，则会话不会受到影响。
处理请求后，gNOI 服务器返回一条 RevokeCertificatesResponse 消息，其中包括：
-
成功吊销的证书 ID 列表。
-
包含证书 ID 和失败原因的吊销错误列表。
示例：吊销证书
在此示例中，客户端执行 gnoi_cert_revoke_certificates.py Python 应用程序，该应用程序会撤销服务器上的两个证书。第一个证书 ID 是设备上的有效标识符。第二个证书 ID 是设备上不存在的标识符。
应用程序使用带有适当参数的消息来 RevokeCertificatesRequest 定义请求。应用程序将 RevokeCertificates() RPC 发送到网络设备以执行操作。
gnoi_cert_revoke_certificates.py应用程序导入grpc_channel模块以建立通道。配置 gNOI 服务中介绍了该grpc_channel模块。应用程序的参数存储在文件中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()该操作会在操作过程中验证新证书。