diff --git a/dns_sync/certificates/material.py b/dns_sync/certificates/material.py new file mode 100644 index 0000000..48cb946 --- /dev/null +++ b/dns_sync/certificates/material.py @@ -0,0 +1,51 @@ +from dataclasses import dataclass + + +@dataclass(frozen=True, repr=False) +class CertificateMaterial: + """ + Secret-bearing certificate material for transient use only. + + Instances must never be persisted, placed in generic reconciliation + objects, or included in logs. + """ + + certificate: bytes + certificate_key: bytes + intermediate_certificate: bytes | None = None + + def __post_init__(self) -> None: + if not isinstance(self.certificate, bytes) or not self.certificate: + raise ValueError( + "Certificate material requires certificate bytes" + ) + + if ( + not isinstance(self.certificate_key, bytes) + or not self.certificate_key + ): + raise ValueError( + "Certificate material requires certificate-key bytes" + ) + + if self.intermediate_certificate is not None and ( + not isinstance(self.intermediate_certificate, bytes) + or not self.intermediate_certificate + ): + raise ValueError( + "Intermediate certificate must be non-empty bytes" + ) + + def __repr__(self) -> str: + intermediate = ( + "" + if self.intermediate_certificate is not None + else "None" + ) + return ( + "CertificateMaterial(" + "certificate=, " + "certificate_key=, " + f"intermediate_certificate={intermediate}" + ")" + ) diff --git a/dns_sync/npm_client.py b/dns_sync/npm_client.py index be124ad..d2cf7fc 100644 --- a/dns_sync/npm_client.py +++ b/dns_sync/npm_client.py @@ -1,10 +1,14 @@ import os +from io import BytesIO +from pathlib import PurePosixPath from typing import Any +from zipfile import BadZipFile, ZipFile import requests import urllib3 from .adapters import Endpoint +from .certificates.material import CertificateMaterial class NpmClient: @@ -236,6 +240,138 @@ class NpmClient: ) return data + @staticmethod + def _pem_bytes(value: Any, label: str) -> bytes: + if not isinstance(value, str) or not value.strip(): + raise RuntimeError( + f"NPM certificate is missing {label}" + ) + return value.encode("utf-8") + + @staticmethod + def _zip_pem_member( + archive: ZipFile, + prefix: str, + ) -> bytes: + matches = [] + + for info in archive.infolist(): + if info.is_dir(): + continue + + basename = PurePosixPath(info.filename).name + + if not basename.endswith(".pem"): + continue + + stem = basename[:-4] + + if not stem.startswith(prefix): + continue + + suffix = stem[len(prefix):] + + if not suffix.isdigit(): + continue + + matches.append(info) + + if len(matches) != 1: + raise RuntimeError( + "NPM certificate download must contain exactly one " + f"{prefix}N.pem member" + ) + + data = archive.read(matches[0]) + + if not data: + raise RuntimeError( + f"NPM certificate download {prefix} member is empty" + ) + + return data + + def certificate_material( + self, + local_id: str | int, + ) -> CertificateMaterial: + certificate_id = self._local_id( + local_id, + "certificate", + ) + + detail = self.certificate(certificate_id) + provider = str( + detail.get("provider") or "" + ).strip().lower() + + if provider == "other": + meta = detail.get("meta") + if not isinstance(meta, dict): + raise RuntimeError( + "NPM custom certificate has invalid metadata" + ) + + intermediate = meta.get( + "intermediate_certificate" + ) + + return CertificateMaterial( + certificate=self._pem_bytes( + meta.get("certificate"), + "certificate", + ), + certificate_key=self._pem_bytes( + meta.get("certificate_key"), + "certificate_key", + ), + intermediate_certificate=( + None + if intermediate in (None, "") + else self._pem_bytes( + intermediate, + "intermediate_certificate", + ) + ), + ) + + if provider == "letsencrypt": + response = self.request( + "GET", + ( + "/nginx/certificates/" + f"{certificate_id}/download" + ), + expected_status=200, + ) + + try: + with ZipFile(BytesIO(response.content)) as archive: + return CertificateMaterial( + certificate=self._zip_pem_member( + archive, + "cert", + ), + certificate_key=self._zip_pem_member( + archive, + "privkey", + ), + intermediate_certificate=( + self._zip_pem_member( + archive, + "chain", + ) + ), + ) + except BadZipFile as exc: + raise RuntimeError( + "NPM certificate download was not a valid ZIP archive" + ) from exc + + raise RuntimeError( + f"Unsupported NPM certificate provider: {provider!r}" + ) + def create_certificate( self, payload: dict[str, Any], @@ -258,7 +394,7 @@ class NpmClient: certificate: bytes, certificate_key: bytes, intermediate_certificate: bytes | None = None, - ) -> dict[str, Any]: + ) -> None: certificate_id = self._local_id( local_id, "certificate", @@ -290,14 +426,9 @@ class NpmClient: intermediate_certificate, ) - response = self.request( + self.request( "POST", f"/nginx/certificates/{certificate_id}/upload", expected_status=200, files=files, ) - - return self._json_object( - response, - "certificate upload", - ) diff --git a/tests/test_npm_client.py b/tests/test_npm_client.py index 47520a2..b82e97f 100644 --- a/tests/test_npm_client.py +++ b/tests/test_npm_client.py @@ -1,9 +1,12 @@ import os import unittest +from io import BytesIO +from zipfile import ZipFile from unittest.mock import Mock, patch from dns_sync.adapters import Endpoint from dns_sync.config import AppConfig +from dns_sync.certificates.material import CertificateMaterial from dns_sync.npm_client import NpmClient @@ -97,6 +100,31 @@ class NpmClientMutationTests(unittest.TestCase): response.json.return_value = data return response + + @staticmethod + def certificate_zip(): + buffer = BytesIO() + + with ZipFile(buffer, "w") as archive: + archive.writestr( + "cert4.pem", + b"LEAF-CERTIFICATE", + ) + archive.writestr( + "fullchain4.pem", + b"FULLCHAIN", + ) + archive.writestr( + "chain4.pem", + b"INTERMEDIATE-CHAIN", + ) + archive.writestr( + "privkey4.pem", + b"PRIVATE-KEY", + ) + + return buffer.getvalue() + def test_gets_one_proxy_host_by_endpoint_local_id(self): client = self.client() client.get_json = Mock( @@ -174,6 +202,125 @@ class NpmClientMutationTests(unittest.TestCase): ) self.assertEqual(result["id"], 67) + def test_reads_custom_certificate_material_from_metadata(self): + client = self.client() + client.certificate = Mock( + return_value={ + "id": 67, + "provider": "other", + "meta": { + "certificate": "CUSTOM-CERTIFICATE", + "certificate_key": "CUSTOM-PRIVATE-KEY", + "intermediate_certificate": "CUSTOM-CHAIN", + }, + } + ) + + material = client.certificate_material("67") + + self.assertEqual( + material, + CertificateMaterial( + certificate=b"CUSTOM-CERTIFICATE", + certificate_key=b"CUSTOM-PRIVATE-KEY", + intermediate_certificate=b"CUSTOM-CHAIN", + ), + ) + + client.certificate.assert_called_once_with("67") + + def test_reads_letsencrypt_material_from_download_zip(self): + client = self.client() + client.certificate = Mock( + return_value={ + "id": 10, + "provider": "letsencrypt", + } + ) + + response = Mock() + response.content = self.certificate_zip() + client.request = Mock(return_value=response) + + material = client.certificate_material("10") + + self.assertEqual( + material.certificate, + b"LEAF-CERTIFICATE", + ) + self.assertEqual( + material.certificate_key, + b"PRIVATE-KEY", + ) + self.assertEqual( + material.intermediate_certificate, + b"INTERMEDIATE-CHAIN", + ) + + client.request.assert_called_once_with( + "GET", + "/nginx/certificates/10/download", + expected_status=200, + ) + + def test_certificate_material_repr_redacts_secrets(self): + material = CertificateMaterial( + certificate=b"VISIBLE-CERT-SECRET", + certificate_key=b"VISIBLE-KEY-SECRET", + intermediate_certificate=b"VISIBLE-CHAIN-SECRET", + ) + + rendered = repr(material) + + self.assertNotIn( + "VISIBLE-CERT-SECRET", + rendered, + ) + self.assertNotIn( + "VISIBLE-KEY-SECRET", + rendered, + ) + self.assertNotIn( + "VISIBLE-CHAIN-SECRET", + rendered, + ) + self.assertIn("", rendered) + + def test_certificate_material_rejects_missing_secret_parts(self): + client = self.client() + + client.certificate = Mock( + return_value={ + "id": 67, + "provider": "other", + "meta": { + "certificate": "CERT", + "certificate_key": "", + }, + } + ) + + with self.assertRaisesRegex( + RuntimeError, + "certificate_key", + ): + client.certificate_material("67") + + def test_certificate_material_rejects_unknown_provider(self): + client = self.client() + client.certificate = Mock( + return_value={ + "id": 99, + "provider": "mystery-provider", + } + ) + + with self.assertRaisesRegex( + RuntimeError, + "Unsupported NPM certificate provider", + ): + client.certificate_material("99") + def test_creates_certificate_metadata_with_json_payload(self): client = self.client() payload = { @@ -200,9 +347,13 @@ class NpmClientMutationTests(unittest.TestCase): def test_uploads_certificate_material_as_multipart(self): client = self.client() - client.request = Mock( - return_value=self.response({"id": 84}) + response = self.response( + { + "certificate": "SECRET-CERTIFICATE", + "certificate_key": "SECRET-PRIVATE-KEY", + } ) + client.request = Mock(return_value=response) result = client.upload_certificate( "84", @@ -230,7 +381,8 @@ class NpmClientMutationTests(unittest.TestCase): ), }, ) - self.assertEqual(result["id"], 84) + self.assertIsNone(result) + response.json.assert_not_called() def test_certificate_upload_requires_both_required_parts(self): client = self.client()