feat(gcp): add Shodan check for GCP External Addresses

docs/tutorials/configuration_file.md

@@ -151,5 +151,8 @@ azure:
 
 # GCP Configuration
 gcp:
+  # GCP Compute Configuration
+  # gcp.compute_public_address_shodan
+  shodan_api_key: null
 
 ```

docs/tutorials/pentesting.md

@@ -62,12 +62,17 @@ prowler <provider> --categories internet-exposed
 
 ### Shodan
 
-Prowler allows you check if any elastic ip in your AWS Account is exposed in Shodan with `-N`/`--shodan <shodan_api_key>` option:
+Prowler allows you check if any public IPs in your Cloud environments are exposed in Shodan with `-N`/`--shodan <shodan_api_key>` option:
 
+For example, you can check if any of your AWS EC2 instances has an elastic IP exposed in shodan:
 ```console
 prowler aws -N/--shodan <shodan_api_key> -c ec2_elastic_ip_shodan
 ```
 Also, you can check if any of your Azure Subscription has an public IP exposed in shodan:
 ```console
 prowler azure -N/--shodan <shodan_api_key> -c network_public_ip_shodan
 ```
+And finally, you can check if any of your GCP projects has an public IP address exposed in shodan:
+```console
+prowler gcp -N/--shodan <shodan_api_key> -c compute_public_address_shodan
+```

prowler/config/config.yaml

@@ -103,3 +103,6 @@ azure:
 
 # GCP Configuration
 gcp:
+  # GCP Compute Configuration
+  # gcp.compute_public_address_shodan
+  shodan_api_key: null

prowler/providers/common/outputs.py

@@ -105,6 +105,12 @@ def __init__(self, arguments, audit_info, allowlist_file, bulk_checks_metadata):
         # First call Provider_Output_Options init
         super().__init__(arguments, allowlist_file, bulk_checks_metadata)
 
+        # Confire Shodan API
+        if arguments.shodan:
+            audit_info = change_config_var(
+                "shodan_api_key", arguments.shodan, audit_info
+            )
+
         # Check if custom output filename was input, if not, set the default
         if (
             not hasattr(arguments, "output_filename")

prowler/providers/gcp/lib/arguments/arguments.py

@@ -20,3 +20,13 @@ def init_parser(self):
         default=[],
         help="GCP Project IDs to be scanned by Prowler",
     )
+
+    # 3rd Party Integrations
+    gcp_3rd_party_subparser = gcp_parser.add_argument_group("3rd Party Integrations")
+    gcp_3rd_party_subparser.add_argument(
+        "-N",
+        "--shodan",
+        nargs="?",
+        default=None,
+        help="Shodan API key used by check compute_public_address_shodan.",
+    )

prowler/providers/gcp/lib/service/service.py

@@ -31,6 +31,7 @@ def __init__(
         )
         # Only project ids that have their API enabled will be scanned
         self.project_ids = self.__is_api_active__(audit_info.project_ids)
+        self.audit_config = audit_info.audit_config
 
     def __get_client__(self):
         return self.client

prowler/providers/gcp/services/compute/compute_public_address_shodan/compute_public_address_shodan.metadata.json

@@ -0,0 +1,34 @@
+{
+  "Provider": "compute",
+  "CheckID": "compute_public_address_shodan",
+  "CheckTitle": "Check if any of the Public Addresses are in Shodan (requires Shodan API KEY).",
+  "CheckType": [
+    "Infrastructure Security"
+  ],
+  "ServiceName": "compute",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "",
+  "Severity": "high",
+  "ResourceType": "AwsEc2Eip",
+  "Description": "Check if any of the Public Addresses are in Shodan (requires Shodan API KEY).",
+  "Risk": "Sites like Shodan index exposed systems and further expose them to wider audiences as a quick way to find exploitable systems.",
+  "RelatedUrl": "",
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Check Identified IPs; consider changing them to private ones and delete them from Shodan.",
+      "Url": "https://www.shodan.io/"
+    }
+  },
+  "Categories": [
+    "internet-exposed"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

prowler/providers/gcp/services/compute/compute_public_address_shodan/compute_public_address_shodan.py

@@ -0,0 +1,40 @@
+import shodan
+
+from prowler.lib.check.models import Check, Check_Report_GCP
+from prowler.lib.logger import logger
+from prowler.providers.gcp.services.compute.compute_client import compute_client
+
+
+class compute_public_address_shodan(Check):
+    def execute(self):
+        findings = []
+        shodan_api_key = compute_client.audit_config.get("shodan_api_key")
+        if shodan_api_key:
+            api = shodan.Shodan(shodan_api_key)
+            for address in compute_client.addresses:
+                if address.type == "EXTERNAL":
+                    report = Check_Report_GCP(self.metadata())
+                    report.project_id = address.project_id
+                    report.resource_id = address.id
+                    report.location = address.region
+                    try:
+                        shodan_info = api.host(address.ip)
+                        report.status = "FAIL"
+                        report.status_extended = f"Public Address {address.ip} listed in Shodan with open ports {str(shodan_info['ports'])} and ISP {shodan_info['isp']} in {shodan_info['country_name']}. More info at https://www.shodan.io/host/{address.ip}."
+                        findings.append(report)
+                    except shodan.APIError as error:
+                        if "No information available for that IP" in error.value:
+                            report.status = "PASS"
+                            report.status_extended = (
+                                f"Public Address {address.ip} is not listed in Shodan."
+                            )
+                            findings.append(report)
+                            continue
+                        else:
+                            logger.error(f"Unknown Shodan API Error: {error.value}")
+
+        else:
+            logger.error(
+                "No Shodan API Key -- Please input a Shodan API Key with -N/--shodan or in config.yaml"
+            )
+        return findings

prowler/providers/gcp/services/compute/compute_service.py

@@ -13,6 +13,7 @@ def __init__(self, audit_info):
         self.instances = []
         self.networks = []
         self.subnets = []
+        self.addresses = []
         self.firewalls = []
         self.projects = []
         self.load_balancers = []
@@ -25,6 +26,7 @@ def __init__(self, audit_info):
         self.__get_networks__()
         self.__threading_call__(self.__get_subnetworks__, self.regions)
         self.__get_firewalls__()
+        self.__threading_call__(self.__get_addresses__, self.regions)
 
     def __get_regions__(self):
         for project_id in self.project_ids:
@@ -197,6 +199,36 @@ def __get_subnetworks__(self, region):
                     f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
                 )
 
+    def __get_addresses__(self, region):
+        for project_id in self.project_ids:
+            try:
+                request = self.client.addresses().list(
+                    project=project_id, region=region
+                )
+                while request is not None:
+                    response = request.execute(
+                        http=self.__get_AuthorizedHttp_client__()
+                    )
+                    for address in response.get("items", []):
+                        self.addresses.append(
+                            Address(
+                                name=address["name"],
+                                id=address["id"],
+                                project_id=project_id,
+                                type=address.get("addressType", "EXTERNAL"),
+                                ip=address["address"],
+                                region=region,
+                            )
+                        )
+
+                    request = self.client.subnetworks().list_next(
+                        previous_request=request, previous_response=response
+                    )
+            except Exception as error:
+                logger.error(
+                    f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+
     def __get_firewalls__(self):
         for project_id in self.project_ids:
             try:
@@ -297,6 +329,15 @@ class Subnet(BaseModel):
     region: str
 
 
+class Address(BaseModel):
+    name: str
+    id: str
+    ip: str
+    type: str
+    project_id: str
+    region: str
+
+
 class Firewall(BaseModel):
     name: str
     id: str

tests/providers/gcp/services/compute/compute_public_address_shodan/compute_public_address_shodan_test.py

@@ -0,0 +1,79 @@
+from unittest import mock
+
+from prowler.providers.gcp.services.compute.compute_service import Address
+from tests.providers.gcp.lib.audit_info_utils import GCP_PROJECT_ID
+
+
+class Test_compute_public_address_shodan:
+    def test_no_public_ip_addresses(self):
+        compute_client = mock.MagicMock
+        compute_client.addresses = {}
+        compute_client.audit_info = mock.MagicMock
+
+        with mock.patch(
+            "prowler.providers.gcp.services.compute.compute_service.Network",
+            new=compute_client,
+        ) as service_client, mock.patch(
+            "prowler.providers.gcp.services.compute.compute_client.compute_client",
+            new=service_client,
+        ):
+            from prowler.providers.gcp.services.compute.compute_public_address_shodan.compute_public_address_shodan import (
+                compute_public_address_shodan,
+            )
+
+            compute_client.audit_config = {"shodan_api_key": "api_key"}
+
+            check = compute_public_address_shodan()
+            result = check.execute()
+            assert len(result) == 0
+
+    def test_compute_ip_in_shodan(self):
+        compute_client = mock.MagicMock
+        public_ip_id = "id"
+        public_ip_name = "name"
+        ip_address = "ip_address"
+        shodan_info = {
+            "ports": [80, 443],
+            "isp": "Microsoft Corporation",
+            "country_name": "country_name",
+        }
+        compute_client.audit_info = mock.MagicMock
+
+        compute_client.addresses = [
+            Address(
+                id=public_ip_id,
+                name=public_ip_name,
+                type="EXTERNAL",
+                ip=ip_address,
+                region="region",
+                network="network",
+                project_id=GCP_PROJECT_ID,
+            )
+        ]
+
+        with mock.patch(
+            "prowler.providers.gcp.services.compute.compute_service.Network",
+            new=compute_client,
+        ) as service_client, mock.patch(
+            "prowler.providers.gcp.services.compute.compute_client.compute_client",
+            new=service_client,
+        ), mock.patch(
+            "prowler.providers.gcp.services.compute.compute_public_address_shodan.compute_public_address_shodan.shodan.Shodan.host",
+            return_value=shodan_info,
+        ):
+            from prowler.providers.gcp.services.compute.compute_public_address_shodan.compute_public_address_shodan import (
+                compute_public_address_shodan,
+            )
+
+            compute_client.audit_config = {"shodan_api_key": "api_key"}
+            check = compute_public_address_shodan()
+            result = check.execute()
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == f"Public Address {ip_address} listed in Shodan with open ports {str(shodan_info['ports'])} and ISP {shodan_info['isp']} in {shodan_info['country_name']}. More info at https://www.shodan.io/host/{ip_address}."
+            )
+            assert result[0].project_id == GCP_PROJECT_ID
+            assert result[0].location == "region"
+            assert result[0].resource_id == public_ip_id