mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-19 10:31:51 +00:00
feat(gcp): add Shodan check for GCP External Addresses (#3486)
Co-authored-by: Pepe Fagoaga <pepe@prowler.com>
This commit is contained in:
@@ -151,5 +151,8 @@ azure:
|
||||
|
||||
# GCP Configuration
|
||||
gcp:
|
||||
# GCP Compute Configuration
|
||||
# gcp.compute_public_address_shodan
|
||||
shodan_api_key: null
|
||||
|
||||
```
|
||||
|
||||
@@ -62,8 +62,9 @@ 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
|
||||
```
|
||||
@@ -71,3 +72,7 @@ Also, you can check if any of your Azure Subscription has an public IP exposed i
|
||||
```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
|
||||
```
|
||||
|
||||
@@ -103,3 +103,6 @@ azure:
|
||||
|
||||
# GCP Configuration
|
||||
gcp:
|
||||
# GCP Compute Configuration
|
||||
# gcp.compute_public_address_shodan
|
||||
shodan_api_key: null
|
||||
|
||||
@@ -78,7 +78,7 @@ class Azure_Output_Options(Provider_Output_Options):
|
||||
# First call Provider_Output_Options init
|
||||
super().__init__(arguments, allowlist_file, bulk_checks_metadata)
|
||||
|
||||
# Confire Shodan API
|
||||
# Configure Shodan API
|
||||
if arguments.shodan:
|
||||
audit_info = change_config_var(
|
||||
"shodan_api_key", arguments.shodan, audit_info
|
||||
@@ -105,6 +105,12 @@ class Gcp_Output_Options(Provider_Output_Options):
|
||||
# First call Provider_Output_Options init
|
||||
super().__init__(arguments, allowlist_file, bulk_checks_metadata)
|
||||
|
||||
# Configure 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")
|
||||
@@ -122,7 +128,7 @@ class Aws_Output_Options(Provider_Output_Options):
|
||||
# First call Provider_Output_Options init
|
||||
super().__init__(arguments, allowlist_file, bulk_checks_metadata)
|
||||
|
||||
# Confire Shodan API
|
||||
# Configure Shodan API
|
||||
if arguments.shodan:
|
||||
audit_info = change_config_var(
|
||||
"shodan_api_key", arguments.shodan, audit_info
|
||||
|
||||
@@ -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.",
|
||||
)
|
||||
|
||||
@@ -31,6 +31,7 @@ class GCPService:
|
||||
)
|
||||
# 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
|
||||
|
||||
+34
@@ -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": "GCPComputeAddress",
|
||||
"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": ""
|
||||
}
|
||||
+40
@@ -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
|
||||
@@ -13,6 +13,7 @@ class Compute(GCPService):
|
||||
self.instances = []
|
||||
self.networks = []
|
||||
self.subnets = []
|
||||
self.addresses = []
|
||||
self.firewalls = []
|
||||
self.projects = []
|
||||
self.load_balancers = []
|
||||
@@ -25,6 +26,7 @@ class Compute(GCPService):
|
||||
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 @@ class Compute(GCPService):
|
||||
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
|
||||
|
||||
@@ -130,6 +130,7 @@ class Test_Common_Output_Options:
|
||||
arguments.verbose = True
|
||||
arguments.output_filename = "output_test_filename"
|
||||
arguments.only_logs = False
|
||||
arguments.shodan = "test-api-key"
|
||||
arguments.unix_timestamp = False
|
||||
|
||||
audit_info = self.set_mocked_gcp_audit_info()
|
||||
|
||||
+79
@@ -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
|
||||
Reference in New Issue
Block a user