diff --git a/lemur/certificates/service.py b/lemur/certificates/service.py index b05f489a0..1c47880f5 100644 --- a/lemur/certificates/service.py +++ b/lemur/certificates/service.py @@ -1046,9 +1046,31 @@ def is_attached_to_endpoint(certificate_name, endpoint_name): This method talks to elb and finds the real time information. :param certificate_name: :param endpoint_name: - :return: True if certificate is attached to the given endpoint, False otherwise + :return: True if certificate is attached to the given endpoint, False otherwise. + Returns False when the endpoint no longer exists (nothing to be attached to). + When the source plugin cannot verify attachment (does not implement + get_endpoint_certificate_names()), returns True (fail-closed) so a possibly + still-deployed certificate is never silently revoked. """ endpoint = endpoint_service.get_by_name(endpoint_name) + if endpoint is None: + current_app.logger.error( + "Endpoint %s not found; assuming certificate %s is not attached", + endpoint_name, + certificate_name, + ) + capture_exception() + return False + if not hasattr(endpoint.source.plugin, "get_endpoint_certificate_names"): + current_app.logger.error( + "Source plugin %s does not implement 'get_endpoint_certificate_names()'; " + "assuming certificate %s is attached to endpoint %s (fail-closed)", + endpoint.source.plugin_name, + certificate_name, + endpoint_name, + ) + capture_exception() + return True attached_certificates = endpoint.source.plugin.get_endpoint_certificate_names( endpoint ) diff --git a/lemur/tests/test_certificates.py b/lemur/tests/test_certificates.py index 53be9a845..50a5e4f34 100644 --- a/lemur/tests/test_certificates.py +++ b/lemur/tests/test_certificates.py @@ -2037,3 +2037,74 @@ def test_send_source_destination_pairing_metrics(certificate): assert dst_tags_by_name["shared"]["datacenter"] == "us1.prod" assert dst_tags_by_name["orphan-dst"]["has_source"] == "false" assert "datacenter" not in dst_tags_by_name["orphan-dst"] +# is_attached_to_endpoint — hasattr guard tests +# --------------------------------------------------------------------------- + + +def test_is_attached_to_endpoint_plugin_missing_method(app): + """Plugin without get_endpoint_certificate_names returns True (fail-closed, no AttributeError).""" + from unittest.mock import MagicMock, patch + from lemur.certificates.service import is_attached_to_endpoint + + plugin = MagicMock(spec=[]) # spec=[] → hasattr returns False for everything + plugin.plugin_name = "test-no-method-source" + source = MagicMock() + source.plugin = plugin + endpoint = MagicMock() + endpoint.source = source + + with patch("lemur.certificates.service.endpoint_service") as mock_ep_svc: + mock_ep_svc.get_by_name.return_value = endpoint + result = is_attached_to_endpoint("my-cert", "my-endpoint") + + assert result is True + + +def test_is_attached_to_endpoint_endpoint_not_found(app): + """Endpoint lookup returning None returns False (no AttributeError) — nothing to be attached to.""" + from unittest.mock import patch + from lemur.certificates.service import is_attached_to_endpoint + + with patch("lemur.certificates.service.endpoint_service") as mock_ep_svc: + mock_ep_svc.get_by_name.return_value = None + result = is_attached_to_endpoint("my-cert", "missing-endpoint") + + assert result is False + + +def test_is_attached_to_endpoint_plugin_has_method_cert_present(): + """Plugin with get_endpoint_certificate_names returns True when cert is in list.""" + from unittest.mock import MagicMock, patch + from lemur.certificates.service import is_attached_to_endpoint + + plugin = MagicMock() + plugin.get_endpoint_certificate_names.return_value = ["my-cert", "other-cert"] + source = MagicMock() + source.plugin = plugin + endpoint = MagicMock() + endpoint.source = source + + with patch("lemur.certificates.service.endpoint_service") as mock_ep_svc: + mock_ep_svc.get_by_name.return_value = endpoint + result = is_attached_to_endpoint("my-cert", "my-endpoint") + + assert result is True + + +def test_is_attached_to_endpoint_plugin_has_method_cert_absent(): + """Plugin with get_endpoint_certificate_names returns False when cert not in list.""" + from unittest.mock import MagicMock, patch + from lemur.certificates.service import is_attached_to_endpoint + + plugin = MagicMock() + plugin.get_endpoint_certificate_names.return_value = ["other-cert"] + source = MagicMock() + source.plugin = plugin + endpoint = MagicMock() + endpoint.source = source + + with patch("lemur.certificates.service.endpoint_service") as mock_ep_svc: + mock_ep_svc.get_by_name.return_value = endpoint + result = is_attached_to_endpoint("my-cert", "my-endpoint") + + assert result is False