diff --git a/poetry.lock b/poetry.lock index 9fe24bc1..2e74f94b 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1128,14 +1128,14 @@ files = [ [[package]] name = "nethsm" -version = "3.0.0" +version = "3.1.0rc1" description = "Python Library to manage NetHSM(s)" optional = false python-versions = ">=3.9" groups = ["main"] files = [ - {file = "nethsm-3.0.0-py3-none-any.whl", hash = "sha256:146fc8e09a0ceba04b7333b16be29742610d30e044a3e98457a1de2fa2e94b31"}, - {file = "nethsm-3.0.0.tar.gz", hash = "sha256:0bb1e217b2d14157c7740f43d44bfda45fcd1b95a3baf3806f65b8d08c440000"}, + {file = "nethsm-3.1.0rc1-py3-none-any.whl", hash = "sha256:baa78c83e71fc62e3f86eb75d0113e6959732295603e1d0a803db74429c817b1"}, + {file = "nethsm-3.1.0rc1.tar.gz", hash = "sha256:a825d27c55c1e54576ba0a7c91677114b79a8c3902cecfed305f20f50ec59ebd"}, ] [package.dependencies] @@ -1936,4 +1936,4 @@ pcsc = ["pyscard"] [metadata] lock-version = "2.1" python-versions = ">= 3.11, <3.15" -content-hash = "e8c480ebafb3fddfef778d596cd1d68fe85d0142afc32b0caaaa573178f8e5e9" +content-hash = "ff1d7d8fd1ad5cfbf86afb17127135e087380c9a411c6a3912a0b9080529332f" diff --git a/poetry.toml b/poetry.toml index 737d6cf3..ac42afee 100644 --- a/poetry.toml +++ b/poetry.toml @@ -1,4 +1,4 @@ [solver] min-release-age = 7 # certify: 2026.10.19 for https://github.com/certifi/python-certifi/security/advisories/GHSA-6q48-6c7w-9gw2 -min-release-age-exclude = ["certifi", "nitrokey"] +min-release-age-exclude = ["certifi", "nethsm", "nitrokey"] diff --git a/pynitrokey/cli/nethsm.py b/pynitrokey/cli/nethsm.py index a061e51c..b5b7cdfb 100644 --- a/pynitrokey/cli/nethsm.py +++ b/pynitrokey/cli/nethsm.py @@ -1260,65 +1260,88 @@ def set_unattended_boot(ctx: Context, status: str) -> None: print(f"Updated the unattended boot configuration for NetHSM {nethsm.host}") -def get_api_or_key_id(api: bool, key_id: Optional[str]) -> tuple[bool, Optional[str]]: +def get_api_or_cluster_or_key_id( + api: bool, cluster: bool, key_id: Optional[str] +) -> tuple[bool, bool, Optional[str]]: """Helper method for operations that can be executed either for the API - certificate or for the certificate stored for a key.""" + certificate, the cluster certificate or for the certificate stored for a key.""" if api and key_id: raise click.ClickException("--api and --key-id are mutually exclusive") - - if not api and not key_id: - choice = click.Choice(["api", "key"], case_sensitive=False) - method = prompt("For stored key or for NetHSM TLS interface?", type=choice) + if cluster and key_id: + raise click.ClickException("--cluster and --key-id are mutually exclusive") + if api and cluster: + raise click.ClickException("--api and --cluster are mutually exclusive") + + if not api and not key_id and not cluster: + choice = click.Choice(["api", "cluster", "key"], case_sensitive=False) + method = prompt( + "For stored key, for NetHSM TLS interface or for cluster peer traffic?", type=choice + ) if method == "api": api = True + elif method == "cluster": + cluster = True elif method == "key": key_id = prompt("Key ID") else: raise ValueError("Unexpected method") - return (api, key_id) + return (api, cluster, key_id) @nethsm.command() @click.option("-a", "--api", is_flag=True, help="Set the certificate for the NetHSM TLS interface") +@click.option( + "-c", "--cluster", is_flag=True, help="Set the certificate for the cluster peer traffic" +) @click.option("-k", "--key-id", help="The ID of the key to set the certificate for") @click.argument("filename") @click.pass_context -def set_certificate(ctx: Context, api: bool, key_id: Optional[str], filename: str) -> None: +def set_certificate( + ctx: Context, api: bool, cluster: bool, key_id: Optional[str], filename: str +) -> None: """Set a certificate on the NetHSM. If the --api option is set, the certificate used for the NetHSM TLS interface - is set. If the --key-id option is set, the certificate for a key stored on - the NetHSM is set. + is set. If the --cluster option is set, the certificate used for the cluster + peer traffic is set. If the --key-id option is set, the certificate for a key + stored on the NetHSM is set. This command requires authentication as a user with the Administrator role.""" - api, key_id = get_api_or_key_id(api, key_id) + api, cluster, key_id = get_api_or_cluster_or_key_id(api, cluster, key_id) with connect(ctx) as nethsm: with open(filename, "rb") as f: if key_id: nethsm.set_key_certificate(key_id, f) print(f"Updated the certificate for key {key_id} on NetHSM {nethsm.host}") - else: + elif api: nethsm.set_certificate(f) print(f"Updated the API certificate for NetHSM {nethsm.host}") + elif cluster: + nethsm.set_cluster_certificate(f) + print(f"Updated the cluster certificate for NetHSM {nethsm.host}") @nethsm.command() @click.option("-a", "--api", is_flag=True, help="Get the certificate for the NetHSM TLS interface") +@click.option( + "-c", "--cluster", is_flag=True, help="Get the certificate for the cluster peer traffic" +) @click.option("-k", "--key-id", help="The ID of the key to get the certificate for") @click.pass_context -def get_certificate(ctx: Context, api: bool, key_id: Optional[str]) -> None: +def get_certificate(ctx: Context, api: bool, cluster: bool, key_id: Optional[str]) -> None: """Get a certificate from the NetHSM. If the --api option is set, the certificate used for the NetHSM TLS interface - is queried. If the --key-id option is set, the certificate for a key stored on - the NetHSM is queried. + is queried. If the --cluster option is set, the certificate used for the + cluster peer traffic is queried. If the --key-id option is set, the + certificate for a key stored on the NetHSM is queried. This command requires authentication as a user with the Administrator role. The certificate for a key can also be queried by a user with the Operator role.""" - api, key_id = get_api_or_key_id(api, key_id) + api, cluster, key_id = get_api_or_cluster_or_key_id(api, cluster, key_id) with connect(ctx) as nethsm: if key_id: cert = nethsm.get_key_certificate(key_id) @@ -1326,8 +1349,10 @@ def get_certificate(ctx: Context, api: bool, key_id: Optional[str]) -> None: print(cert.decode()) except UnicodeError: print(cert) - else: + elif api: print(nethsm.get_certificate()) + elif cluster: + print(nethsm.get_cluster_certificate()) @nethsm.command() @@ -1345,6 +1370,7 @@ def delete_certificate(ctx: Context, key_id: str) -> None: @nethsm.command() @click.option("-a", "--api", is_flag=True, help="Generate a CSR for the NetHSM TLS interface") +@click.option("-c", "--cluster", is_flag=True, help="Generate a CSR for the cluster peer traffic") @click.option("-k", "--key-id", help="The ID of the key to generate the CSR for") @click.option("--country", default="", prompt=True, help="The country name") @click.option("--state-or-province", default="", prompt=True, help="The state or province name") @@ -1361,6 +1387,7 @@ def delete_certificate(ctx: Context, key_id: str) -> None: def csr( ctx: Context, api: bool, + cluster: bool, key_id: Optional[str], country: str, state_or_province: str, @@ -1375,8 +1402,9 @@ def csr( """Generate a certificate signing request. If the --api option is set, the CSR is generated for the NetHSM, for - example to replace the self-signed initial certificate. If the --key-id - option is set, the CSR is generated for a key stored on the NetHSM. + example to replace the self-signed initial certificate. If the --cluster + option is set, the CSR is generated for the cluster peer traffic. If the + --key-id option is set, the CSR is generated for a key stored on the NetHSM. By default, the certificate contains a Subject Alternative Name (SAN) extension with the common name. If the --no-san option is set, the @@ -1397,7 +1425,7 @@ def csr( if no_san: subject_alt_names = [] - api, key_id = get_api_or_key_id(api, key_id) + api, cluster, key_id = get_api_or_cluster_or_key_id(api, cluster, key_id) with connect(ctx) as nethsm: if key_id: csr = nethsm.key_csr( @@ -1411,7 +1439,7 @@ def csr( email_address=email_address, subject_alt_names=subject_alt_names, ) - else: + elif api: csr = nethsm.csr( country=country, state_or_province=state_or_province, @@ -1422,6 +1450,17 @@ def csr( email_address=email_address, subject_alt_names=subject_alt_names, ) + elif cluster: + csr = nethsm.cluster_csr( + country=country, + state_or_province=state_or_province, + locality=locality, + organization=organization, + organizational_unit=organizational_unit, + common_name=common_name, + email_address=email_address, + subject_alt_names=subject_alt_names, + ) print(csr) @@ -2010,6 +2049,42 @@ def get_cluster_ca_certificate(ctx: Context) -> None: print(nethsm.get_cluster_ca_certificate()) +@nethsm.command() +@click.pass_context +def get_cluster_public_key(ctx: Context) -> None: + with connect(ctx) as nethsm: + print(nethsm.get_cluster_public_key()) + + +@nethsm.command() +@click.option( + "type", + "-t", + "--type", + type=TYPE_TLS_KEY_TYPE, + prompt=True, + help="The type for the generated key", +) +@click.option("-l", "--length", type=int, help="The length of the generated key") +@click.pass_context +def generate_cluster_key(ctx: Context, type: str, length: Optional[int]) -> None: + """Generate key pair for the cluster peer traffic. + + This command requires authentication as a user with the Administrator + role.""" + key_type = nethsm_sdk.TlsKeyType.from_string(type) + if key_type == nethsm_sdk.TlsKeyType.RSA: + if not length: + length = click.prompt("Length", type=int) + else: + if length: + raise click.ClickException("-l/--length may only be set for RSA keys") + + with connect(ctx) as nethsm: + nethsm.generate_cluster_key(key_type, length) + print(f"Key for cluster peer traffic generated on NetHSM {nethsm.host}") + + @nethsm.command() @click.argument("filename") @click.pass_context @@ -2017,3 +2092,4 @@ def set_cluster_ca_certificate(ctx: Context, filename: str) -> None: with connect(ctx) as nethsm: with open(filename, "rb") as f: nethsm.set_cluster_ca_certificate(f) + print(f"Updated the cluster CA certificate for NetHSM {nethsm.host}") diff --git a/pyproject.toml b/pyproject.toml index 0c12494f..2e9dcad2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,7 +29,7 @@ dependencies = [ "hidapi ==0.14.0.post2 ; sys_platform == 'linux'", "intelhex >=2.3, <3", "libusb1 >=3, <4", - "nethsm >=3, <4", + "nethsm ==3.1.0rc1", "nitrokey >=0.5, <0.6", "nkdfu >=0.2, <0.3", "pyusb >=1.2, <2", diff --git a/tests/test_nethsm.py b/tests/test_nethsm.py index 0381e4d5..102429e9 100644 --- a/tests/test_nethsm.py +++ b/tests/test_nethsm.py @@ -1,11 +1,20 @@ import base64 +import datetime import io +import ipaddress import re +import secrets import shlex from contextlib import redirect_stderr, redirect_stdout +from pathlib import Path +from tempfile import TemporaryDirectory from click import Command, Context from conftest import UserData +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.x509.oid import NameOID from nethsm import NetHSM, Role from utilities import add_user @@ -23,6 +32,109 @@ ) +class CA: + """ + CA implementation based on the cryptography documentation: + https://cryptography.io/en/latest/x509/tutorial/#creating-a-ca-hierarchy + """ + + def __init__(self) -> None: + self.ca_key = ec.generate_private_key(ec.SECP256R1()) + self.int_key = ec.generate_private_key(ec.SECP256R1()) + self.ca_id = secrets.token_hex(8) + + now = datetime.datetime.now(datetime.timezone.utc) + key_usage = x509.KeyUsage( + digital_signature=True, + content_commitment=False, + key_encipherment=False, + data_encipherment=False, + key_agreement=False, + key_cert_sign=True, + crl_sign=True, + encipher_only=False, + decipher_only=False, + ) + + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, f"NetHSM Test CA {self.ca_id}")]) + builder = x509.CertificateBuilder() + builder = builder.subject_name(name) + builder = builder.public_key(self.ca_key.public_key()) + builder = builder.issuer_name(name) + builder = builder.serial_number(x509.random_serial_number()) + builder = builder.not_valid_before(now) + builder = builder.not_valid_after(now + datetime.timedelta(days=1)) + builder = builder.add_extension( + x509.BasicConstraints(ca=True, path_length=None), critical=True + ) + builder = builder.add_extension(key_usage, critical=True) + builder = builder.add_extension( + x509.SubjectKeyIdentifier.from_public_key(self.ca_key.public_key()), critical=False + ) + self.ca_cert = builder.sign(self.ca_key, hashes.SHA256()) + + def sign(self, csr: x509.CertificateSigningRequest) -> bytes: + issuer_cert = self.ca_cert + issuer_key = self.ca_key + + now = datetime.datetime.now(datetime.timezone.utc) + ip = x509.IPAddress(ipaddress.IPv4Address("127.0.0.1")) + + builder = x509.CertificateBuilder() + builder = builder.subject_name(csr.subject) + builder = builder.issuer_name(issuer_cert.subject) + builder = builder.public_key(csr.public_key()) + builder = builder.serial_number(x509.random_serial_number()) + builder = builder.not_valid_before(now) + builder = builder.not_valid_after(now + datetime.timedelta(days=1)) + builder = builder.add_extension(x509.SubjectAlternativeName([ip]), critical=False) + builder = builder.add_extension( + x509.BasicConstraints(ca=False, path_length=None), critical=True + ) + builder = builder.add_extension( + x509.KeyUsage( + digital_signature=True, + content_commitment=False, + key_encipherment=True, + data_encipherment=False, + key_agreement=False, + key_cert_sign=False, + crl_sign=True, + encipher_only=False, + decipher_only=False, + ), + critical=True, + ) + builder = builder.add_extension( + x509.ExtendedKeyUsage( + [x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH, x509.oid.ExtendedKeyUsageOID.SERVER_AUTH] + ), + critical=False, + ) + builder = builder.add_extension( + x509.SubjectKeyIdentifier.from_public_key(csr.public_key()), critical=False + ) + builder = builder.add_extension( + x509.AuthorityKeyIdentifier.from_issuer_subject_key_identifier( + issuer_cert.extensions.get_extension_for_class(x509.SubjectKeyIdentifier).value + ), + critical=False, + ) + cert = builder.sign(issuer_key, hashes.SHA256()) + + # sanity check + store = x509.verification.Store([self.ca_cert]) + verifier = x509.verification.PolicyBuilder().store(store).build_server_verifier(ip) + verifier.verify(cert, intermediates=[]) + + cert_bytes = cert.public_bytes(serialization.Encoding.PEM) + return cert_bytes + + @property + def ca_cert_pem(self) -> bytes: + return self.ca_cert.public_bytes(serialization.Encoding.PEM) + + def get_context(nethsm: NetHSM) -> Context: config = Config( host=nethsm.host, @@ -141,3 +253,41 @@ def test_nethsm_move_key(nethsm: NetHSM) -> None: cmd = "move-key testkeymoveold testkeymovenew" result = run_command(ctx, cmd) assert f"Key testkeymoveold moved to testkeymovenew on NetHSM {nethsm.host}\n" == result + + +def test_nethsm_cluster_ca_cert(nethsm: NetHSM) -> None: + ctx = get_context(nethsm) + + cmd = "generate-cluster-key --type rsa --length 4096" + result = run_command(ctx, cmd) + assert f"Key for cluster peer traffic generated on NetHSM {nethsm.host}\n" == result + + cmd = "get-cluster-public-key" + result = run_command(ctx, cmd) + public_key = serialization.load_pem_public_key(result.encode()) + + cmd = ( + "csr --cluster --country DE --state-or-province Brandenburg --locality Teltow --organization Nitrokey " + "--organizational-unit Test --common-name 'NetHSM cluster test' --email-address test@example.com" + ) + result = run_command(ctx, cmd) + csr = x509.load_pem_x509_csr(result.encode()) + assert csr.public_key() == public_key + + ca = CA() + cert = ca.sign(csr) + + with TemporaryDirectory() as d: + path = Path(d) + cert_path = path / "cert.pem" + cert_path.write_bytes(cert) + ca_path = path / "ca.pem" + ca_path.write_bytes(ca.ca_cert_pem) + + cmd = f"set-certificate --cluster {cert_path}" + result = run_command(ctx, cmd) + assert f"Updated the cluster certificate for NetHSM {nethsm.host}\n" == result + + cmd = f"set-cluster-ca-certificate {ca_path}" + result = run_command(ctx, cmd) + assert f"Updated the cluster CA certificate for NetHSM {nethsm.host}\n" == result