Add unit tests for PIV commands + related fixes

README.md

@@ -16,7 +16,9 @@ The config file approach simplifies planning, setup and daily use while maintain
   - Fully within one process, does not invoke external CLI tools (except in unit tests)
 - Integrate daily operations under a single tool:
   - OpenSSH certificate creation and signing, including hardware token **sk-ed25519** and **sk-ecdsa** keys
-  - X.509 certificate creationg and signing (TLS, SSH, X.509)
+  - X.509 certificate creationg and signing
+  - TLS server cert creation
+  - Windows login certs for Yubikey with PIV
   - Password derivation for VMs etc.
 - Improved Secret Sharing ceremony vs. YubiHSM setup util (vs. yubihsm-setup)
   - password protected shares (optional)

hsm-conf.yml

@@ -559,8 +559,9 @@ piv:
     dc_cert_templates:
       "default":
         validity_days: 1825
-        attribs:
-            common_name: 'Example PIV DC D1'
+        basic_constraints:
+          ca: false         # End-entity certificate
+          path_len: null
         crl_distribution_points:
           urls:
             - "http://crl.example.local/piv.crl"  # Use CRL in LAN for PIV
@@ -578,6 +579,9 @@ piv:
     user_cert_templates:
       "default":
         validity_days: 365
+        basic_constraints:
+          ca: false
+          path_len: null
         attribs:
             common_name: 'Example PIV User U1'
         crl_distribution_points:

hsm_secrets/piv/__init__.py

