Exclude expired disabled NPM certificate dependencies

This commit is contained in:
Lyncoln
2026-09-20 10:53:26 -04:00
parent e4ca80d4d5
commit a93e067bfd
2 changed files with 289 additions and 1 deletions
+156
View File
@@ -11,9 +11,11 @@ from dns_sync.adapters import (
DiscoveredObject,
Endpoint,
NormalizedObject,
ObjectReference,
Observation,
)
from dns_sync.orchestrator import discover_reconcile_plan
from dns_sync.reconciliation import ReconciliationConflict
from dns_sync.persistence import SQLiteStateStore
@@ -343,6 +345,160 @@ class ObservationAuthorityTests(unittest.TestCase):
persisted,
)
def test_expired_certificate_on_disabled_proxy_is_removed_from_desired(
self,
):
endpoint = Endpoint(
"npm",
"192.168.1.3:81",
)
adapter = FakeAdapter(
endpoint=endpoint,
capabilities=capability_set(
CERTIFICATES,
PROXY_HOSTS,
),
discovered={
CERTIFICATES: [
DiscoveredObject(
object_type=CERTIFICATES,
local_key="56",
payload={
"domain_names": [
"admin.trunavigate.org"
],
"provider": "letsencrypt",
"valid_from": None,
"expires_on": (
"2026-06-22T03:40:57.000Z"
),
},
)
],
PROXY_HOSTS: [
DiscoveredObject(
object_type=PROXY_HOSTS,
local_key="29",
payload={
"domain_names": [
"admin.trunavigate.org"
],
"forward_scheme": "http",
"forward_host": "192.168.1.4",
"forward_port": 3283,
"enabled": False,
},
references=(
ObjectReference(
relation="certificate",
object_type=CERTIFICATES,
local_key="56",
),
),
)
],
},
)
with SQLiteStateStore(":memory:") as store:
result = discover_reconcile_plan(
[adapter],
store,
target_ip="192.168.1.101",
run_at=T2,
)
certificates = [
object
for object in result.desired
if object.object_type == CERTIFICATES
]
proxies = [
object
for object in result.desired
if object.object_type == PROXY_HOSTS
]
self.assertEqual(certificates, [])
self.assertEqual(len(proxies), 1)
self.assertEqual(
proxies[0].logical_key,
"admin.trunavigate.org",
)
self.assertEqual(
proxies[0].references,
(),
)
def test_expired_certificate_on_enabled_proxy_fails_closed(
self,
):
endpoint = Endpoint(
"npm",
"192.168.1.3:81",
)
adapter = FakeAdapter(
endpoint=endpoint,
capabilities=capability_set(
CERTIFICATES,
PROXY_HOSTS,
),
discovered={
CERTIFICATES: [
DiscoveredObject(
object_type=CERTIFICATES,
local_key="56",
payload={
"domain_names": [
"active.example.com"
],
"provider": "letsencrypt",
"valid_from": None,
"expires_on": (
"2026-06-22T03:40:57.000Z"
),
},
)
],
PROXY_HOSTS: [
DiscoveredObject(
object_type=PROXY_HOSTS,
local_key="29",
payload={
"domain_names": [
"active.example.com"
],
"forward_scheme": "http",
"forward_host": "192.168.1.4",
"forward_port": 3283,
"enabled": True,
},
references=(
ObjectReference(
relation="certificate",
object_type=CERTIFICATES,
local_key="56",
),
),
)
],
},
)
with SQLiteStateStore(":memory:") as store:
with self.assertRaisesRegex(
ReconciliationConflict,
"Enabled proxy host references an expired certificate",
):
discover_reconcile_plan(
[adapter],
store,
target_ip="192.168.1.101",
run_at=T2,
)
def test_all_failed_sources_preserve_persisted_desired_without_planning(self):
npm_endpoint = Endpoint(
"npm",