feat(containerregistry): add new check containerregistry_uses_private_link (#5375)

Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com>

tests/providers/azure/services/appinsights/appinsights_ensure_is_configured/appinsights_ensure_is_configured_test.py

@@ -52,7 +52,7 @@ class Test_appinsights_ensure_is_configured:
             assert result[0].location == "global"
             assert (
                 result[0].status_extended
-                == f"There are no AppInsight configured in susbscription {AZURE_SUBSCRIPTION_ID}."
+                == f"There are no AppInsight configured in subscription {AZURE_SUBSCRIPTION_ID}."
             )
 
     def test_appinsights_configured(self):
@@ -89,5 +89,5 @@ class Test_appinsights_ensure_is_configured:
             assert result[0].location == "global"
             assert (
                 result[0].status_extended
-                == f"There is at least one AppInsight configured in susbscription {AZURE_SUBSCRIPTION_ID}."
+                == f"There is at least one AppInsight configured in subscription {AZURE_SUBSCRIPTION_ID}."
             )

tests/providers/azure/services/containerregistry/containerregistry_admin_user_disabled/containerregistry_admin_user_disabled_test.py

@@ -59,6 +59,7 @@ class TestContainerRegistryAdminUserDisabled:
                         admin_user_enabled=True,
                         network_rule_set=None,
                         monitor_diagnostic_settings=[],
+                        private_endpoint_connections=[],
                     )
                 }
             }
@@ -115,6 +116,7 @@ class TestContainerRegistryAdminUserDisabled:
                         admin_user_enabled=False,
                         network_rule_set=None,
                         monitor_diagnostic_settings=[],
+                        private_endpoint_connections=[],
                     )
                 }
             }

tests/providers/azure/services/containerregistry/containerregistry_not_publicly_accessible/containerregistry_not_publicly_accessible_test.py

@@ -60,6 +60,7 @@ class Test_containerregistry_not_publicly_accessible:
                         public_network_access="Enabled",
                         admin_user_enabled=True,
                         network_rule_set=NetworkRuleSet(default_action="Allow"),
