1493 lines
56 KiB
Diff
1493 lines
56 KiB
Diff
From 3d63c2bc1ec7e89fe2ce6360ddd72748a6d1e1c5 Mon Sep 17 00:00:00 2001
|
||
From: James Chapman <jachapma@redhat.com>
|
||
Date: Thu, 5 Feb 2026 15:19:07 +0000
|
||
Subject: [PATCH] Issue 6951 - Dynamic Certificate refresh phase 4 - Update
|
||
lib389 and dsconf (#7171)
|
||
|
||
Desciption:
|
||
Add the CertManager abstraction layer and DynamicCerts backend module. Enhance
|
||
the NssSsl backend to support importing PKCS#12 containers, containing cert and
|
||
private key. Update dsconf to support new PKCS#12-related args, allowing users
|
||
to supply passwords via text, stdin, or file.
|
||
|
||
Fix:
|
||
- Introduce CertManager abstraction layer for uniform cert management.
|
||
- Implement DynamicCerts backend with add/list/delete operations.
|
||
- Extend NssSsl.add_cert to handle PKCS#12 files with passwords via text, stdin, or file.
|
||
- Update dsconf CLI to accept PKCS#12 arguments (--pkcs12-pin-text, --pkcs12-pin-stdin, --pkcs12-pin-path).
|
||
|
||
Relates: https://github.com/389ds/389-ds-base/issues/6951
|
||
|
||
Reviewed by: @progier389, @mreynolds389, @droideck (Thank you)
|
||
---
|
||
.../tests/suites/clu/ca_cert_bundle_test.py | 5 +-
|
||
.../tests/suites/clu/dsctl_tls_test.py | 4 +-
|
||
.../suites/tls/tls_import_ca_chain_test.py | 2 +-
|
||
src/lib389/lib389/cert_manager.py | 145 ++++++
|
||
src/lib389/lib389/cli_conf/security.py | 252 ++++++----
|
||
src/lib389/lib389/dyncerts.py | 458 ++++++++++++++++++
|
||
src/lib389/lib389/nss_ssl.py | 196 ++++++--
|
||
src/lib389/lib389/utils.py | 79 ++-
|
||
8 files changed, 987 insertions(+), 154 deletions(-)
|
||
create mode 100644 src/lib389/lib389/cert_manager.py
|
||
create mode 100644 src/lib389/lib389/dyncerts.py
|
||
|
||
diff --git a/dirsrvtests/tests/suites/clu/ca_cert_bundle_test.py b/dirsrvtests/tests/suites/clu/ca_cert_bundle_test.py
|
||
index 1d2ccc5a6..40ab0e093 100644
|
||
--- a/dirsrvtests/tests/suites/clu/ca_cert_bundle_test.py
|
||
+++ b/dirsrvtests/tests/suites/clu/ca_cert_bundle_test.py
|
||
@@ -83,7 +83,6 @@ jBqLRMRQN4FvzuCZiMl/DwJv4yhAZ8hylYjRjqjY/fEPvhvJRncPVy8z
|
||
-----END CERTIFICATE-----
|
||
"""
|
||
|
||
-
|
||
def test_ca_cert_bundle(topo):
|
||
"""Test we can add a CAS certificate bundle
|
||
|
||
@@ -106,6 +105,8 @@ def test_ca_cert_bundle(topo):
|
||
|
||
"""
|
||
inst = topo.standalone
|
||
+ # Init NSS DB for dyncerts
|
||
+ inst.enable_tls()
|
||
lc = LogCapture()
|
||
|
||
# Create PEM file with 2 CA certs
|
||
@@ -119,6 +120,8 @@ def test_ca_cert_bundle(topo):
|
||
args = FakeArgs()
|
||
args.name = ['CA_CERT_1', 'CA_CERT_2']
|
||
args.file = pem_file
|
||
+ # Allow CA certs that fail verification (self-signed)
|
||
+ args.force = True
|
||
cacert_add(inst, DEFAULT_SUFFIX, log, args)
|
||
|
||
# List CA certs
|
||
diff --git a/dirsrvtests/tests/suites/clu/dsctl_tls_test.py b/dirsrvtests/tests/suites/clu/dsctl_tls_test.py
|
||
index 9ce5b4e2a..df55b0f83 100644
|
||
--- a/dirsrvtests/tests/suites/clu/dsctl_tls_test.py
|
||
+++ b/dirsrvtests/tests/suites/clu/dsctl_tls_test.py
|
||
@@ -57,11 +57,11 @@ def test_tls_command_returns_error_text(topo):
|
||
# dsctl localhost tls import-ca
|
||
try:
|
||
invalid_file = topo.standalone.confdir + '/dse.ldif'
|
||
- tls.add_cert(nickname="bad", input_file=invalid_file)
|
||
+ tls.add_cert(nickname="bad", cert_file=invalid_file)
|
||
assert False
|
||
except ValueError as e:
|
||
assert '255' not in str(e)
|
||
- assert 'Unable to load PEM file' in str(e)
|
||
+ assert 'Unable to load certificate' in str(e)
|
||
|
||
# dsctl localhost tls import-server-cert
|
||
try:
|
||
diff --git a/dirsrvtests/tests/suites/tls/tls_import_ca_chain_test.py b/dirsrvtests/tests/suites/tls/tls_import_ca_chain_test.py
|
||
index 0be3a9aaa..4714703f0 100644
|
||
--- a/dirsrvtests/tests/suites/tls/tls_import_ca_chain_test.py
|
||
+++ b/dirsrvtests/tests/suites/tls/tls_import_ca_chain_test.py
|
||
@@ -38,7 +38,7 @@ def test_tls_import_chain(topology_st):
|
||
tls.reinit()
|
||
|
||
with pytest.raises(ValueError):
|
||
- tls.add_cert(nickname='CA_CHAIN_1', input_file=CA_CHAIN_FILE)
|
||
+ tls.add_cert(nickname='CA_CHAIN_1', cert_file=CA_CHAIN_FILE)
|
||
|
||
with pytest.raises(ValueError):
|
||
tls.import_rsa_crt(crt=CRT_CHAIN_FILE)
|
||
diff --git a/src/lib389/lib389/cert_manager.py b/src/lib389/lib389/cert_manager.py
|
||
new file mode 100644
|
||
index 000000000..91c9c952f
|
||
--- /dev/null
|
||
+++ b/src/lib389/lib389/cert_manager.py
|
||
@@ -0,0 +1,145 @@
|
||
+# --- BEGIN COPYRIGHT BLOCK ---
|
||
+# Copyright (C) 2026 Red Hat, Inc.
|
||
+# All rights reserved.
|
||
+#
|
||
+# License: GPL (version 3 or any later version).
|
||
+# See LICENSE for details.
|
||
+# --- END COPYRIGHT BLOCK ---
|
||
+
|
||
+import os
|
||
+import logging
|
||
+from typing import Optional
|
||
+from lib389 import DirSrv
|
||
+from lib389.nss_ssl import NssSsl
|
||
+from lib389.dyncerts import DynamicCerts
|
||
+from lib389.config import RSA
|
||
+
|
||
+log = logging.getLogger(__name__)
|
||
+
|
||
+class CertManager:
|
||
+ """
|
||
+ Certificate manager for 389 Directory Server.
|
||
+
|
||
+ Automatically selects DynamicCerts backend if available via LDAPI,
|
||
+ otherwise falls back to NSS DB.
|
||
+ """
|
||
+
|
||
+ def __init__(self, instance):
|
||
+ """
|
||
+ Initialise a CertManager object, which selects the appropriate
|
||
+ certificate handler (DynamicCert or NSS).
|
||
+
|
||
+ :param instance: DirSrv instance
|
||
+ :raises ValueError: If instance is None
|
||
+ """
|
||
+ if not isinstance(instance, DirSrv):
|
||
+ raise ValueError("A DirSrv instance is required")
|
||
+
|
||
+ self.dirsrv = instance
|
||
+
|
||
+ if self.dirsrv.status():
|
||
+ self.cert_handler = DynamicCerts(instance=self.dirsrv)
|
||
+ else:
|
||
+ self.cert_handler = NssSsl(dirsrv=self.dirsrv)
|
||
+
|
||
+ self.cert_handler_name = type(self.cert_handler).__name__
|
||
+
|
||
+ def list_certs(self):
|
||
+ """
|
||
+ Return a list of all certificates exposed by the backend.
|
||
+
|
||
+ :return: list of certificate dictionaries or tuples
|
||
+ """
|
||
+ return self.cert_handler.list_certs()
|
||
+
|
||
+ def list_ca_certs(self):
|
||
+ """
|
||
+ Return a list of all CA certificates exposed by the backend.
|
||
+
|
||
+ :return: list of CA certificate dictionaries or tuples
|
||
+ """
|
||
+ return self.cert_handler.list_ca_certs()
|
||
+
|
||
+ def get_cert(self, nickname: str):
|
||
+ """
|
||
+ Get certificate details by nickname/CN.
|
||
+
|
||
+ :param nickname: Certificate nickname
|
||
+ :return: backend certificate object or None
|
||
+ """
|
||
+ return self.cert_handler.get_cert_details(nickname=nickname)
|
||
+
|
||
+ def del_cert(self, nickname: str):
|
||
+ """
|
||
+ Delete a certificate by nickname.
|
||
+
|
||
+ :param nickname: Certificate nickname
|
||
+ :raises ValueError: If the nickname is empty.
|
||
+ """
|
||
+ if not nickname:
|
||
+ raise ValueError("Certificate nickname cannot be empty")
|
||
+
|
||
+ self.cert_handler.del_cert(nickname)
|
||
+
|
||
+ def add_cert(
|
||
+ self,
|
||
+ cert_file: str,
|
||
+ nickname: str,
|
||
+ pkcs12_password: Optional[str] = None,
|
||
+ primary: bool = False,
|
||
+ ca: bool = False,
|
||
+ force: bool = False
|
||
+ ):
|
||
+ """
|
||
+ Add or replace a certificate.
|
||
+
|
||
+ :param cert_file: Path to certificate file (PEM, DER, or PKCS#12)
|
||
+ :param nickname: Certificate nickname/CN
|
||
+ :param pkcs12_password: Password for PKCS#12, if any
|
||
+ :param primary: Set as the server's primary SSL certificate
|
||
+ :param ca: Whether this certificate is a CA certificate
|
||
+ :param force: Force the addition of a certificate that cannot be verified
|
||
+ :raises ValueError: If cert_file or nickname is invalid
|
||
+ """
|
||
+ if not os.path.isfile(cert_file):
|
||
+ raise ValueError(f"Certificate file not found: {cert_file}")
|
||
+ if not nickname:
|
||
+ raise ValueError("Certificate nickname must not be empty")
|
||
+
|
||
+ self.cert_handler.add_cert(
|
||
+ nickname=nickname,
|
||
+ cert_file=cert_file,
|
||
+ pkcs12_password=pkcs12_password,
|
||
+ ca=ca,
|
||
+ force=force
|
||
+ )
|
||
+
|
||
+ if primary:
|
||
+ try:
|
||
+ RSA(self.dirsrv).set("nsSSLPersonalitySSL", nickname)
|
||
+ log.info(f"Set certificate '{nickname}' as primary SSL certificate.")
|
||
+ except Exception as e:
|
||
+ log.error(f"Failed to set primary SSL cert '{nickname}': {e}")
|
||
+ raise
|
||
+
|
||
+ def add_ca_cert(self, cert_file: str, nickname: str, force: bool = False):
|
||
+ """
|
||
+ Add one or more CA certificates from a PEM bundle or single DER.
|
||
+
|
||
+ :param cert_file: Path to certificate file (PEM, DER)
|
||
+ :param nickname: Certificate nickname
|
||
+ :raises ValueError: If file is missing
|
||
+ """
|
||
+ if not os.path.exists(cert_file):
|
||
+ raise ValueError(f"Certificate file does not exist: {cert_file}")
|
||
+
|
||
+ self.cert_handler.add_ca_cert(cert_file, nickname, force=force)
|
||
+
|
||
+ def edit_cert_trust(self, nickname: str, trust_flags: str):
|
||
+ """
|
||
+ Edit trust flags on an existing certificate.
|
||
+
|
||
+ :param nickname: Certificate nickname
|
||
+ :param trust_flags: NSS style trust flag triplet, e.g. 'CT,,'
|
||
+ """
|
||
+ return self.cert_handler.edit_cert_trust(nickname=nickname, trust_flags=trust_flags)
|
||
diff --git a/src/lib389/lib389/cli_conf/security.py b/src/lib389/lib389/cli_conf/security.py
|
||
index eba138feb..18e444c4c 100644
|
||
--- a/src/lib389/lib389/cli_conf/security.py
|
||
+++ b/src/lib389/lib389/cli_conf/security.py
|
||
@@ -9,8 +9,10 @@
|
||
from collections import OrderedDict, namedtuple
|
||
import json
|
||
import os
|
||
+import sys
|
||
from lib389.config import Config, Encryption, RSA
|
||
-from lib389.nss_ssl import NssSsl, CERT_NAME, CA_NAME
|
||
+from lib389.nss_ssl import NssSsl
|
||
+from lib389.cert_manager import CertManager
|
||
from lib389.cli_base import _warn, CustomHelpFormatter
|
||
|
||
|
||
@@ -140,6 +142,52 @@ def _security_generic_toggle_parsers(parent, cls, attr, help_pattern):
|
||
return list(map(add_parser, ('Enable', 'Disable'), ('on', 'off')))
|
||
|
||
|
||
+def _resolve_pkcs12_password(args):
|
||
+ if args.pkcs12_pin_text:
|
||
+ return args.pkcs12_pin_text
|
||
+
|
||
+ if args.pkcs12_pin_stdin:
|
||
+ return sys.stdin.readline().rstrip("\n")
|
||
+
|
||
+ if args.pkcs12_pin_path:
|
||
+ with open(args.pkcs12_pin_path) as f:
|
||
+ return f.read().rstrip("\n")
|
||
+
|
||
+ return None
|
||
+
|
||
+def _dump_cert(cert, json_output: bool = False, log = None):
|
||
+ """
|
||
+ Print or return a certificate's details in text or JSON format.
|
||
+
|
||
+ :param cert: dict describing a certificate
|
||
+ :param json_output: If True print JSON else print text
|
||
+ :param log: Optional logger to output text
|
||
+ """
|
||
+ if not isinstance(cert, dict):
|
||
+ raise TypeError(f"Expected dict, got {type(cert)}")
|
||
+
|
||
+ if json_output:
|
||
+ return {
|
||
+ "type": "certificate",
|
||
+ "attrs": {
|
||
+ "nickname": cert["cn"],
|
||
+ "subject": cert["subject"],
|
||
+ "issuer": cert["issuer"],
|
||
+ "expires": cert["expires"],
|
||
+ "flags": cert["trust_flags"],
|
||
+ }
|
||
+ }
|
||
+ else:
|
||
+ msg = (
|
||
+ f"Certificate Name: {cert['cn']}\n"
|
||
+ f"Subject DN: {cert['subject']}\n"
|
||
+ f"Issuer DN: {cert['issuer']}\n"
|
||
+ f"Expires: {cert['expires']}\n"
|
||
+ f"Trust Flags: {cert['trust_flags']}\n"
|
||
+ )
|
||
+ if log:
|
||
+ log.info(msg)
|
||
+
|
||
def security_enable(inst, basedn, log, args):
|
||
dbpath = inst.get_cert_dir()
|
||
tlsdb = NssSsl(dbpath=dbpath)
|
||
@@ -150,7 +198,7 @@ def security_enable(inst, basedn, log, args):
|
||
|
||
if len(certs) == 1:
|
||
# If there is only cert make sure it is set as the server certificate
|
||
- RSA(inst).set('nsSSLPersonalitySSL', certs[0][0])
|
||
+ RSA(inst).set('nsSSLPersonalitySSL', certs[0]['cn'])
|
||
elif args.cert_name is not None:
|
||
# A certificate nickname was provided, set it as the server certificate
|
||
RSA(inst).set('nsSSLPersonalitySSL', args.cert_name)
|
||
@@ -222,26 +270,27 @@ def cert_add(inst, basedn, log, args):
|
||
if not os.path.isfile(args.file):
|
||
raise ValueError(f'Certificate file "{args.file}" does not exist')
|
||
|
||
- tlsdb = NssSsl(dirsrv=inst)
|
||
- if not tlsdb._db_exists(even_partial=True): # we want to be very careful
|
||
- log.info('Security database does not exist. Creating a new one in {}.'.format(inst.get_cert_dir()))
|
||
- tlsdb.reinit()
|
||
- try:
|
||
- tlsdb.get_cert_details(args.name)
|
||
- raise ValueError("Certificate already exists with the same name")
|
||
- except ValueError:
|
||
- pass
|
||
-
|
||
- if args.primary_cert:
|
||
- # This is the server's primary certificate, update RSA entry
|
||
- RSA(inst).set('nsSSLPersonalitySSL', args.name)
|
||
+ pkcs12_password = None
|
||
+ pkcs12_file = args.file.lower().endswith((".p12", ".pfx"))
|
||
+ if pkcs12_file:
|
||
+ pkcs12_password = _resolve_pkcs12_password(args)
|
||
|
||
- # Add the cert
|
||
- tlsdb.add_cert(args.name, args.file)
|
||
+ certmgr = CertManager(instance=inst)
|
||
+ cert = certmgr.get_cert(args.name)
|
||
+ if cert:
|
||
+ log.info(f"Certificate '{args.name}' already exists, skipping")
|
||
+ return
|
||
|
||
+ certmgr.add_cert(
|
||
+ args.file,
|
||
+ args.name,
|
||
+ pkcs12_password=pkcs12_password,
|
||
+ primary=args.primary_cert,
|
||
+ ca=False,
|
||
+ force=args.force
|
||
+ )
|
||
log.info("Successfully added certificate")
|
||
|
||
-
|
||
def cacert_add(inst, basedn, log, args):
|
||
"""Add CA certificate, or CA certificate bundle
|
||
"""
|
||
@@ -249,100 +298,81 @@ def cacert_add(inst, basedn, log, args):
|
||
if not os.path.isfile(args.file):
|
||
raise ValueError(f'Certificate file "{args.file}" does not exist')
|
||
|
||
- tls = NssSsl(dirsrv=inst)
|
||
- if not tls._db_exists(even_partial=True): # we want to be very careful
|
||
- log.info('Security database does not exist. Creating a new one in {}.'.format(inst.get_cert_dir()))
|
||
- tls.reinit()
|
||
-
|
||
- tls.add_ca_cert_bundle(args.file, args.name)
|
||
+ # Does it make sense to add a CA cert from p12 container ?
|
||
+ if args.file.lower().endswith((".p12", ".pfx")):
|
||
+ raise ValueError("PKCS#12 CA certificates not supported. Use PEM or DER file")
|
||
|
||
+ certmgr = CertManager(instance=inst)
|
||
+ certmgr.add_ca_cert(args.file, args.name, force=args.force)
|
||
+ log.info("Successfully added CA certificate")
|
||
|
||
def cert_list(inst, basedn, log, args):
|
||
"""List all the server certificates
|
||
"""
|
||
- cert_list = []
|
||
- tlsdb = NssSsl(dirsrv=inst)
|
||
- certs = tlsdb.list_certs()
|
||
- for cert in certs:
|
||
- if args.json:
|
||
- cert_list.append(
|
||
- {
|
||
- "type": "certificate",
|
||
- "attrs": {
|
||
- 'nickname': cert[0],
|
||
- 'subject': cert[1],
|
||
- 'issuer': cert[2],
|
||
- 'expires': cert[3],
|
||
- 'flags': cert[4],
|
||
- }
|
||
- }
|
||
- )
|
||
- else:
|
||
- log.info('Certificate Name: {}'.format(cert[0]))
|
||
- log.info('Subject DN: {}'.format(cert[1]))
|
||
- log.info('Issuer DN: {}'.format(cert[2]))
|
||
- log.info('Expires: {}'.format(cert[3]))
|
||
- log.info('Trust Flags: {}\n'.format(cert[4]))
|
||
- if args.json:
|
||
- log.info(json.dumps(cert_list, indent=4))
|
||
+ certmgr = CertManager(instance=inst)
|
||
+ certs = certmgr.list_certs()
|
||
+ if not certs:
|
||
+ log.info("No certificates found.")
|
||
+ return
|
||
|
||
+ if args.json:
|
||
+ output = [_dump_cert(cert, json_output=True) for cert in certs]
|
||
+ log.info(json.dumps(output, indent=4))
|
||
+ else:
|
||
+ for cert in certs:
|
||
+ _dump_cert(cert, json_output=False, log=log)
|
||
|
||
def cacert_list(inst, basedn, log, args):
|
||
"""List all CA certs
|
||
"""
|
||
- cert_list = []
|
||
- tlsdb = NssSsl(dirsrv=inst)
|
||
- certs = tlsdb.list_certs(ca=True)
|
||
- for cert in certs:
|
||
- if args.json:
|
||
- cert_list.append(
|
||
- {
|
||
- "type": "certificate",
|
||
- "attrs": {
|
||
- 'nickname': cert[0],
|
||
- 'subject': cert[1],
|
||
- 'issuer': cert[2],
|
||
- 'expires': cert[3],
|
||
- 'flags': cert[4],
|
||
- }
|
||
- }
|
||
- )
|
||
- else:
|
||
- log.info('Certificate Name: {}'.format(cert[0]))
|
||
- log.info('Subject DN: {}'.format(cert[1]))
|
||
- log.info('Issuer DN: {}'.format(cert[2]))
|
||
- log.info('Expires: {}'.format(cert[3]))
|
||
- log.info('Trust Flags: {}\n'.format(cert[4]))
|
||
- if args.json:
|
||
- log.info(json.dumps(cert_list, indent=4))
|
||
+ certmgr = CertManager(instance=inst)
|
||
+ ca_certs = certmgr.list_ca_certs()
|
||
+ if not ca_certs:
|
||
+ log.info("No CA certificates found.")
|
||
+ return
|
||
|
||
+ if args.json:
|
||
+ output = [_dump_cert(cert, json_output=True) for cert in ca_certs]
|
||
+ log.info(json.dumps(output, indent=4))
|
||
+ else:
|
||
+ for cert in ca_certs:
|
||
+ _dump_cert(cert, json_output=False, log=log)
|
||
|
||
def cert_get(inst, basedn, log, args):
|
||
"""Get the details about a server certificate
|
||
"""
|
||
- tlsdb = NssSsl(dirsrv=inst)
|
||
- details = tlsdb.get_cert_details(args.name)
|
||
+ certmgr = CertManager(instance=inst)
|
||
+ cert = certmgr.get_cert(args.name)
|
||
+ if not cert:
|
||
+ log.error(f"Certificate '{args.name}' not found.")
|
||
+ return
|
||
+
|
||
+ if "C" in cert.get("trust_flags", ""):
|
||
+ return
|
||
+
|
||
if args.json:
|
||
- log.info(json.dumps(
|
||
- {
|
||
- "type": "certificate",
|
||
- "attrs": {
|
||
- 'nickname': details[0],
|
||
- 'subject': details[1],
|
||
- 'issuer': details[2],
|
||
- 'expires': details[3],
|
||
- 'flags': details[4],
|
||
- }
|
||
- }, indent=4
|
||
- )
|
||
- )
|
||
+ output = _dump_cert(cert, json_output=args.json)
|
||
+ log.info(json.dumps(output, indent=4))
|
||
else:
|
||
- log.info('Certificate Name: {}'.format(details[0]))
|
||
- log.info('Subject DN: {}'.format(details[1]))
|
||
- log.info('Issuer DN: {}'.format(details[2]))
|
||
- log.info('Expires: {}'.format(details[3]))
|
||
- log.info('Trust Flags: {}'.format(details[4]))
|
||
+ _dump_cert(cert, json_output=args.json, log=log)
|
||
+
|
||
+def cacert_get(inst, basedn, log, args):
|
||
+ """Get the details about a CA certificate
|
||
+ """
|
||
+ certmgr = CertManager(instance=inst)
|
||
+ cert = certmgr.get_cert(args.name)
|
||
+ if not cert:
|
||
+ log.error(f"Certificate '{args.name}' not found.")
|
||
+ return
|
||
+
|
||
+ if "C" not in cert.get("trust_flags", ""):
|
||
+ return
|
||
|
||
+ if args.json:
|
||
+ output = _dump_cert(cert, json_output=args.json)
|
||
+ log.info(json.dumps(output, indent=4))
|
||
+ else:
|
||
+ _dump_cert(cert, json_output=args.json, log=log)
|
||
|
||
def csr_list(inst, basedn, log, args):
|
||
"""
|
||
@@ -412,18 +442,20 @@ def csr_del(inst, basedn, log, args):
|
||
def cert_edit(inst, basedn, log, args):
|
||
"""Edit cert
|
||
"""
|
||
- tlsdb = NssSsl(dirsrv=inst)
|
||
- tlsdb.edit_cert_trust(args.name, args.flags)
|
||
+ certmgr = CertManager(instance=inst)
|
||
+ certmgr.edit_cert_trust(args.name, args.flags)
|
||
log.info("Successfully edited certificate trust flags")
|
||
|
||
|
||
def cert_del(inst, basedn, log, args):
|
||
"""Delete cert
|
||
"""
|
||
- tlsdb = NssSsl(dirsrv=inst)
|
||
- tlsdb.del_cert(args.name)
|
||
- log.info(f"Successfully deleted certificate")
|
||
-
|
||
+ certmgr = CertManager(instance=inst)
|
||
+ try:
|
||
+ certmgr.del_cert(args.name)
|
||
+ log.info(f"Successfully deleted certificate")
|
||
+ except ValueError as e:
|
||
+ log.error(f"Failed to delete certificate '{args.name}': {e}")
|
||
|
||
def key_list(inst, basedn, log, args):
|
||
"""
|
||
@@ -509,6 +541,11 @@ def create_parser(subparsers):
|
||
help='Sets the name/nickname of the certificate')
|
||
cert_add_parser.add_argument('--primary-cert', action='store_true',
|
||
help="Sets this certificate as the server's certificate")
|
||
+ cert_add_parser.add_argument('--pkcs12-pin-text', help='The PKCS#12 password as plain text. WARNING: Password may appear' \
|
||
+ ' in process list or shell history. Use --pkcs12-pin-stdin or --pkcs12-pin-path to prevent password exposure.')
|
||
+ cert_add_parser.add_argument('--pkcs12-pin-stdin', help='Read the PKCS#12 password from stdin', action='store_true')
|
||
+ cert_add_parser.add_argument('--pkcs12-pin-path', help='Path to a file containing the PKCS#12 password')
|
||
+ cert_add_parser.add_argument('--do-it', dest="force", help="Force the addition of a certificate that cannot be verified",action='store_true', default=False)
|
||
cert_add_parser.set_defaults(func=cert_add)
|
||
|
||
cert_edit_parser = certs_sub.add_parser('set-trust-flags', help='Set the Trust flags',
|
||
@@ -519,7 +556,7 @@ def create_parser(subparsers):
|
||
cert_edit_parser.set_defaults(func=cert_edit)
|
||
|
||
cert_del_parser = certs_sub.add_parser('del', help='Delete a certificate',
|
||
- description=('Delete a certificate from the NSS database'))
|
||
+ description=('Delete a server certificate from the NSS database or DynamicCerts backend.'))
|
||
cert_del_parser.add_argument('name', help='The name/nickname of the certificate')
|
||
cert_del_parser.set_defaults(func=cert_del)
|
||
|
||
@@ -529,19 +566,20 @@ def create_parser(subparsers):
|
||
cert_get_parser.set_defaults(func=cert_get)
|
||
|
||
cert_list_parser = certs_sub.add_parser('list', help='List the server certificates',
|
||
- description=('Lists the server certificates in the NSS database'))
|
||
+ description=('List all server certificates in the NSS database or DynamicCerts backend.'))
|
||
cert_list_parser.set_defaults(func=cert_list)
|
||
|
||
# CA certificate management
|
||
cacerts = security_sub.add_parser('ca-certificate', help='Manage TLS certificate authorities', formatter_class=CustomHelpFormatter)
|
||
cacerts_sub = cacerts.add_subparsers(help='ca-certificate')
|
||
cacert_add_parser = cacerts_sub.add_parser('add', help='Add a Certificate Authority', description=(
|
||
- 'Add a Certificate Authority to the NSS database'))
|
||
+ 'Add a CA certificate (PEM or DER only) to the NSS database or DynamicCerts backend.'))
|
||
cacert_add_parser.add_argument('--file', required=True,
|
||
- help='Sets the file name of the CA certificate')
|
||
- cacert_add_parser.add_argument('--name', nargs='+', required=True,
|
||
- help='Sets the name/nickname of the CA certificate, if adding a PEM bundle then specify multiple names one for '
|
||
+ help='Path to the CA certificate file (PEM or DER). If adding a PEM bundle then specify multiple names one for '
|
||
'each certificate, otherwise a number increment will be added to the previous name.')
|
||
+ cacert_add_parser.add_argument('--name', nargs='+', required=True,
|
||
+ help='Sets the name/nickname of the CA certificate')
|
||
+ cacert_add_parser.add_argument('--do-it', dest="force", help="Force the addition of a certificate that cannot be verified",action='store_true', default=False)
|
||
cacert_add_parser.set_defaults(func=cacert_add)
|
||
|
||
cacert_edit_parser = cacerts_sub.add_parser('set-trust-flags', help='Set the Trust flags',
|
||
@@ -559,7 +597,7 @@ def create_parser(subparsers):
|
||
cacert_get_parser = cacerts_sub.add_parser('get', help="Displays a Certificate Authority's information",
|
||
description=('Get detailed information about a CA certificate, like trust attributes, expiration dates, Subject and Issuer DN'))
|
||
cacert_get_parser.add_argument('name', help='The name/nickname of the CA certificate')
|
||
- cacert_get_parser.set_defaults(func=cert_get)
|
||
+ cacert_get_parser.set_defaults(func=cacert_get)
|
||
|
||
cacert_list_parser = cacerts_sub.add_parser('list', help='List the Certificate Authorities',
|
||
description=('List the CA certificates in the NSS database'))
|
||
diff --git a/src/lib389/lib389/dyncerts.py b/src/lib389/lib389/dyncerts.py
|
||
new file mode 100644
|
||
index 000000000..22e1a9a36
|
||
--- /dev/null
|
||
+++ b/src/lib389/lib389/dyncerts.py
|
||
@@ -0,0 +1,458 @@
|
||
+# --- BEGIN COPYRIGHT BLOCK ---
|
||
+# Copyright (C) 2026 Red Hat, Inc.
|
||
+# All rights reserved.
|
||
+#
|
||
+# License: GPL (version 3 or any later version).
|
||
+# See LICENSE for details.
|
||
+# --- END COPYRIGHT BLOCK ---
|
||
+
|
||
+import datetime
|
||
+import os
|
||
+import ldap
|
||
+import logging
|
||
+import re
|
||
+import tempfile
|
||
+from typing import Optional
|
||
+from lib389._mapped_object import DSLdapObjects, DSLdapObject
|
||
+from lib389.utils import cert_is_ca, pem_to_der, is_pem_cert, ensure_str
|
||
+from cryptography import x509
|
||
+from cryptography.hazmat.backends import default_backend
|
||
+from cryptography.hazmat.primitives import serialization
|
||
+from cryptography.hazmat.primitives.serialization import pkcs12
|
||
+
|
||
+log = logging.getLogger(__name__)
|
||
+
|
||
+DYCATTR_CN = "cn"
|
||
+DYNCERT_SUFFIX = "cn=dynamiccertificates"
|
||
+
|
||
+DYCATTR_PREFIX = "dsdynamiccertificate"
|
||
+DYCATTR_CERTDER = DYCATTR_PREFIX + "der"
|
||
+DYCATTR_PKEYDER = DYCATTR_PREFIX + "privatekeyder"
|
||
+DYCATTR_SUBJECT = DYCATTR_PREFIX + "subject"
|
||
+DYCATTR_ISSUER = DYCATTR_PREFIX + "issuer"
|
||
+DYCATTR_TRUST = DYCATTR_PREFIX + "trustflags"
|
||
+DYCATTR_NOTAFTER = DYCATTR_PREFIX + "notafter"
|
||
+DYCATTR_FORCE = DYCATTR_PREFIX + "Force"
|
||
+DYCATTR_ISCA = DYCATTR_PREFIX + "IsCA"
|
||
+
|
||
+CA_NAME = 'Self-Signed-CA'
|
||
+CERT_NAME = 'Server-Cert'
|
||
+
|
||
+class DynamicCert(DSLdapObject):
|
||
+ """
|
||
+ Represents a single DynamicCert LDAP entry.
|
||
+ """
|
||
+
|
||
+ _must_attributes = [DYCATTR_CN]
|
||
+
|
||
+ def __init__(self, instance, dn: Optional[str] = None):
|
||
+ """
|
||
+ Initialise a DynamicCert object.
|
||
+
|
||
+ :param instance: DirSrv instance
|
||
+ :param dn: Entry distinguished name (DN)
|
||
+ """
|
||
+ super(DynamicCert, self).__init__(instance, dn)
|
||
+ self._rdn_attribute = DYCATTR_CN
|
||
+ self._create_objectclasses = ["top", "extensibleObject"]
|
||
+ self._protected = False
|
||
+ self._basedn = DYNCERT_SUFFIX
|
||
+
|
||
+ def _normalise_timestamp(self, raw: str):
|
||
+ """
|
||
+ Convert DynamicCert timestamp to NSS like format.
|
||
+
|
||
+ :param raw: Raw DynamicCert timestamp (e.g. 20260109181934Z)
|
||
+ :return: Formatted timestamp as "YYYY-MM-DD HH:MM:SS", or original string
|
||
+ """
|
||
+ try:
|
||
+ year = int(raw[0:4])
|
||
+ month = int(raw[4:6]) + 1 # PRExplodedTime.tm_month is 0–11
|
||
+ day = int(raw[6:8])
|
||
+ hour = int(raw[8:10])
|
||
+ minute = int(raw[10:12])
|
||
+ second = int(raw[12:14])
|
||
+ return datetime.datetime(year, month, day, hour, minute, second).strftime("%Y-%m-%d %H:%M:%S")
|
||
+ except Exception:
|
||
+ return raw
|
||
+
|
||
+ def del_cert(self):
|
||
+ """
|
||
+ Delete this DynamicCert entry from LDAP.
|
||
+
|
||
+ :raises ValueError: If the DynamicCert object does not have a DN
|
||
+ :raises ldap.LDAPError: If an LDAP operation fails (other than NO_SUCH_OBJECT)
|
||
+ """
|
||
+ if not self._dn:
|
||
+ raise ValueError("Cannot delete DynamicCert without a DN")
|
||
+
|
||
+ try:
|
||
+ self.delete()
|
||
+ except ldap.NO_SUCH_OBJECT:
|
||
+ log.warning(f"DynamicCert already deleted: {self._dn}")
|
||
+ except ldap.LDAPError as e:
|
||
+ log.error(f"Failed to delete DynamicCert: {self._dn}: {e}")
|
||
+ raise
|
||
+
|
||
+ def edit_trust(self, trust_flags: str):
|
||
+ """
|
||
+ Edit certificate trust flags.
|
||
+
|
||
+ :param trust_flags: Comma separated trust flags string (SSL,Email,ObjectSigning)
|
||
+ :raises ValueError: If trust flags are invalid or empty
|
||
+ """
|
||
+ if not trust_flags:
|
||
+ raise ValueError("Trust flags cannot be empty")
|
||
+
|
||
+ trust_fields = trust_flags.strip().split(",")
|
||
+
|
||
+ if len(trust_fields) != 3:
|
||
+ raise ValueError("Trust flags must have 3 comma separated fields")
|
||
+
|
||
+ # Allowed field values (NSS)
|
||
+ valid_flags = set("pPcCTu")
|
||
+ for field in trust_fields:
|
||
+ if field and any(flag not in valid_flags for flag in field):
|
||
+ raise ValueError(f"Invalid characters in trust flags: '{trust_flags}'")
|
||
+
|
||
+ try:
|
||
+ self.replace(DYCATTR_TRUST, trust_flags)
|
||
+ except ldap.LDAPError as e:
|
||
+ log.error(f"Failed to update trust flags for {self._dn}: {e}")
|
||
+ raise
|
||
+
|
||
+class DynamicCerts(DSLdapObjects):
|
||
+ """
|
||
+ Collection of DynamicCert entries under cn=dynamiccertificates.
|
||
+ """
|
||
+
|
||
+ def __init__(self, instance):
|
||
+ """
|
||
+ Initialise the DynamicCerts collection.
|
||
+
|
||
+ :param instance: DirSrv instance
|
||
+ """
|
||
+ super(DynamicCerts, self).__init__(instance=instance)
|
||
+ self._objectclasses = ["extensibleObject"]
|
||
+ self._filterattrs = [DYCATTR_CN, DYCATTR_SUBJECT, DYCATTR_ISSUER]
|
||
+ self._childobject = DynamicCert
|
||
+ self._basedn = DYNCERT_SUFFIX
|
||
+
|
||
+ def add_cert(self,
|
||
+ cert_file: str,
|
||
+ nickname: str,
|
||
+ pkcs12_password: Optional[str] = None,
|
||
+ ca: bool = False,
|
||
+ force: bool = False,
|
||
+ ):
|
||
+ """
|
||
+ Add or update a certificate (PEM, DER, or PKCS#12).
|
||
+
|
||
+ :param cert_file: Path to certificate file (PEM, DER, or PKCS#12)
|
||
+ :param nickname: Certificate nickname
|
||
+ :param pkcs12_password: Password for PKCS#12, if any
|
||
+ :param ca: Whether this certificate is a CA certificate
|
||
+ :param force: Force the addition of a certificate that cannot be verified
|
||
+ """
|
||
+ if not nickname:
|
||
+ raise ValueError("Certificate CN cannot be empty")
|
||
+
|
||
+ if not os.path.isfile(cert_file):
|
||
+ raise ValueError(f"Certificate file does not exist: {cert_file}")
|
||
+
|
||
+ if pkcs12_password and not isinstance(pkcs12_password, str):
|
||
+ raise TypeError("PKCS#12 password must be a string")
|
||
+
|
||
+ with open(cert_file, "rb") as f:
|
||
+ cert_bytes = f.read()
|
||
+
|
||
+ der_cert = None
|
||
+ der_privkey = None
|
||
+
|
||
+ if cert_file.lower().endswith((".p12", ".pfx")):
|
||
+ try:
|
||
+ privkey, cert, _ = pkcs12.load_key_and_certificates(
|
||
+ cert_bytes, pkcs12_password.encode() if pkcs12_password else None,
|
||
+ backend=default_backend()
|
||
+ )
|
||
+ except Exception as e:
|
||
+ raise ValueError(f"Failed to load PKCS#12 file: {cert_file}: {e}")
|
||
+
|
||
+ if cert is None:
|
||
+ raise ValueError("PKCS#12 file contains no certificate")
|
||
+
|
||
+ der_cert = cert.public_bytes(serialization.Encoding.DER)
|
||
+ if privkey is not None:
|
||
+ der_privkey = privkey.private_bytes(
|
||
+ encoding=serialization.Encoding.DER,
|
||
+ format=serialization.PrivateFormat.PKCS8,
|
||
+ encryption_algorithm=serialization.NoEncryption()
|
||
+ )
|
||
+ else:
|
||
+ try:
|
||
+ der_cert = pem_to_der(cert_bytes) if is_pem_cert(cert_bytes) else cert_bytes
|
||
+ except Exception as e:
|
||
+ raise ValueError(f"Failed to parse certificate '{cert_file}': {e}")
|
||
+
|
||
+ if ca and not cert_is_ca(cert_file):
|
||
+ raise ValueError(f"Certificate ({nickname}) is not a CA certificate")
|
||
+
|
||
+ attrs = {
|
||
+ "cn": [nickname.encode()],
|
||
+ "objectClass": [b"top", b"extensibleObject"],
|
||
+ DYCATTR_CERTDER: [ der_cert, ]
|
||
+ }
|
||
+ if der_privkey:
|
||
+ attrs[DYCATTR_PKEYDER] = [der_privkey]
|
||
+
|
||
+ if ca:
|
||
+ attrs[DYCATTR_TRUST] = [b"CT,,"]
|
||
+
|
||
+ if force:
|
||
+ attrs[DYCATTR_FORCE] = [b"TRUE"]
|
||
+
|
||
+ if not der_cert:
|
||
+ raise ValueError(f"Failed to extract DER bytes from {cert_file}")
|
||
+
|
||
+ # Escape the CN to handle special chars in the nickname
|
||
+ escaped_cn = ldap.dn.escape_dn_chars(nickname)
|
||
+ dn = ensure_str(f"cn={escaped_cn},{self._basedn}")
|
||
+ # Raw CN used for lookup
|
||
+ cert_obj = self.get_cert_obj(nickname)
|
||
+ if not cert_obj:
|
||
+ cert_obj = DynamicCert(self._instance, dn)
|
||
+ cert_obj.create(properties=attrs)
|
||
+ else:
|
||
+ log.info(f"Updating existing certificate; {nickname}")
|
||
+ attrs_list = [(attr, vals) for attr, vals in attrs.items()]
|
||
+ cert_obj.replace_many(*attrs_list)
|
||
+
|
||
+ def add_ca_cert(self,
|
||
+ cert_file: str,
|
||
+ nickname: str,
|
||
+ pkcs12_password: Optional[str] = None,
|
||
+ force: bool = False
|
||
+ ):
|
||
+ """
|
||
+ Add a CA certificate from a PEM bundle or single PEM/DER file.
|
||
+
|
||
+ :param cert_file: Path to the certificate file (PEM or DER)
|
||
+ :param nickname: Certificate nickname
|
||
+ :param pkcs12_password: Password for PKCS#12, if any
|
||
+ :param force: Force the addition of a certificate that cannot be verified
|
||
+ :raises ValueError: If file is invalid or file does not exist
|
||
+ """
|
||
+ if not os.path.exists(cert_file):
|
||
+ raise ValueError(f"Certificate file does not exist: {cert_file}")
|
||
+
|
||
+ # Normalise nickname(s)
|
||
+ if isinstance(nickname, str):
|
||
+ nicknames = [nickname]
|
||
+ elif isinstance(nickname, list):
|
||
+ nicknames = nickname
|
||
+ else:
|
||
+ raise TypeError(f"nickname must be str or list[str], got {type(nickname)}")
|
||
+
|
||
+ # PEM (may be bundle)
|
||
+ if cert_file.lower().endswith(".pem"):
|
||
+ with open(cert_file, "r") as f:
|
||
+ pem_data = f.read()
|
||
+
|
||
+ pem_certs = re.findall(
|
||
+ r"-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----",
|
||
+ pem_data,
|
||
+ re.DOTALL
|
||
+ )
|
||
+
|
||
+ if not pem_certs:
|
||
+ raise ValueError("No certificates found in PEM file")
|
||
+
|
||
+ temp_files = []
|
||
+ try:
|
||
+ for idx, cert in enumerate(pem_certs):
|
||
+ with tempfile.NamedTemporaryFile(delete=False, suffix=".pem", mode="w") as tmp:
|
||
+ tmp.write(cert.strip() + "\n")
|
||
+ tmp_cert_path = tmp.name
|
||
+ temp_files.append(tmp_cert_path)
|
||
+
|
||
+ # Determine nickname, could be a list, or not enough
|
||
+ if idx < len(nicknames):
|
||
+ ca_nick = nicknames[idx]
|
||
+ else:
|
||
+ ca_nick = f"{nicknames[-1]}{idx}"
|
||
+
|
||
+ # Check we dont trample over installer certs
|
||
+ if ca_nick.lower() in (CERT_NAME.lower(), CA_NAME.lower()):
|
||
+ raise ValueError(f"You may not import a CA with the nickname {CERT_NAME} or {CA_NAME}")
|
||
+
|
||
+ if not cert_is_ca(tmp_cert_path):
|
||
+ raise ValueError(f"Certificate ({ca_nick}) is not a CA certificate")
|
||
+
|
||
+ try:
|
||
+ # Handle existing LDAP object
|
||
+ if self.get_cert_details(ca_nick):
|
||
+ if not force:
|
||
+ raise ValueError(
|
||
+ f"Certificate already exists with the same name ({ca_nick})"
|
||
+ )
|
||
+ else:
|
||
+ log.info(f"Overwriting existing CA cert {ca_nick}")
|
||
+ self.del_cert(ca_nick)
|
||
+ except ValueError:
|
||
+ pass
|
||
+
|
||
+ self.add_cert(
|
||
+ tmp_cert_path,
|
||
+ ca_nick,
|
||
+ pkcs12_password=pkcs12_password,
|
||
+ ca=True,
|
||
+ force=force
|
||
+ )
|
||
+
|
||
+ finally:
|
||
+ for tmp_file in temp_files:
|
||
+ try:
|
||
+ os.remove(tmp_file)
|
||
+ except OSError as e:
|
||
+ log.debug(f"Failed to remove tmp cert file: {tmp_file}: {e}")
|
||
+ else:
|
||
+ # Single binary cert
|
||
+ if len(nicknames) != 1:
|
||
+ raise ValueError("Single cert requires exactly one nickname")
|
||
+
|
||
+ ca_nick = nicknames[0]
|
||
+ try:
|
||
+ if self.get_cert_details(ca_nick):
|
||
+ if force:
|
||
+ self.del_cert(ca_nick)
|
||
+ else:
|
||
+ raise ValueError(f"Certificate already exists: {ca_nick}")
|
||
+ except ValueError:
|
||
+ pass
|
||
+
|
||
+ self.add_cert(cert_file, ca_nick, pkcs12_password=pkcs12_password, ca=True, force=force)
|
||
+
|
||
+ def del_cert(self, nickname: str):
|
||
+ """
|
||
+ Delete a certificate.
|
||
+
|
||
+ :param nickname: The certificate nickname to delete.
|
||
+ :raises ValueError: If the nickname is empty or the entry cannot be found.
|
||
+ """
|
||
+ if not nickname:
|
||
+ raise ValueError("Certificate nickname cannot be empty")
|
||
+
|
||
+ cert_obj = self.get_cert_obj(nickname)
|
||
+ if not cert_obj:
|
||
+ raise ValueError(f"Certificate entry not found: {nickname}")
|
||
+
|
||
+ cert_obj.del_cert()
|
||
+
|
||
+ def list_certs(self):
|
||
+ """
|
||
+ List all server certificates.
|
||
+
|
||
+ :return: A list of certificate dictionaries for each certificate
|
||
+ """
|
||
+ cert_objects = self.list()
|
||
+ certs = []
|
||
+ for cert in cert_objects:
|
||
+ if cert._dn == self._basedn:
|
||
+ continue
|
||
+ der_cert = cert.get_attr_vals_bytes(DYCATTR_CERTDER)[0]
|
||
+ if der_cert:
|
||
+ if not cert_is_ca(der_cert):
|
||
+ certs.append({
|
||
+ "cn": cert.get_attr_vals_utf8(DYCATTR_CN)[0],
|
||
+ "subject": cert.get_attr_vals_utf8(DYCATTR_SUBJECT)[0],
|
||
+ "issuer": cert.get_attr_vals_utf8(DYCATTR_ISSUER)[0],
|
||
+ "expires": cert._normalise_timestamp(cert.get_attr_vals_utf8(DYCATTR_NOTAFTER)[0]),
|
||
+ "trust_flags": cert.get_attr_vals_utf8(DYCATTR_TRUST)[0],
|
||
+ })
|
||
+ return certs
|
||
+
|
||
+ def list_ca_certs(self):
|
||
+ """
|
||
+ List all ca certificates.
|
||
+
|
||
+ :return: A list of certificate dictionaries for each certificate
|
||
+ """
|
||
+ cert_objects = self.list()
|
||
+ certs = []
|
||
+ for cert in cert_objects:
|
||
+ if cert._dn == self._basedn:
|
||
+ continue
|
||
+ der_cert = cert.get_attr_vals_bytes(DYCATTR_CERTDER)[0]
|
||
+ if der_cert:
|
||
+ if cert_is_ca(der_cert):
|
||
+ certs.append({
|
||
+ "cn": cert.get_attr_vals_utf8(DYCATTR_CN)[0],
|
||
+ "subject": cert.get_attr_vals_utf8(DYCATTR_SUBJECT)[0],
|
||
+ "issuer": cert.get_attr_vals_utf8(DYCATTR_ISSUER)[0],
|
||
+ "expires": cert._normalise_timestamp(cert.get_attr_vals_utf8(DYCATTR_NOTAFTER)[0]),
|
||
+ "trust_flags": cert.get_attr_vals_utf8(DYCATTR_TRUST)[0],
|
||
+ })
|
||
+ return certs
|
||
+
|
||
+ def get_cert_obj(self, nickname: str):
|
||
+ """
|
||
+ Retrieve a certificate object.
|
||
+
|
||
+ :param cn: Certificate nickname
|
||
+ :raises ValueError: If the cn is empty
|
||
+ :return: DynamicCert object if found, else None
|
||
+ """
|
||
+ if not nickname:
|
||
+ raise ValueError("Certificate CN cannot be empty")
|
||
+ try:
|
||
+ cert = self.get(nickname)
|
||
+ except ldap.NO_SUCH_OBJECT:
|
||
+ return None
|
||
+
|
||
+ return cert
|
||
+
|
||
+ def get_cert_details(self, nickname: str):
|
||
+ """
|
||
+ Get a certificates details.
|
||
+
|
||
+ :param nickname: Certificate nickname
|
||
+ :raises ValueError: If the nickname is empty
|
||
+ :return: DynamicCert object if found, else None
|
||
+ """
|
||
+ if not nickname:
|
||
+ raise ValueError("Certificate CN cannot be empty")
|
||
+ try:
|
||
+ cert = self.get(nickname)
|
||
+ except ldap.NO_SUCH_OBJECT:
|
||
+ return None
|
||
+
|
||
+ return {
|
||
+ "cn": cert.get_attr_vals_utf8(DYCATTR_CN)[0],
|
||
+ "subject": cert.get_attr_vals_utf8(DYCATTR_SUBJECT)[0],
|
||
+ "issuer": cert.get_attr_vals_utf8(DYCATTR_ISSUER)[0],
|
||
+ "expires": cert._normalise_timestamp(cert.get_attr_vals_utf8(DYCATTR_NOTAFTER)[0]),
|
||
+ "trust_flags": cert.get_attr_vals_utf8(DYCATTR_TRUST)[0],
|
||
+ }
|
||
+
|
||
+ def edit_cert_trust(self, nickname: str, trust_flags: str):
|
||
+ """
|
||
+ Edit trust flags on an existing certificate.
|
||
+
|
||
+ :param nickname: Certificate nickname
|
||
+ :param trust_flags: 3 field NSS trust string
|
||
+ """
|
||
+ cert_obj = self.get_cert_obj(nickname)
|
||
+ if not cert_obj:
|
||
+ raise ValueError(f"Certificate {nickname} does not exist")
|
||
+
|
||
+ try:
|
||
+ cert_obj.edit_trust(trust_flags=trust_flags)
|
||
+ except ValueError as ve:
|
||
+ log.error(f"Invalid input for certificate '{nickname}': {ve}")
|
||
+ raise
|
||
+ except ldap.LDAPError as le:
|
||
+ log.error(f"LDAP error while updating certificate '{nickname}': {le}")
|
||
+ raise
|
||
+ except Exception as e:
|
||
+ log.error(f"Unexpected error while updating certificate '{nickname}': {e}")
|
||
+ raise
|
||
diff --git a/src/lib389/lib389/nss_ssl.py b/src/lib389/lib389/nss_ssl.py
|
||
index 2fd2b9b89..764434166 100644
|
||
--- a/src/lib389/lib389/nss_ssl.py
|
||
+++ b/src/lib389/lib389/nss_ssl.py
|
||
@@ -1,5 +1,5 @@
|
||
# --- BEGIN COPYRIGHT BLOCK ---
|
||
-# Copyright (C) 2023 Red Hat, Inc.
|
||
+# Copyright (C) 2026 Red Hat, Inc.
|
||
# All rights reserved.
|
||
#
|
||
# License: GPL (version 3 or any later version).
|
||
@@ -18,6 +18,7 @@ import shutil
|
||
import logging
|
||
import subprocess
|
||
import uuid
|
||
+from typing import Optional
|
||
from datetime import datetime, timedelta
|
||
from subprocess import check_output, run, PIPE
|
||
from lib389.passwd import password_generate
|
||
@@ -25,7 +26,6 @@ from lib389._mapped_object_lint import DSLint
|
||
from lib389.lint import DSCERTLE0001, DSCERTLE0002
|
||
from lib389.utils import ensure_str, format_cmd_list, DSVersion, cert_is_ca
|
||
|
||
-
|
||
KEYBITS = 4096
|
||
CA_NAME = 'Self-Signed-CA'
|
||
CERT_NAME = 'Server-Cert'
|
||
@@ -85,9 +85,8 @@ class NssSsl(DSLint):
|
||
all_certs = self._rsa_cert_list()
|
||
for cert in all_certs:
|
||
cert_list.append(self.get_cert_details(cert[0]))
|
||
-
|
||
for cert in cert_list:
|
||
- cert_date = cert[3].split()[0]
|
||
+ cert_date = cert['expires'].split()[0]
|
||
diff_date = datetime.strptime(cert_date, '%Y-%m-%d').date() - datetime.today().date()
|
||
if diff_date < timedelta(days=0):
|
||
# Expired
|
||
@@ -1099,12 +1098,12 @@ only.
|
||
def get_cert_details(self, nickname):
|
||
"""Get the trust flags, subject DN, issuer, and expiration date
|
||
|
||
- return a list:
|
||
- 0 - nickname
|
||
- 1 - subject
|
||
- 2 - issuer
|
||
- 3 - expire date
|
||
- 4 - trust_flags
|
||
+ :return: dict containing certificate details with keys:
|
||
+ - "cn" : Certificate nickname (str)
|
||
+ - "subject" : Subject DN (str)
|
||
+ - "issuer" : Issuer DN (str)
|
||
+ - "expires" : Expiration date/time as a string (YYYY-MM-DD HH:MM:SS)
|
||
+ - "trust_flags" : Trust flags from NSS (str)
|
||
"""
|
||
all_certs = self._rsa_cert_list()
|
||
for cert in all_certs:
|
||
@@ -1142,26 +1141,33 @@ only.
|
||
issuer = issuer[1:-1]
|
||
break
|
||
|
||
- return ([nickname, subject, issuer, str(end_date), trust_flags])
|
||
+ return {
|
||
+ "cn": nickname,
|
||
+ "subject": subject,
|
||
+ "issuer": issuer,
|
||
+ "expires": str(end_date),
|
||
+ "trust_flags": trust_flags,
|
||
+ }
|
||
|
||
- # Did not find cert with that name
|
||
- raise ValueError("Certificate '{}' not found in NSS database".format(nickname))
|
||
+ return None
|
||
|
||
def list_certs(self, ca=False):
|
||
all_certs = self._rsa_cert_list()
|
||
certs = []
|
||
- for cert in all_certs:
|
||
- trust_flags = cert[1]
|
||
+ for nickname, trust_flags in all_certs:
|
||
if (ca and "CT" in trust_flags) or (not ca and "CT" not in trust_flags):
|
||
- certs.append(self.get_cert_details(cert[0]))
|
||
+ cert_details = self.get_cert_details(nickname)
|
||
+ certs.append(cert_details)
|
||
return certs
|
||
|
||
def list_ca_certs(self):
|
||
- return [
|
||
- cert
|
||
- for cert in self._rsa_cert_list()
|
||
- if self._rsa_cert_is_catrust(cert)
|
||
- ]
|
||
+ ca_certs = []
|
||
+ for cert in self._rsa_cert_list():
|
||
+ if self._rsa_cert_is_catrust(cert):
|
||
+ # cert[0] is the nickname
|
||
+ cert_details = self.get_cert_details(cert[0])
|
||
+ ca_certs.append(cert_details)
|
||
+ return ca_certs
|
||
|
||
def list_client_ca_certs(self):
|
||
return [
|
||
@@ -1170,41 +1176,107 @@ only.
|
||
if self._rsa_cert_is_caclienttrust(cert)
|
||
]
|
||
|
||
- def add_cert(self, nickname, input_file, ca=False):
|
||
+ def add_cert(self,
|
||
+ nickname: str,
|
||
+ cert_file: str,
|
||
+ pkcs12_password: Optional[str] = None,
|
||
+ ca: bool = False,
|
||
+ force: bool = False):
|
||
"""Add server or CA cert
|
||
+
|
||
+ :param nickname: Certificate nickname
|
||
+ :param cert_file: Path to certificate file (PEM, DER, or PKCS#12)
|
||
+ :param ca: Whether this is a CA certificate
|
||
+ :param pkcs12_password: Password for PKCS#12, if any
|
||
+ :param force: Force the addition of a certificate that cannot be verified
|
||
"""
|
||
+ if not nickname:
|
||
+ raise ValueError("Certificate nickname must not be empty")
|
||
|
||
# Verify input_file exists
|
||
- if not os.path.exists(input_file):
|
||
- raise ValueError("The certificate file ({}) does not exist".format(input_file))
|
||
+ if not os.path.isfile(cert_file):
|
||
+ raise ValueError("The certificate file ({}) does not exist".format(cert_file))
|
||
|
||
pem_file = True
|
||
- if not input_file.lower().endswith(".pem"):
|
||
+ if not cert_file.lower().endswith(".pem"):
|
||
pem_file = False
|
||
else:
|
||
- self._assert_not_chain(input_file)
|
||
+ self._assert_not_chain(cert_file)
|
||
|
||
if ca:
|
||
# Verify this is a CA cert
|
||
- if not cert_is_ca(input_file):
|
||
+ if not cert_is_ca(cert_file):
|
||
raise ValueError(f"Certificate ({nickname}) is not a CA certificate")
|
||
trust_flags = "CT,,"
|
||
else:
|
||
# Verify this is a server cert
|
||
- if cert_is_ca(input_file):
|
||
+ if cert_is_ca(cert_file, pkcs12_password=pkcs12_password):
|
||
raise ValueError(f"Certificate ({nickname}) is not a server certificate")
|
||
trust_flags = ",,"
|
||
|
||
+ pkcs12_file = None
|
||
+ if cert_file.lower().endswith((".p12", ".pfx")):
|
||
+ pkcs12_file = True
|
||
+
|
||
+ if pkcs12_file:
|
||
+ self.log.info("Importing PKCS#12 into NSS: %s", cert_file)
|
||
+
|
||
+ if pkcs12_password is None:
|
||
+ pkcs12_password = ""
|
||
+
|
||
+ cmd = [
|
||
+ "pk12util",
|
||
+ "-v",
|
||
+ "-n", nickname,
|
||
+ "-i", cert_file,
|
||
+ "-d", self._certdb,
|
||
+ "-k", f"{self._certdb}/{PWD_TXT}",
|
||
+ "-W", pkcs12_password,
|
||
+ ]
|
||
+
|
||
+ # Mask the password in logs
|
||
+ masked_cmd = [arg if arg != pkcs12_password else "****" for arg in cmd]
|
||
+ self.log.info(f"nss import p12 cmd: {format_cmd_list(masked_cmd)}", )
|
||
+ try:
|
||
+ check_output(cmd, stderr=subprocess.STDOUT)
|
||
+ except subprocess.CalledProcessError as e:
|
||
+ raise ValueError(f"Failed to import PKCS#12 {cert_file} {e}")
|
||
+
|
||
+ if ca:
|
||
+ cmd = [
|
||
+ "certutil",
|
||
+ "-M",
|
||
+ "-n", nickname,
|
||
+ "-t", "CT,,",
|
||
+ "-d", self._certdb,
|
||
+ "-f", f"{self._certdb}/{PWD_TXT}",
|
||
+ ]
|
||
+ self.log.debug(f"set CA trust cmd: {format_cmd_list(cmd)}")
|
||
+ try:
|
||
+ check_output(cmd, stderr=subprocess.STDOUT)
|
||
+ except subprocess.CalledProcessError as e:
|
||
+ raise ValueError(f"Failed to set CA trust for {nickname} {e}")
|
||
+
|
||
+ return
|
||
+
|
||
+ pem_file = cert_file.lower().endswith(".pem")
|
||
+ if pem_file:
|
||
+ self._assert_not_chain(cert_file)
|
||
+
|
||
+ trust_flags = "CT,," if ca else ",,"
|
||
+
|
||
cmd = [
|
||
'/usr/bin/certutil',
|
||
'-A',
|
||
'-d', self._certdb,
|
||
'-n', nickname,
|
||
'-t', trust_flags,
|
||
- '-i', input_file,
|
||
+ '-i', cert_file,
|
||
'-f',
|
||
'%s/%s' % (self._certdb, PWD_TXT),
|
||
+
|
||
]
|
||
+
|
||
if pem_file:
|
||
cmd.append('-a')
|
||
|
||
@@ -1270,6 +1342,62 @@ only.
|
||
if os.path.exists(p12_bundle):
|
||
os.remove(p12_bundle)
|
||
|
||
+ def add_ca_cert(self, cert_file: str, nickname: str, force: bool = False):
|
||
+ """
|
||
+ Add a CA certificate from a PEM bundle or single PEM/DER file.
|
||
+
|
||
+ Adapter function to match abstraction layer interface.
|
||
+
|
||
+ :param cert_file: path to the certificate file
|
||
+ :param nickname: nickname to assign
|
||
+ :param force: Force the addition of a certificate that cannot be verified
|
||
+ """
|
||
+ # Verify input_file exists
|
||
+ if not os.path.exists(cert_file):
|
||
+ raise ValueError(f"The certificate file ({cert_file}) does not exist")
|
||
+
|
||
+ # Normalise nickname(s)
|
||
+ if isinstance(nickname, list):
|
||
+ nicknames = nickname
|
||
+ elif isinstance(nickname, str):
|
||
+ nicknames = [nickname]
|
||
+ else:
|
||
+ raise TypeError(f"nickname must be str or list[str], got: {type(nickname)}")
|
||
+
|
||
+ # Allow overwrite only for single cert
|
||
+ if len(nicknames) == 1:
|
||
+ single_nick = nicknames[0]
|
||
+ try:
|
||
+ if self.get_cert_details(single_nick):
|
||
+ if not force:
|
||
+ raise ValueError(
|
||
+ f"Certificate already exists with the same name ({single_nick})"
|
||
+ )
|
||
+ else:
|
||
+ log.info(f"Overwriting existing certificate ({single_nick})")
|
||
+ self.del_cert(single_nick)
|
||
+ except ValueError:
|
||
+ pass
|
||
+
|
||
+ # Offload to PEM bundle handler
|
||
+ if cert_file.lower().endswith(".pem"):
|
||
+ return self.add_ca_cert_bundle(
|
||
+ cert_file=cert_file,
|
||
+ nicknames=nicknames
|
||
+ )
|
||
+
|
||
+ if len(nicknames) != 1:
|
||
+ raise ValueError(
|
||
+ "Binary CA cert requires exactly one nickname"
|
||
+ )
|
||
+
|
||
+ if not cert_is_ca(cert_file):
|
||
+ raise ValueError(f"Certificate ({nickname}) is not a CA certificate")
|
||
+
|
||
+ self.add_cert(nicknames[0], cert_file, ca=True)
|
||
+
|
||
+ log.info(f"Successfully added CA certificate ({nickname})")
|
||
+
|
||
def add_ca_cert_bundle(self, cert_file, nicknames):
|
||
"""
|
||
Add a PEM file that could be a bundle of CA certs
|
||
@@ -1327,14 +1455,8 @@ only.
|
||
ca_cert_name = nicknames[names_len - 1] + str(ca_count)
|
||
|
||
# Check if certificate nickname exists
|
||
- name_exists = False
|
||
- try:
|
||
- self.get_cert_details(ca_cert_name)
|
||
- name_exists = True
|
||
- except ValueError:
|
||
- pass
|
||
-
|
||
- if name_exists:
|
||
+ cert = self.get_cert_details(ca_cert_name)
|
||
+ if cert:
|
||
# Not good, cleanup and raise error
|
||
try:
|
||
for tmp_file in ca_files_to_cleanup:
|
||
diff --git a/src/lib389/lib389/utils.py b/src/lib389/lib389/utils.py
|
||
index f8fba816b..042def090 100644
|
||
--- a/src/lib389/lib389/utils.py
|
||
+++ b/src/lib389/lib389/utils.py
|
||
@@ -32,11 +32,14 @@ import operator
|
||
import subprocess
|
||
import math
|
||
import errno
|
||
+from typing import Optional, Union
|
||
from socket import getfqdn
|
||
from ldapurl import LDAPUrl
|
||
from contextlib import closing
|
||
from cryptography import x509
|
||
from cryptography.hazmat.backends import default_backend
|
||
+from cryptography.hazmat.primitives import serialization
|
||
+from cryptography.hazmat.primitives.serialization import pkcs12
|
||
import lib389
|
||
from pathlib import Path
|
||
from subprocess import check_output
|
||
@@ -1993,17 +1996,62 @@ def check_cert_info(cert_file_name, search_text):
|
||
return search_text.lower() in cert_text.lower()
|
||
|
||
|
||
-def cert_is_ca(cert_file_name):
|
||
- with open(cert_file_name, "rb") as f:
|
||
- if is_cert_der(cert_file_name):
|
||
- cert = x509.load_der_x509_certificate(f.read(), default_backend())
|
||
+def cert_is_ca(cert_data, pkcs12_password: Optional[Union[str, bytes]] = None):
|
||
+ """
|
||
+ Determine if a certificate is a CA.
|
||
+
|
||
+ Supports PEM, DER, and PKCS#12 certificates, from bytes or file paths.
|
||
+ """
|
||
+ # If passed bytes directly (DER,PEM)
|
||
+ cert_file = None
|
||
+ if isinstance(cert_data, (bytes, bytearray)):
|
||
+ data = bytes(cert_data)
|
||
+ else:
|
||
+ cert_file = cert_data
|
||
+ try:
|
||
+ with open(cert_file, "rb") as f:
|
||
+ data = f.read()
|
||
+ except OSError as e:
|
||
+ raise ValueError(f"Unable to load certificate '{cert_data}': {e}")
|
||
+
|
||
+ try:
|
||
+ # p12
|
||
+ if cert_file and cert_file.lower().endswith((".p12", ".pfx")):
|
||
+ try:
|
||
+ privkey, cert, _ = pkcs12.load_key_and_certificates(
|
||
+ data, ensure_bytes(pkcs12_password),
|
||
+ backend=default_backend()
|
||
+ )
|
||
+ except Exception as e:
|
||
+ raise ValueError(f"Failed to load PKCS#12 file: {cert_file}: {e}")
|
||
+
|
||
+ if cert is None:
|
||
+ raise ValueError("No certificate found in PKCS#12 container")
|
||
+
|
||
+ # Bytes
|
||
+ elif cert_file is None:
|
||
+ try:
|
||
+ cert = x509.load_der_x509_certificate(data, default_backend())
|
||
+ except Exception:
|
||
+ cert = x509.load_pem_x509_certificate(data, default_backend())
|
||
+
|
||
+ # File
|
||
else:
|
||
- cert = x509.load_pem_x509_certificate(f.read(), default_backend())
|
||
+ try:
|
||
+ cert = x509.load_pem_x509_certificate(data, default_backend())
|
||
+ except Exception:
|
||
+ cert = x509.load_der_x509_certificate(data, default_backend())
|
||
+
|
||
+ except ValueError as ve:
|
||
+ raise ValueError(f"Unable to load certificate '{cert_file}': {ve}")
|
||
|
||
try:
|
||
+ # Check key usage
|
||
key_usage = cert.extensions.get_extension_for_oid(x509.oid.ExtensionOID.KEY_USAGE)
|
||
if not key_usage.value.key_cert_sign:
|
||
return False
|
||
+
|
||
+ # Check constraints
|
||
basic_constraints = cert.extensions.get_extension_for_oid(
|
||
x509.oid.ExtensionOID.BASIC_CONSTRAINTS
|
||
)
|
||
@@ -2011,10 +2059,29 @@ def cert_is_ca(cert_file_name):
|
||
return False
|
||
else:
|
||
return True
|
||
+
|
||
except x509.ExtensionNotFound:
|
||
# No extensions, check the cert info directly
|
||
- return check_cert_info(cert_file_name, "CA:TRUE")
|
||
+ return check_cert_info(cert_file, "CA:TRUE")
|
||
+
|
||
+def pem_to_der(blob: bytes):
|
||
+ """
|
||
+ Convert PEM certificate bytes to DER format.
|
||
|
||
+ :param blob: PEM encoded certificate bytes
|
||
+ :return: DER encoded certificate bytes
|
||
+ """
|
||
+ cert = x509.load_pem_x509_certificate(blob)
|
||
+ return cert.public_bytes(serialization.Encoding.DER)
|
||
+
|
||
+def is_pem_cert(blob: bytes):
|
||
+ """
|
||
+ Check if the given blob is a PEM certificate.
|
||
+
|
||
+ :param blob: Certificate data bytes
|
||
+ :return: True if PEM format, else False
|
||
+ """
|
||
+ return b"-----BEGIN CERTIFICATE-----" in blob
|
||
|
||
def get_passwd_from_file(passwd_file):
|
||
if os.path.exists(passwd_file):
|
||
--
|
||
2.52.0
|
||
|