misc: resolve comments

lib/charms/glauth_k8s/v0/ldap.py

@@ -151,7 +151,7 @@ def _on_ldap_requested(self, event: LdapRequestedEvent) -> None:
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 2
+LIBPATCH = 1
 
 PYDEPS = ["pydantic~=2.5.3"]
 

lib/charms/tls_certificates_interface/v2/tls_certificates.py

@@ -308,7 +308,7 @@ def _on_all_certificates_invalidated(self, event: AllCertificatesInvalidatedEven
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 20
+LIBPATCH = 21
 
 PYDEPS = ["cryptography", "jsonschema"]
 
@@ -693,6 +693,105 @@ def generate_ca(
     return cert.public_bytes(serialization.Encoding.PEM)
 
 
+def get_certificate_extensions(
+    authority_key_identifier: bytes,
+    csr: x509.CertificateSigningRequest,
+    alt_names: Optional[List[str]],
+    is_ca: bool,
+) -> List[x509.Extension]:
+    """Generates a list of certificate extensions from a CSR and other known information.
+
+    Args:
+        authority_key_identifier (bytes): Authority key identifier
+        csr (x509.CertificateSigningRequest): CSR
+        alt_names (list): List of alt names to put on cert - prefer putting SANs in CSR
+        is_ca (bool): Whether the certificate is a CA certificate
+
+    Returns:
+        List[x509.Extension]: List of extensions
+    """
+    cert_extensions_list: List[x509.Extension] = [
+        x509.Extension(
+            oid=ExtensionOID.AUTHORITY_KEY_IDENTIFIER,
+            value=x509.AuthorityKeyIdentifier(
+                key_identifier=authority_key_identifier,
+                authority_cert_issuer=None,
+                authority_cert_serial_number=None,
+            ),
+            critical=False,
+        ),
+        x509.Extension(
+            oid=ExtensionOID.SUBJECT_KEY_IDENTIFIER,
+            value=x509.SubjectKeyIdentifier.from_public_key(csr.public_key()),
+            critical=False,
+        ),
+        x509.Extension(
+            oid=ExtensionOID.BASIC_CONSTRAINTS,
+            critical=True,
+            value=x509.BasicConstraints(ca=is_ca, path_length=None),
+        ),
+    ]
+
+    sans: List[x509.GeneralName] = []
+    san_alt_names = [x509.DNSName(name) for name in alt_names] if alt_names else []
+    sans.extend(san_alt_names)
+    try:
+        loaded_san_ext = csr.extensions.get_extension_for_class(x509.SubjectAlternativeName)
+        sans.extend(
+            [x509.DNSName(name) for name in loaded_san_ext.value.get_values_for_type(x509.DNSName)]
+        )
+        sans.extend(
+            [x509.IPAddress(ip) for ip in loaded_san_ext.value.get_values_for_type(x509.IPAddress)]
+        )
+        sans.extend(
+            [
+                x509.RegisteredID(oid)
+                for oid in loaded_san_ext.value.get_values_for_type(x509.RegisteredID)
+            ]
+        )
+    except x509.ExtensionNotFound:
+        pass
+
+    if sans:
+        cert_extensions_list.append(
+            x509.Extension(
+                oid=ExtensionOID.SUBJECT_ALTERNATIVE_NAME,
+                critical=False,
+                value=x509.SubjectAlternativeName(sans),
+            )
+        )
+
+    if is_ca:
+        cert_extensions_list.append(
+            x509.Extension(
+                ExtensionOID.KEY_USAGE,
+                critical=True,
+                value=x509.KeyUsage(
+                    digital_signature=False,
+                    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,
+                ),
+            )
+        )
+
+    existing_oids = {ext.oid for ext in cert_extensions_list}
+    for extension in csr.extensions:
+        if extension.oid == ExtensionOID.SUBJECT_ALTERNATIVE_NAME:
+            continue
+        if extension.oid in existing_oids:
+            logger.warning("Extension %s is managed by the TLS provider, ignoring.", extension.oid)
+            continue
+        cert_extensions_list.append(extension)
+
+    return cert_extensions_list
+
+
 def generate_certificate(
     csr: bytes,
     ca: bytes,
@@ -730,74 +829,24 @@ def generate_certificate(
         .serial_number(x509.random_serial_number())
         .not_valid_before(datetime.utcnow())
         .not_valid_after(datetime.utcnow() + timedelta(days=validity))
-        .add_extension(
-            x509.AuthorityKeyIdentifier(
-                key_identifier=ca_pem.extensions.get_extension_for_class(
-                    x509.SubjectKeyIdentifier
-                ).value.key_identifier,
-                authority_cert_issuer=None,
-                authority_cert_serial_number=None,
-            ),
-            critical=False,
-        )
-        .add_extension(
-            x509.SubjectKeyIdentifier.from_public_key(csr_object.public_key()), critical=False
-        )
     )
-
-    extensions_list = csr_object.extensions
-    san_ext: Optional[x509.Extension] = None
-    if alt_names:
-        full_sans_dns = alt_names.copy()
+    extensions = get_certificate_extensions(
+        authority_key_identifier=ca_pem.extensions.get_extension_for_class(
+            x509.SubjectKeyIdentifier
+        ).value.key_identifier,
+        csr=csr_object,
+        alt_names=alt_names,
+        is_ca=is_ca,
+    )
+    for extension in extensions:
         try:
-            loaded_san_ext = csr_object.extensions.get_extension_for_class(
-                x509.SubjectAlternativeName
+            certificate_builder = certificate_builder.add_extension(
+                extval=extension.value,
+                critical=extension.critical,
             )
-            full_sans_dns.extend(loaded_san_ext.value.get_values_for_type(x509.DNSName))
-        except ExtensionNotFound:
-            pass
-        finally:
-            san_ext = Extension(
-                ExtensionOID.SUBJECT_ALTERNATIVE_NAME,
-                False,
-                x509.SubjectAlternativeName([x509.DNSName(name) for name in full_sans_dns]),
-            )
-            if not extensions_list:
-                extensions_list = x509.Extensions([san_ext])
-
-    for extension in extensions_list:
-        if extension.value.oid == ExtensionOID.SUBJECT_ALTERNATIVE_NAME and san_ext:
-            extension = san_ext
-
-        certificate_builder = certificate_builder.add_extension(
-            extension.value,
-            critical=extension.critical,
-        )
-
-    if is_ca:
-        certificate_builder = certificate_builder.add_extension(
-            x509.BasicConstraints(ca=True, path_length=None), critical=True
-        )
-        certificate_builder = certificate_builder.add_extension(
-            x509.KeyUsage(
-                digital_signature=False,
-                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,
-            ),
-            critical=True,
-        )
-    else:
-        certificate_builder = certificate_builder.add_extension(
-            x509.BasicConstraints(ca=False, path_length=None), critical=False
-        )
+        except ValueError as e:
+            logger.warning("Failed to add extension %s: %s", extension.oid, e)
 
-    certificate_builder._version = x509.Version.v3
     cert = certificate_builder.sign(private_key, hashes.SHA256())  # type: ignore[arg-type]
     return cert.public_bytes(serialization.Encoding.PEM)
 

metadata.yaml

@@ -53,3 +53,7 @@ provides:
     description: |
       Transfer certificates to client charmed operators.
     interface: certificate_transfer
+
+peers:
+  glauth-peers:
+    interface: glauth_peers

src/charm.py

@@ -18,8 +18,8 @@
 from charms.glauth_utils.v0.glauth_auxiliary import AuxiliaryProvider, AuxiliaryRequestedEvent
 from charms.grafana_k8s.v0.grafana_dashboard import GrafanaDashboardProvider
 from charms.loki_k8s.v0.loki_push_api import LogProxyConsumer, PromtailDigestError
+from charms.observability_libs.v0.cert_handler import CertChanged
 from charms.observability_libs.v0.kubernetes_service_patch import KubernetesServicePatch
-from charms.observability_libs.v1.cert_handler import CertChanged
 from charms.prometheus_k8s.v0.prometheus_scrape import MetricsEndpointProvider
 from configs import ConfigFile, DatabaseConfig, pebble_layer
 from constants import (

src/integrations.py

@@ -1,7 +1,7 @@
 import hashlib
-import json
 import logging
 import subprocess
+from contextlib import suppress
 from dataclasses import dataclass
 from secrets import token_bytes
 from typing import Optional
@@ -11,7 +11,7 @@
 )
 from charms.glauth_k8s.v0.ldap import LdapProviderBaseData, LdapProviderData
 from charms.glauth_utils.v0.glauth_auxiliary import AuxiliaryData
-from charms.observability_libs.v1.cert_handler import CertHandler
+from charms.observability_libs.v0.cert_handler import CertHandler
 from configs import DatabaseConfig
 from constants import (
     CERTIFICATE_FILE,
@@ -25,6 +25,7 @@
 )
 from database import Capability, Group, Operation, User
 from ops.charm import CharmBase
+from ops.pebble import PathError
 from tenacity import (
     Retrying,
     retry_if_exception_type,
@@ -137,25 +138,26 @@ def __init__(self, charm: CharmBase) -> None:
         self.cert_handler = CertHandler(
             charm,
             key="glauth-server-cert",
+            peer_relation_name="glauth-peers",
             cert_subject=hostname,
-            sans=[hostname],
+            extra_sans_dns=[hostname],
         )
 
     @property
     def _ca_cert(self) -> Optional[str]:
-        return self.cert_handler.ca_cert
+        return self.cert_handler.ca
 
     @property
     def _server_key(self) -> Optional[str]:
-        return self.cert_handler.server_cert
+        return self.cert_handler.key
 
     @property
     def _server_cert(self) -> Optional[str]:
-        return self.cert_handler.server_cert
+        return self.cert_handler.cert
 
     @property
     def _ca_chain(self) -> list[str]:
-        return json.loads(self.cert_handler.chain or "[]")
+        return self.cert_handler.chain
 
     @property
     def cert_data(self) -> CertificateData:
@@ -183,9 +185,9 @@ def certs_ready(self) -> bool:
         return all((self._ca_cert, self._ca_chain, self._server_key, self._server_cert))
 
     def _prepare_certificates(self) -> None:
-        SERVER_CA_CERT.write_text(self.cert_handler.ca_cert)  # type: ignore[arg-type]
-        SERVER_KEY.write_text(self.cert_handler.private_key)  # type: ignore[arg-type]
-        SERVER_CERT.write_text(self.cert_handler.server_cert)  # type: ignore[arg-type]
+        SERVER_CA_CERT.write_text(self._ca_cert)  # type: ignore[arg-type]
+        SERVER_KEY.write_text(self._server_key)  # type: ignore[arg-type]
+        SERVER_CERT.write_text(self._server_cert)  # type: ignore[arg-type]
 
         try:
             for attempt in Retrying(
@@ -211,10 +213,9 @@ def _add_certificates(self) -> None:
         self._container.push(SERVER_CERT, self._server_cert, make_dirs=True)
 
     def _remove_certificates(self) -> None:
-        self._container.remove(CERTIFICATE_FILE)
-        self._container.remove_path(SERVER_CA_CERT)
-        self._container.remove_path(SERVER_KEY)
-        self._container.remove_path(SERVER_CERT)
+        with suppress(PathError):
+            for file in (CERTIFICATE_FILE, SERVER_CA_CERT, SERVER_KEY, SERVER_CERT):
+                self._container.remove_path(file)
 
 
 class CertificatesTransferIntegration: