450 lines
12 KiB
Python
450 lines
12 KiB
Python
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
|
|
|
|
|
|
class NpmClientEndpointTests(unittest.TestCase):
|
|
def config(self):
|
|
return AppConfig.from_env()
|
|
|
|
def test_default_client_uses_first_configured_endpoint(self):
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NPM_ENDPOINTS": "192.168.1.3,192.168.1.4:8081",
|
|
"DNS_TARGET_IP": "192.168.1.101",
|
|
"NPM_IDENTITY": "test@example.com",
|
|
"NPM_SECRET": "not-a-real-secret",
|
|
},
|
|
clear=True,
|
|
):
|
|
config = self.config()
|
|
with patch(
|
|
"dns_sync.npm_client.requests.Session.request"
|
|
) as request:
|
|
client = NpmClient(config)
|
|
|
|
request.assert_not_called()
|
|
self.assertEqual(client.api_url, "http://192.168.1.3:81/api")
|
|
|
|
def test_client_can_select_specific_configured_endpoint(self):
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NPM_ENDPOINTS": "192.168.1.3,192.168.1.4:8081",
|
|
"DNS_TARGET_IP": "192.168.1.101",
|
|
"NPM_IDENTITY": "test@example.com",
|
|
"NPM_SECRET": "not-a-real-secret",
|
|
},
|
|
clear=True,
|
|
):
|
|
config = self.config()
|
|
client = NpmClient(config, config.npm_endpoints[1])
|
|
|
|
self.assertEqual(
|
|
client.api_url,
|
|
"http://192.168.1.4:8081/api",
|
|
)
|
|
self.assertEqual(
|
|
client.endpoint.identity,
|
|
"npm://192.168.1.4:8081",
|
|
)
|
|
|
|
def test_client_rejects_non_npm_endpoint(self):
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"PROXY_IP": "192.168.1.101",
|
|
"NPM_IDENTITY": "test@example.com",
|
|
"NPM_SECRET": "not-a-real-secret",
|
|
},
|
|
clear=True,
|
|
):
|
|
config = self.config()
|
|
endpoint = Endpoint("verizon", "192.168.1.1")
|
|
|
|
with self.assertRaisesRegex(RuntimeError, "npm endpoint"):
|
|
NpmClient(config, endpoint)
|
|
|
|
|
|
|
|
|
|
class NpmClientMutationTests(unittest.TestCase):
|
|
def client(self):
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"NPM_ENDPOINTS": "192.168.1.3",
|
|
"DNS_TARGET_IP": "192.168.1.101",
|
|
"NPM_IDENTITY": "test@example.com",
|
|
"NPM_SECRET": "not-a-real-secret",
|
|
},
|
|
clear=True,
|
|
):
|
|
config = AppConfig.from_env()
|
|
return NpmClient(
|
|
config,
|
|
config.npm_endpoints[0],
|
|
)
|
|
|
|
@staticmethod
|
|
def response(data):
|
|
response = Mock()
|
|
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(
|
|
return_value={
|
|
"id": 12,
|
|
"domain_names": ["app.example.com"],
|
|
}
|
|
)
|
|
|
|
result = client.proxy_host("12")
|
|
|
|
client.get_json.assert_called_once_with(
|
|
"/nginx/proxy-hosts/12"
|
|
)
|
|
self.assertEqual(result["id"], 12)
|
|
|
|
def test_creates_proxy_host_with_json_payload(self):
|
|
client = self.client()
|
|
payload = {
|
|
"domain_names": ["app.example.com"],
|
|
"forward_scheme": "http",
|
|
"forward_host": "192.168.1.20",
|
|
"forward_port": 8080,
|
|
"certificate_id": 67,
|
|
}
|
|
client.request = Mock(
|
|
return_value=self.response({"id": 12})
|
|
)
|
|
|
|
result = client.create_proxy_host(payload)
|
|
|
|
client.request.assert_called_once_with(
|
|
"POST",
|
|
"/nginx/proxy-hosts",
|
|
expected_status=201,
|
|
json=payload,
|
|
)
|
|
self.assertEqual(result["id"], 12)
|
|
|
|
def test_updates_proxy_host_with_json_payload(self):
|
|
client = self.client()
|
|
payload = {
|
|
"forward_host": "192.168.1.30",
|
|
}
|
|
client.request = Mock(
|
|
return_value=self.response({"id": 12})
|
|
)
|
|
|
|
result = client.update_proxy_host(
|
|
"12",
|
|
payload,
|
|
)
|
|
|
|
client.request.assert_called_once_with(
|
|
"PUT",
|
|
"/nginx/proxy-hosts/12",
|
|
expected_status=200,
|
|
json=payload,
|
|
)
|
|
self.assertEqual(result["id"], 12)
|
|
|
|
def test_deletes_proxy_host_by_endpoint_local_id(self):
|
|
client = self.client()
|
|
client.request = Mock()
|
|
|
|
result = client.delete_proxy_host("12")
|
|
|
|
self.assertIsNone(result)
|
|
client.request.assert_called_once_with(
|
|
"DELETE",
|
|
"/nginx/proxy-hosts/12",
|
|
expected_status=200,
|
|
)
|
|
|
|
def test_gets_one_certificate_by_endpoint_local_id(self):
|
|
client = self.client()
|
|
client.get_json = Mock(
|
|
return_value={
|
|
"id": 67,
|
|
"domain_names": ["app.example.com"],
|
|
}
|
|
)
|
|
|
|
result = client.certificate("67")
|
|
|
|
client.get_json.assert_called_once_with(
|
|
"/nginx/certificates/67"
|
|
)
|
|
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("<redacted>", 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_deletes_certificate_by_endpoint_local_id(self):
|
|
client = self.client()
|
|
client.request = Mock()
|
|
|
|
result = client.delete_certificate("84")
|
|
|
|
self.assertIsNone(result)
|
|
client.request.assert_called_once_with(
|
|
"DELETE",
|
|
"/nginx/certificates/84",
|
|
expected_status=200,
|
|
)
|
|
|
|
def test_creates_certificate_metadata_with_json_payload(self):
|
|
client = self.client()
|
|
payload = {
|
|
"provider": "other",
|
|
"nice_name": "cluster copy",
|
|
"domain_names": ["app.example.com"],
|
|
"meta": {
|
|
"cluster_managed": True,
|
|
},
|
|
}
|
|
client.request = Mock(
|
|
return_value=self.response({"id": 84})
|
|
)
|
|
|
|
result = client.create_certificate(payload)
|
|
|
|
client.request.assert_called_once_with(
|
|
"POST",
|
|
"/nginx/certificates",
|
|
expected_status=201,
|
|
json=payload,
|
|
)
|
|
self.assertEqual(result["id"], 84)
|
|
|
|
def test_uploads_certificate_material_as_multipart(self):
|
|
client = self.client()
|
|
response = self.response(
|
|
{
|
|
"certificate": "SECRET-CERTIFICATE",
|
|
"certificate_key": "SECRET-PRIVATE-KEY",
|
|
}
|
|
)
|
|
client.request = Mock(return_value=response)
|
|
|
|
result = client.upload_certificate(
|
|
"84",
|
|
certificate=b"CERTIFICATE",
|
|
certificate_key=b"PRIVATE-KEY",
|
|
intermediate_certificate=b"INTERMEDIATE",
|
|
)
|
|
|
|
client.request.assert_called_once_with(
|
|
"POST",
|
|
"/nginx/certificates/84/upload",
|
|
expected_status=200,
|
|
files={
|
|
"certificate": (
|
|
"certificate.pem",
|
|
b"CERTIFICATE",
|
|
),
|
|
"certificate_key": (
|
|
"certificate_key.pem",
|
|
b"PRIVATE-KEY",
|
|
),
|
|
"intermediate_certificate": (
|
|
"intermediate_certificate.pem",
|
|
b"INTERMEDIATE",
|
|
),
|
|
},
|
|
)
|
|
self.assertIsNone(result)
|
|
response.json.assert_not_called()
|
|
|
|
def test_certificate_upload_requires_both_required_parts(self):
|
|
client = self.client()
|
|
|
|
with self.assertRaisesRegex(
|
|
RuntimeError,
|
|
"requires certificate data",
|
|
):
|
|
client.upload_certificate(
|
|
"84",
|
|
certificate=b"",
|
|
certificate_key=b"PRIVATE-KEY",
|
|
)
|
|
|
|
with self.assertRaisesRegex(
|
|
RuntimeError,
|
|
"requires certificate key data",
|
|
):
|
|
client.upload_certificate(
|
|
"84",
|
|
certificate=b"CERTIFICATE",
|
|
certificate_key=b"",
|
|
)
|
|
|
|
def test_rejects_invalid_endpoint_local_ids(self):
|
|
client = self.client()
|
|
|
|
for local_id in ("", "abc", True, -1):
|
|
with self.subTest(local_id=local_id):
|
|
with self.assertRaisesRegex(
|
|
RuntimeError,
|
|
"id is invalid",
|
|
):
|
|
client.certificate(local_id)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|