feat(network): extract Network resource metadata automated (#6555)

Co-authored-by: Sergio Garcia <hello@mistercloudsec.com>
This commit is contained in:
Rubén De la Torre Vico
2025-01-16 18:41:02 +01:00
committed by Pepe Fagoaga
parent de76a168c0
commit 18a4881a51
9 changed files with 27 additions and 30 deletions
@@ -18,7 +18,9 @@ class network_bastion_host_exists(Check):
status = "PASS"
status_extended = f"Bastion Host from subscription {subscription} available are: {bastion_names}"
report = Check_Report_Azure(self.metadata())
report = Check_Report_Azure(
metadata=self.metadata(), resource_metadata=bastion_hosts
)
report.subscription = subscription
report.resource_name = "Bastion Host"
report.resource_id = "Bastion Host"
@@ -7,12 +7,11 @@ class network_flow_log_captured_sent(Check):
findings = []
for subscription, network_watchers in network_client.network_watchers.items():
for network_watcher in network_watchers:
report = Check_Report_Azure(self.metadata())
report = Check_Report_Azure(
metadata=self.metadata(), resource_metadata=network_watcher
)
report.subscription = subscription
report.resource_name = network_watcher.name
report.resource_id = network_watcher.id
report.status = "FAIL"
report.location = network_watcher.location
report.status_extended = f"Network Watcher {network_watcher.name} from subscription {subscription} has no flow logs"
if network_watcher.flow_logs:
report.status = "FAIL"
@@ -7,11 +7,10 @@ class network_flow_log_more_than_90_days(Check):
findings = []
for subscription, network_watchers in network_client.network_watchers.items():
for network_watcher in network_watchers:
report = Check_Report_Azure(self.metadata())
report = Check_Report_Azure(
metadata=self.metadata(), resource_metadata=network_watcher
)
report.subscription = subscription
report.resource_name = network_watcher.name
report.resource_id = network_watcher.id
report.location = network_watcher.location
if network_watcher.flow_logs:
report.status = "PASS"
report.status_extended = f"Network Watcher {network_watcher.name} from subscription {subscription} has flow logs enabled for more than 90 days"
@@ -7,12 +7,11 @@ class network_http_internet_access_restricted(Check):
findings = []
for subscription, security_groups in network_client.security_groups.items():
for security_group in security_groups:
report = Check_Report_Azure(self.metadata())
report = Check_Report_Azure(
metadata=self.metadata(), resource_metadata=security_group
)
report.subscription = subscription
report.resource_name = security_group.name
report.resource_id = security_group.id
report.status = "PASS"
report.location = security_group.location
report.status_extended = f"Security Group {security_group.name} from subscription {subscription} has HTTP internet access restricted."
rule_fail_condition = any(
(
@@ -13,11 +13,10 @@ class network_public_ip_shodan(Check):
api = shodan.Shodan(shodan_api_key)
for subscription, public_ips in network_client.public_ip_addresses.items():
for ip in public_ips:
report = Check_Report_Azure(self.metadata())
report = Check_Report_Azure(
metadata=self.metadata(), resource_metadata=ip
)
report.subscription = subscription
report.resource_name = ip.name
report.resource_id = ip.id
report.location = ip.location
try:
shodan_info = api.host(ip.ip_address)
report.status = "FAIL"
@@ -7,12 +7,11 @@ class network_rdp_internet_access_restricted(Check):
findings = []
for subscription, security_groups in network_client.security_groups.items():
for security_group in security_groups:
report = Check_Report_Azure(self.metadata())
report = Check_Report_Azure(
metadata=self.metadata(), resource_metadata=security_group
)
report.subscription = subscription
report.resource_name = security_group.name
report.resource_id = security_group.id
report.status = "PASS"
report.location = security_group.location
report.status_extended = f"Security Group {security_group.name} from subscription {subscription} has RDP internet access restricted."
rule_fail_condition = any(
(
@@ -7,12 +7,11 @@ class network_ssh_internet_access_restricted(Check):
findings = []
for subscription, security_groups in network_client.security_groups.items():
for security_group in security_groups:
report = Check_Report_Azure(self.metadata())
report = Check_Report_Azure(
metadata=self.metadata(), resource_metadata=security_group
)
report.subscription = subscription
report.resource_name = security_group.name
report.resource_id = security_group.id
report.status = "PASS"
report.location = security_group.location
report.status_extended = f"Security Group {security_group.name} from subscription {subscription} has SSH internet access restricted."
rule_fail_condition = any(
(
@@ -7,12 +7,11 @@ class network_udp_internet_access_restricted(Check):
findings = []
for subscription, security_groups in network_client.security_groups.items():
for security_group in security_groups:
report = Check_Report_Azure(self.metadata())
report = Check_Report_Azure(
metadata=self.metadata(), resource_metadata=security_group
)
report.subscription = subscription
report.resource_name = security_group.name
report.resource_id = security_group.id
report.status = "PASS"
report.location = security_group.location
report.status_extended = f"Security Group {security_group.name} from subscription {subscription} has UDP internet access restricted."
rule_fail_condition = any(
rule.protocol in ["UDP", "Udp"]
@@ -6,7 +6,9 @@ class network_watcher_enabled(Check):
def execute(self) -> list[Check_Report_Azure]:
findings = []
for subscription, network_watchers in network_client.network_watchers.items():
report = Check_Report_Azure(self.metadata())
report = Check_Report_Azure(
metadata=self.metadata(), resource_metadata=network_watchers
)
report.subscription = subscription
report.resource_name = "Network Watcher"
report.location = "global"