+                        private_endpoint_connections=[],
                         monitor_diagnostic_settings=[
                             {
                                 "id": "id1/id1",
@@ -133,6 +134,7 @@ class Test_containerregistry_not_publicly_accessible:
                         public_network_access="Enabled",
                         admin_user_enabled=False,
                         network_rule_set=NetworkRuleSet(default_action="Deny"),
+                        private_endpoint_connections=[],
                         monitor_diagnostic_settings=[
                             {
                                 "id": "id1/id1",

tests/providers/azure/services/containerregistry/containerregistry_service_test.py

@@ -35,6 +35,7 @@ class TestContainerRegistryService:
                         public_network_access="Enabled",
                         admin_user_enabled=True,
                         network_rule_set=None,
+                        private_endpoint_connections=[],
                         monitor_diagnostic_settings=[
                             {
                                 "id": "id1/id1",

tests/providers/azure/services/containerregistry/containerregistry_uses_private_link/containerregistry_uses_private_link_test.py

@@ -0,0 +1,159 @@
+from unittest import mock
+from unittest.mock import MagicMock
+from uuid import uuid4
+
+from azure.mgmt.containerregistry.models import (
+    PrivateEndpoint,
+    PrivateEndpointConnection,
+    PrivateLinkServiceConnectionState,
+)
+
+from tests.providers.azure.azure_fixtures import (
+    AZURE_SUBSCRIPTION_ID,
+    set_mocked_azure_provider,
+)
+
+
+class Test_containerregistry_uses_private_link:
+    def test_no_container_registries(self):
+        containerregistry_client = MagicMock()
+        containerregistry_client.registries = {}
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_azure_provider(),
+        ), mock.patch(
+            "prowler.providers.azure.services.containerregistry.containerregistry_uses_private_link.containerregistry_uses_private_link.containerregistry_client",
+            new=containerregistry_client,
+        ):
+            from prowler.providers.azure.services.containerregistry.containerregistry_uses_private_link.containerregistry_uses_private_link import (
+                containerregistry_uses_private_link,
+            )
+
+            check = containerregistry_uses_private_link()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_container_registry_not_uses_private_link(self):
+        containerregistry_client = MagicMock()
+        registry_id = str(uuid4())
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_azure_provider(),
+        ), mock.patch(
+            "prowler.providers.azure.services.containerregistry.containerregistry_uses_private_link.containerregistry_uses_private_link.containerregistry_client",
+            new=containerregistry_client,
+        ):
+            from prowler.providers.azure.services.containerregistry.containerregistry_service import (
+                ContainerRegistryInfo,
+            )
+            from prowler.providers.azure.services.containerregistry.containerregistry_uses_private_link.containerregistry_uses_private_link import (
+                containerregistry_uses_private_link,
+            )
+
+            containerregistry_client.registries = {
+                AZURE_SUBSCRIPTION_ID: {
+                    registry_id: ContainerRegistryInfo(
+                        id=registry_id,
+                        name="mock_registry",
+                        location="westeurope",
+                        resource_group="mock_resource_group",
+                        sku="Basic",
+                        login_server="mock_login_server.azurecr.io",
+                        public_network_access="Enabled",
+                        admin_user_enabled=True,
+                        monitor_diagnostic_settings=[],
+                        network_rule_set=[],
+                        private_endpoint_connections=[],
+                    )
+                }
+            }
+
+            check = containerregistry_uses_private_link()
+
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Container Registry mock_registry from subscription {AZURE_SUBSCRIPTION_ID} does not use a private link."
+            )
+            assert result[0].subscription == AZURE_SUBSCRIPTION_ID
+            assert result[0].resource_name == "mock_registry"
+            assert (
+                result[0].resource_id
+                == containerregistry_client.registries[AZURE_SUBSCRIPTION_ID][
+                    registry_id
+                ].id
+            )
+            assert result[0].location == "westeurope"
+
+    def test_container_registry_uses_private_link(self):
+        containerregistry_client = mock.MagicMock()
+        containerregistry_client.registries = {}
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=set_mocked_azure_provider(),
+        ), mock.patch(
+            "prowler.providers.azure.services.containerregistry.containerregistry_uses_private_link.containerregistry_uses_private_link.containerregistry_client",
+            new=containerregistry_client,
+        ):
+            from prowler.providers.azure.services.containerregistry.containerregistry_service import (
+                ContainerRegistryInfo,
+            )
+            from prowler.providers.azure.services.containerregistry.containerregistry_uses_private_link.containerregistry_uses_private_link import (
+                containerregistry_uses_private_link,
+            )
+
+            registry_id = str(uuid4())
+
+            containerregistry_client.registries = {
+                AZURE_SUBSCRIPTION_ID: {
+                    registry_id: ContainerRegistryInfo(
+                        id=registry_id,
+                        name="mock_registry",
+                        location="westeurope",
+                        resource_group="mock_resource_group",
+                        sku="Basic",
+                        login_server="mock_login_server.azurecr.io",
+                        public_network_access="Enabled",
+                        admin_user_enabled=False,
+                        monitor_diagnostic_settings=[],
+                        network_rule_set=[],
+                        private_endpoint_connections=[
+                            PrivateEndpointConnection(
+                                id="/subscriptions/AZURE_SUBSCRIPTION_ID/resourceGroups/mock_resource_group/providers/Microsoft.ContainerRegistry/registries/mock_registry/privateEndpointConnections/myConnection",
+                                private_endpoint=PrivateEndpoint(
+                                    id="/subscriptions/AZURE_SUBSCRIPTION_ID/resourceGroups/mock_resource_group/providers/Microsoft.Network/privateEndpoints/myPrivateEndpoint"
+                                ),
+                                private_link_service_connection_state=PrivateLinkServiceConnectionState(
+                                    status="Approved",
+                                    description="Auto-approved connection",
+                                    actions_required="None",
+                                ),
+                                provisioning_state="Succeeded",
+                            )
+                        ],
+                    )
+                }
+            }
+
+            check = containerregistry_uses_private_link()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == f"Container Registry mock_registry from subscription {AZURE_SUBSCRIPTION_ID} uses a private link."
+            )
+            assert result[0].subscription == AZURE_SUBSCRIPTION_ID
+            assert result[0].resource_name == "mock_registry"
+            assert (
+                result[0].resource_id
+                == containerregistry_client.registries[AZURE_SUBSCRIPTION_ID][
+                    registry_id
+                ].id
+            )
+            assert result[0].location == "westeurope"

tests/providers/azure/services/defender/defender_additional_email_configured_with_a_security_contact/defender_additional_email_configured_with_a_security_contact_test.py

@@ -62,7 +62,7 @@ class Test_defender_additional_email_configured_with_a_security_contact:
             assert result[0].status == "FAIL"
             assert (
                 result[0].status_extended
-                == f"There is not another correct email configured for susbscription {AZURE_SUBSCRIPTION_ID}."
+                == f"There is not another correct email configured for subscription {AZURE_SUBSCRIPTION_ID}."
             )
             assert result[0].subscription == AZURE_SUBSCRIPTION_ID
             assert result[0].resource_name == "default"
@@ -102,7 +102,7 @@ class Test_defender_additional_email_configured_with_a_security_contact:
             assert result[0].status == "FAIL"
             assert (
                 result[0].status_extended
-                == f"There is not another correct email configured for susbscription {AZURE_SUBSCRIPTION_ID}."
+                == f"There is not another correct email configured for subscription {AZURE_SUBSCRIPTION_ID}."
             )
             assert result[0].subscription == AZURE_SUBSCRIPTION_ID
             assert result[0].resource_name == "default"
@@ -142,7 +142,7 @@ class Test_defender_additional_email_configured_with_a_security_contact:
             assert result[0].status == "FAIL"
             assert (
                 result[0].status_extended
-                == f"There is not another correct email configured for susbscription {AZURE_SUBSCRIPTION_ID}."
+                == f"There is not another correct email configured for subscription {AZURE_SUBSCRIPTION_ID}."
             )
             assert result[0].subscription == AZURE_SUBSCRIPTION_ID
             assert result[0].resource_name == "default"
@@ -182,7 +182,7 @@ class Test_defender_additional_email_configured_with_a_security_contact:
             assert result[0].status == "PASS"
             assert (
                 result[0].status_extended
-                == f"There is another correct email configured for susbscription {AZURE_SUBSCRIPTION_ID}."
+                == f"There is another correct email configured for subscription {AZURE_SUBSCRIPTION_ID}."
             )
             assert result[0].subscription == AZURE_SUBSCRIPTION_ID
             assert result[0].resource_name == "default"
@@ -222,7 +222,7 @@ class Test_defender_additional_email_configured_with_a_security_contact:
             assert result[0].status == "PASS"
             assert (
                 result[0].status_extended
-                == f"There is another correct email configured for susbscription {AZURE_SUBSCRIPTION_ID}."
+                == f"There is another correct email configured for subscription {AZURE_SUBSCRIPTION_ID}."
             )
             assert result[0].subscription == AZURE_SUBSCRIPTION_ID
             assert result[0].resource_name == "default"
@@ -261,7 +261,7 @@ class Test_defender_additional_email_configured_with_a_security_contact:
             assert result[0].status == "FAIL"
             assert (
                 result[0].status_extended
-                == f"There is not another correct email configured for susbscription {AZURE_SUBSCRIPTION_ID}."
+                == f"There is not another correct email configured for subscription {AZURE_SUBSCRIPTION_ID}."
             )
             assert result[0].subscription == AZURE_SUBSCRIPTION_ID
             assert result[0].resource_name == "default"

tests/providers/azure/services/defender/defender_ensure_iot_hub_defender_is_on/defender_ensure_iot_hub_defender_is_on_test.py

@@ -84,7 +84,7 @@ class Test_defender_ensure_iot_hub_defender_is_on:
             assert result[0].status == "FAIL"
             assert (
                 result[0].status_extended
-                == f"The security solution iot_sec_solution is disabled in susbscription {AZURE_SUBSCRIPTION_ID}"
+                == f"The security solution iot_sec_solution is disabled in subscription {AZURE_SUBSCRIPTION_ID}"
             )
             assert result[0].resource_name == "iot_sec_solution"
             assert result[0].resource_id == resource_id
@@ -117,7 +117,7 @@ class Test_defender_ensure_iot_hub_defender_is_on:
             assert result[0].status == "PASS"
             assert (
                 result[0].status_extended
-                == f"The security solution iot_sec_solution is enabled in susbscription {AZURE_SUBSCRIPTION_ID}."
+                == f"The security solution iot_sec_solution is enabled in subscription {AZURE_SUBSCRIPTION_ID}."
             )
             assert result[0].resource_name == "iot_sec_solution"
             assert result[0].resource_id == resource_id
@@ -155,7 +155,7 @@ class Test_defender_ensure_iot_hub_defender_is_on:
             assert result[0].status == "PASS"
             assert (
                 result[0].status_extended
-                == f"The security solution iot_sec_solution_enabled is enabled in susbscription {AZURE_SUBSCRIPTION_ID}."
+                == f"The security solution iot_sec_solution_enabled is enabled in subscription {AZURE_SUBSCRIPTION_ID}."
             )
             assert result[0].resource_name == "iot_sec_solution_enabled"
             assert result[0].resource_id == resource_id_enabled
@@ -164,7 +164,7 @@ class Test_defender_ensure_iot_hub_defender_is_on:
             assert result[1].status == "FAIL"
             assert (
                 result[1].status_extended
-                == f"The security solution iot_sec_solution_disabled is disabled in susbscription {AZURE_SUBSCRIPTION_ID}"
+                == f"The security solution iot_sec_solution_disabled is disabled in subscription {AZURE_SUBSCRIPTION_ID}"
             )
             assert result[1].resource_name == "iot_sec_solution_disabled"
             assert result[1].resource_id == resource_id_disabled

tests/providers/azure/services/defender/defender_ensure_notify_alerts_severity_is_high/defender_ensure_notify_alerts_severity_is_high_test.py

@@ -62,7 +62,7 @@ class Test_defender_ensure_notify_alerts_severity_is_high:
             assert result[0].status == "FAIL"
             assert (
                 result[0].status_extended
-                == f"Notifiy alerts are not enabled for severity high in susbscription {AZURE_SUBSCRIPTION_ID}."
+                == f"Notifiy alerts are not enabled for severity high in subscription {AZURE_SUBSCRIPTION_ID}."
             )
             assert result[0].subscription == AZURE_SUBSCRIPTION_ID
             assert result[0].resource_name == "default"
@@ -102,7 +102,7 @@ class Test_defender_ensure_notify_alerts_severity_is_high:
             assert result[0].status == "PASS"
             assert (
                 result[0].status_extended
-                == f"Notifiy alerts are enabled for severity high in susbscription {AZURE_SUBSCRIPTION_ID}."
+                == f"Notifiy alerts are enabled for severity high in subscription {AZURE_SUBSCRIPTION_ID}."
             )
             assert result[0].subscription == AZURE_SUBSCRIPTION_ID
             assert result[0].resource_name == "default"
@@ -141,7 +141,7 @@ class Test_defender_ensure_notify_alerts_severity_is_high:
             assert result[0].status == "FAIL"
             assert (
                 result[0].status_extended
-                == f"Notifiy alerts are not enabled for severity high in susbscription {AZURE_SUBSCRIPTION_ID}."
+                == f"Notifiy alerts are not enabled for severity high in subscription {AZURE_SUBSCRIPTION_ID}."
             )
             assert result[0].subscription == AZURE_SUBSCRIPTION_ID
             assert result[0].resource_name == "default"