@@ -245,8 +245,9 @@ def create_piv_cert(ctx: HsmSecretsCtx, user: str, template: str|None, subject:
 @click.option('--ca', '-c', required=False, help="CA ID (hex) or label, default: from config")
 @click.option('--out', '-o', required=False, type=click.Path(exists=False, dir_okay=False, resolve_path=True, allow_dash=False), help="Output filename, default: deduced from input")
 @click.option('--san', multiple=True, help="Additional (GeneralName) SANs")
+@click.option('--hostname', '-h', required=True, help="Hostname (CommonName) for the DC certificate")
 @click.option('--template', '-t', required=False, help="Template label, default: first template")
-def sign_dc_cert(ctx: HsmSecretsCtx, csr: click.File, validity: int, ca: str, out: str|None, san: List[str], template: str|None):
+def sign_dc_cert(ctx: HsmSecretsCtx, csr: click.File, validity: int, ca: str, out: str|None, san: List[str], hostname: str, template: str|None):
     """Sign a DC Kerberos PKINIT certificate for PIV"""
     csr_path = Path(csr.name)
     with csr_path.open('rb') as f:
@@ -276,6 +277,8 @@ def sign_dc_cert(ctx: HsmSecretsCtx, csr: click.File, validity: int, ca: str, ou
     if validity:
         x509_info.validity_days = validity
 
+    x509_info.attribs.common_name = hostname     # Override CN
+
     # Add explicitly provided SANs
     x509_info.subject_alt_name = x509_info.subject_alt_name or x509_info.SubjectAltName()
     for san_entry in san:
@@ -285,6 +288,10 @@ def sign_dc_cert(ctx: HsmSecretsCtx, csr: click.File, validity: int, ca: str, ou
             raise click.ClickException(f"Provided '{san_type.lower()}' is not a supported X509NameType")
         x509_info.subject_alt_name.names.setdefault(san_type_lower, []).append(san_value)  # type: ignore [arg-type]
 
+    # Add hostname to DNS SANs if not already there
+    if hostname not in (x509_info.subject_alt_name.names['dns'] or []):
+        x509_info.subject_alt_name.names['dns'] = [hostname] + list(x509_info.subject_alt_name.names['dns'] or [])
+
     # Create X509CertBuilder
     cert_builder = X509CertBuilder(ctx.conf, x509_info, csr_obj)
 

hsm_secrets/piv/piv_cert_checks.py

@@ -25,8 +25,8 @@ def _check_specific_subject_alternative_name(self, san: x509.SubjectAlternativeN
 
         if not has_dns:
             self._add_issue("SubjectAlternativeName does not include a DNS name", IssueSeverity.ERROR)
-        if not has_upn:
-            self._add_issue("SubjectAlternativeName does not include a UPN (User Principal Name)", IssueSeverity.WARNING)
+        if has_upn:
+            self._add_issue("DC certificates does not usually include a UPN (User Principal Name)", IssueSeverity.NOTICE)
 
     def _check_specific_subject_common_name_consistency(self, cn_value: str, san: x509.SubjectAlternativeName):
         dns_names = [name.value for name in san if isinstance(name, x509.DNSName)]
@@ -85,12 +85,6 @@ def _check_specific_subject_common_name_consistency(self, cn_value: str, san: x5
                 if email_username.lower() not in cn_value.lower():
                     self._add_issue(f"Subject CN '{cn_value}' does not appear to be related to email '{rfc822_name}'", IssueSeverity.NOTICE)
 
-        directory_name = next((name for name in san if isinstance(name, x509.DirectoryName)), None)
-        if directory_name:
-            dn_cn = next((attr for attr in repr(directory_name).split(',') if attr.startswith('CN=')), None)
-            if dn_cn and dn_cn.lower() != cn_value.lower():
-                self._add_issue(f"Subject CN '{cn_value}' does not match Directory Name CN '{dn_cn}'", IssueSeverity.NOTICE)
-
 
     def _check_subject_and_issuer(self):
         super()._check_subject_and_issuer()
@@ -107,8 +101,9 @@ def _check_policy_extensions(self):
         super()._check_policy_extensions()
         try:
             cert_policies = self.certificate.extensions.get_extension_for_class(x509.CertificatePolicies).value
-            has_piv_auth_policy = any(policy.policy_identifier.dotted_string == "2.16.840.1.101.3.2.1.3.13" for policy in cert_policies)
-            if not has_piv_auth_policy:
-                self._add_issue("Certificate does not include the PIV Authentication policy OID (2.16.840.1.101.3.2.1.3.13)", IssueSeverity.WARNING)
+            # US Federal PIV-I policy (not required for non-US-government PIVs)
+            #has_piv_auth_policy = any(policy.policy_identifier.dotted_string == "2.16.840.1.101.3.2.1.3.13" for policy in cert_policies)
+            #if not has_piv_auth_policy:
+            #    self._add_issue("Certificate does not include the PIV Authentication policy OID (2.16.840.1.101.3.2.1.3.13)", IssueSeverity.WARNING)
         except x509.ExtensionNotFound:
             pass  # Already handled in base class

hsm_secrets/x509/cert_builder.py

@@ -6,6 +6,8 @@
 from cryptography.x509.extensions import ExtensionTypeVar
 import cryptography.x509.oid as x509_oid
 
+import asn1crypto.core  # type: ignore [import]
+
 import datetime
 from datetime import timedelta
 import ipaddress
@@ -434,19 +436,28 @@ def _mk_name_attribs(self) -> x509.Name:
     def _mkext_alt_name(self) -> tuple[x509.SubjectAlternativeName, bool]:
         assert self.cert_def_info.subject_alt_name, "X509Info.subject_alt_name is missing"
         type_to_cls = {
-            "dns": (x509.DNSName, lambda n: n),
-            "ip": (x509.IPAddress, lambda n: ipaddress.ip_address(n)),
-            "rfc822": (x509.RFC822Name, lambda n: n),
-            "uri": (x509.UniformResourceIdentifier, lambda n: n),
-            "directory": (x509.DirectoryName, lambda n: n),
-            "registered_id": (x509.RegisteredID, lambda n: n),
-            "upn": (x509.OtherName, lambda n: x509.OtherName(type_id = x509.ObjectIdentifier("1.3.6.1.4.1.311.20.2.3"), value = n.encode('utf-16-le'))),
-            "oid": (x509.OtherName, lambda n: x509.OtherName(type_id = n.split("=", 1)[0], value = n.split("=", 1)[1].encode('utf-16-le')))
+            "dns": lambda n: x509.DNSName(n),
+            "ip": lambda n: x509.IPAddress(ipaddress.ip_address(n)),
+            "rfc822": lambda n: x509.RFC822Name(n),
+            "uri": lambda n: x509.UniformResourceIdentifier(n),
+            "directory": lambda n: x509.DirectoryName(x509.Name(parse_x500_dn_subject(n))),
+            "registered_id": lambda n: x509.RegisteredID(n),
+                "upn": lambda n: x509.OtherName(
+                    x509.ObjectIdentifier("1.3.6.1.4.1.311.20.2.3"),
+                    asn1crypto.core.OctetString(asn1crypto.core.UTF8String(n).dump()).dump()
+                ),
+            "oid": lambda n: x509.OtherName(
+                x509.ObjectIdentifier(n.split("=", 1)[0]),
+                asn1crypto.core.OctetString(asn1crypto.core.UTF8String(n.split("=", 1)[1]).dump()).dump()
+            )
         }
         san: List[x509.GeneralName] = []
         for san_type, names in (self.cert_def_info.subject_alt_name.names or {}).items():
-            dst_cls, dst_conv = type_to_cls[san_type]
-            san.extend([dst_cls(dst_conv(name)) for name in names])
+            try:
+                dst_conv = type_to_cls[san_type]
+                san.extend([dst_conv(name) for name in names])
+            except Exception as e:
+                raise ValueError(f"Failed to map SAN type '{san_type}' (names: '{str(names)}') to a class: {e}")
         return x509.SubjectAlternativeName(san), self.cert_def_info.subject_alt_name.critical
 
     def _mkext_key_usage(self) -> tuple[x509.KeyUsage, bool]:

hsm_secrets/x509/cert_checker.py

@@ -89,7 +89,7 @@ def _check_key_type_and_size(self):
             if public_key.key_size < 2048:
                 self._add_issue(f"RSA key size ({public_key.key_size}) is less than 2048 bits", IssueSeverity.ERROR)
             elif public_key.key_size < 3072:
-                self._add_issue(f"RSA key size ({public_key.key_size}) is less than 3072 bits", IssueSeverity.WARNING)
+                self._add_issue(f"RSA key size ({public_key.key_size}) is less than 3072 bits", IssueSeverity.NOTICE)
         elif isinstance(public_key, ec.EllipticCurvePublicKey):
             if public_key.curve.key_size < 256:
                 self._add_issue(f"EC key size ({public_key.curve.key_size}) is less than 256 bits", IssueSeverity.ERROR)
@@ -99,7 +99,7 @@ def _check_key_type_and_size(self):
     def _check_validity_period(self):
         max_validity = timedelta(days=398)
         if (self.certificate.not_valid_after_utc - self.certificate.not_valid_before_utc) > max_validity:
-            self._add_issue("Certificate validity period exceeds 398 days", IssueSeverity.WARNING)
+            self._add_issue("Certificate validity period exceeds 398 days", IssueSeverity.NOTICE)
 
     def _check_subject_and_issuer(self):
         if not self.certificate.subject:

requirements.txt

@@ -13,6 +13,7 @@ mnemonic
 pyescrypt
 
 pycryptodome
+asn1crypto
 
 yubihsm-ssh-tool @ git+https://github.com/YubicoLabs/yubihsm-ssh-tool@9cec861
 

run-tests.sh

@@ -143,6 +143,63 @@ test_tls_certificates() {
     [ -f $TEMPDIR/www-example-com.chain.pem ] || { echo "ERROR: Chain bundle not saved"; return 1; }
 }
 
+
+test_piv_user_certificate() {
+    setup
+
+    local output=$(run_cmd piv user-cert -u [email protected] --os-type windows --san "RFC822:[email protected]" --san "DIRECTORY:C=US,O=Organization,CN=test.user" --out $TEMPDIR/testuser-piv)
+    assert_success
+    echo "$output"
+    assert_not_grep "Cert errors" "$output"
+    assert_not_grep "Cert warnings" "$output"
+
+    [ -f $TEMPDIR/testuser-piv.key.pem ] || { echo "ERROR: Key not saved"; return 1; }
+    [ -f $TEMPDIR/testuser-piv.csr.pem ] || { echo "ERROR: CSR not saved"; return 1; }
+    [ -f $TEMPDIR/testuser-piv.cer.pem ] || { echo "ERROR: Certificate not saved"; return 1; }
+
+    local cert_output=$(openssl x509 -in $TEMPDIR/testuser-piv.cer.pem -text -noout)
+    assert_success
+    echo "$cert_output"
+    assert_grep "Subject:.*CN.*=.*[email protected]" "$cert_output"
+    assert_grep "X509v3 Subject Alternative Name:" "$cert_output"
+    assert_grep "test[.]user@example[.]com" "$cert_output"
+    assert_grep "Organization.*test[.]user" "$cert_output"
+    assert_grep "Key Usage: critical" "$cert_output"
+    assert_grep "Extended Key Usage" "$cert_output"
+    assert_grep "Smartcard" "$cert_output"
+    assert_grep "Client Authentication" "$cert_output"
+}
+
+test_piv_dc_certificate() {
+    setup
+
+    # Generate a CSR for the DC certificate
+    openssl ecparam -genkey -name secp384r1 -out $TEMPDIR/dc.key.pem
+    openssl req -new -key $TEMPDIR/dc.key.pem -out $TEMPDIR/dc.csr.pem -subj "/CN=dc01.example.com"
+
+    local output=$(run_cmd piv sign-dc-cert $TEMPDIR/dc.csr.pem --hostname "dc01.example.com" --san "DNS:dc.example.com" --out $TEMPDIR/dc.cer.pem)
+    assert_success
+    echo "$output"
+    assert_not_grep "Cert errors" "$output"
+    assert_not_grep "Cert warnings" "$output"
+
+    [ -f $TEMPDIR/dc.cer.pem ] || { echo "ERROR: Signed certificate not saved"; return 1; }
+
+    local cert_output=$(openssl x509 -in $TEMPDIR/dc.cer.pem -text -noout)
+    assert_success
+    echo "$cert_output"
+    assert_grep "Subject:.*CN.*=.*dc01.example.com" "$cert_output"
+    assert_grep "X509v3 Subject Alternative Name:" "$cert_output"
+    assert_grep "DNS:dc01.example.com" "$cert_output"
+    assert_grep "DNS:dc.example.com" "$cert_output"
+    assert_grep "Key Usage: critical" "$cert_output"
+    assert_grep "Digital Signature, Key Encipherment" "$cert_output"
+    assert_grep "Extended Key Usage:" "$cert_output"
+    assert_grep "Server Authentication" "$cert_output"
+    assert_grep "KDC" "$cert_output"
+}
+
+
 test_crl_commands() {
     setup
 
@@ -366,6 +423,8 @@ run_test test_password_derivation
 run_test test_wrapped_backup
 run_test test_ssh_user_certificates
 run_test test_ssh_host_certificates
+run_test test_piv_user_certificate
+run_test test_piv_dc_certificate
 run_test test_logging_commands
 
 echo "All tests passed successfully!"