mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 04:51:51 +00:00
feat(cloudflare): extend dns and zone services check coverage (#9426)
Co-authored-by: Andoni Alonso <14891798+andoniaf@users.noreply.github.com>
This commit is contained in:
Hugo Pereira Brito
committed by
GitHub
GitHub
parent
9e7ecb39fa
commit
6bb8dc6168
@@ -51,6 +51,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
|
||||
- CIS 6.0 for M365 provider [(#9779)](https://github.com/prowler-cloud/prowler/pull/9779)
|
||||
- CIS 5.0 compliance framework for the Azure provider [(#9777)](https://github.com/prowler-cloud/prowler/pull/9777)
|
||||
- `Cloudflare` Bot protection, WAF, Privacy, Anti-Scraping and Zone configuration checks [(#9425)](https://github.com/prowler-cloud/prowler/pull/9425)
|
||||
- `Cloudflare` `waf` and `dns record` checks [(#9426)](https://github.com/prowler-cloud/prowler/pull/9426)
|
||||
|
||||
### Changed
|
||||
|
||||
|
||||
@@ -779,7 +779,10 @@ class CheckReportCloudflare(Check_Report):
|
||||
|
||||
@property
|
||||
def zone_name(self) -> str:
|
||||
"""Zone name."""
|
||||
"""Zone name - for DNS records use zone_name attribute, for zones use name."""
|
||||
zone_name = getattr(self._zone, "zone_name", None)
|
||||
if zone_name:
|
||||
return zone_name
|
||||
return getattr(self._zone, "name", "")
|
||||
|
||||
@property
|
||||
@@ -792,7 +795,10 @@ class CheckReportCloudflare(Check_Report):
|
||||
|
||||
@property
|
||||
def region(self) -> str:
|
||||
"""Cloudflare is a global service."""
|
||||
"""Return zone_name as region for zone-scoped resources, otherwise global."""
|
||||
zone_name = getattr(self._zone, "zone_name", None)
|
||||
if zone_name:
|
||||
return zone_name
|
||||
return "global"
|
||||
|
||||
|
||||
|
||||
+36
@@ -0,0 +1,36 @@
|
||||
{
|
||||
"Provider": "cloudflare",
|
||||
"CheckID": "dns_record_cname_target_valid",
|
||||
"CheckTitle": "DNS records pointing to hostnames have valid targets without takeover risk",
|
||||
"CheckType": [],
|
||||
"ServiceName": "dns",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "high",
|
||||
"ResourceType": "DNSRecord",
|
||||
"ResourceGroup": "network",
|
||||
"Description": "**Cloudflare DNS records** (CNAME, MX, NS, SRV) that point to hostnames are assessed for **dangling record** vulnerabilities by checking if the target domain resolves to a valid address, preventing **subdomain takeover**, **mail interception**, and **service hijacking** attacks.",
|
||||
"Risk": "Dangling **DNS records** pointing to non-existent targets create multiple vulnerabilities.\n- **Confidentiality**: dangling CNAME/NS allows subdomain takeover; dangling MX allows mail interception\n- **Integrity**: attackers can impersonate your organization, intercept emails, or hijack services\n- **Availability**: legitimate services may be disrupted or redirected to attacker-controlled infrastructure",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://developers.cloudflare.com/dns/manage-dns-records/how-to/create-dns-records/"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Log in to the Cloudflare dashboard and select your account and domain\n2. Go to DNS > Records\n3. Identify CNAME, MX, NS, or SRV records with dangling targets\n4. Either update the record to point to a valid target or delete the record\n5. If the target service was decommissioned, remove the DNS record",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Remove or update **dangling DNS records** to prevent takeover and interception attacks.\n- Regularly audit DNS records when decommissioning services\n- Remove CNAME, MX, NS, and SRV records pointing to deprovisioned resources\n- Monitor for unauthorized changes to DNS records\n- Consider using DNS monitoring tools to detect dangling records",
|
||||
"Url": "https://hub.prowler.com/checks/cloudflare/dns_record_cname_target_valid"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"internet-exposed"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": "Subdomain takeover occurs when a CNAME or NS record points to a service that has been deprovisioned, allowing attackers to claim that service and control the subdomain. Similarly, dangling MX records can allow mail interception, and dangling SRV records can expose service discovery vulnerabilities."
|
||||
}
|
||||
+109
@@ -0,0 +1,109 @@
|
||||
import socket
|
||||
|
||||
from prowler.lib.check.models import Check, CheckReportCloudflare
|
||||
from prowler.providers.cloudflare.services.dns.dns_client import dns_client
|
||||
|
||||
# Record types that point to hostnames and can be dangling:
|
||||
# - CNAME: Alias to another hostname
|
||||
# - MX: Mail server hostname (dangling = potential mail interception)
|
||||
# - NS: Nameserver delegation (dangling = subdomain takeover)
|
||||
# - SRV: Service location hostname
|
||||
DANGLING_RISK_TYPES = {"CNAME", "MX", "NS", "SRV"}
|
||||
|
||||
# Risk descriptions for each record type
|
||||
RISK_DESCRIPTIONS = {
|
||||
"CNAME": "subdomain takeover risk",
|
||||
"MX": "potential mail interception risk",
|
||||
"NS": "subdomain delegation takeover risk",
|
||||
"SRV": "service discovery vulnerability",
|
||||
}
|
||||
|
||||
|
||||
class dns_record_cname_target_valid(Check):
|
||||
"""Ensure that DNS records pointing to hostnames have valid, resolvable targets.
|
||||
|
||||
Dangling DNS records that point to non-existent or unresolvable targets pose
|
||||
significant security risks. CNAME and NS records can lead to subdomain takeover,
|
||||
MX records can allow mail interception, and SRV records can expose service
|
||||
vulnerabilities. Attackers can claim orphaned target resources and serve
|
||||
malicious content, intercept email, or hijack services under your domain.
|
||||
"""
|
||||
|
||||
def execute(self) -> list[CheckReportCloudflare]:
|
||||
"""Execute the dangling DNS record validation check.
|
||||
|
||||
Iterates through all DNS records that point to hostnames (CNAME, MX, NS, SRV)
|
||||
and attempts to resolve their targets using DNS lookup. Records pointing to
|
||||
unresolvable targets are flagged as potential security risks.
|
||||
|
||||
Returns:
|
||||
A list of CheckReportCloudflare objects with PASS status if the
|
||||
target resolves successfully, or FAIL status if the target
|
||||
cannot be resolved (dangling record).
|
||||
"""
|
||||
findings = []
|
||||
|
||||
for record in dns_client.records:
|
||||
# Check record types that point to hostnames
|
||||
if record.type not in DANGLING_RISK_TYPES:
|
||||
continue
|
||||
|
||||
report = CheckReportCloudflare(
|
||||
metadata=self.metadata(),
|
||||
resource=record,
|
||||
)
|
||||
|
||||
target = self._extract_target(record.type, record.content)
|
||||
is_valid = self._check_target_resolves(target)
|
||||
risk_desc = RISK_DESCRIPTIONS.get(record.type, "security risk")
|
||||
|
||||
if is_valid:
|
||||
report.status = "PASS"
|
||||
report.status_extended = f"{record.type} record {record.name} points to valid target {target}."
|
||||
else:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
f"{record.type} record {record.name} points to potentially dangling "
|
||||
f"target {target} - {risk_desc}."
|
||||
)
|
||||
findings.append(report)
|
||||
|
||||
return findings
|
||||
|
||||
def _extract_target(self, record_type: str, content: str) -> str:
|
||||
"""Extract the target hostname from record content.
|
||||
|
||||
Different record types have different content formats:
|
||||
- CNAME: hostname
|
||||
- MX: priority hostname (e.g., "10 mail.example.com")
|
||||
- NS: hostname
|
||||
- SRV: Cloudflare returns "weight port hostname" (e.g., "5 80 sip.example.com")
|
||||
"""
|
||||
if record_type == "MX":
|
||||
# MX format: "priority hostname"
|
||||
parts = content.split(None, 1)
|
||||
return parts[1] if len(parts) > 1 else content
|
||||
elif record_type == "SRV":
|
||||
# SRV format from Cloudflare: "weight port hostname"
|
||||
parts = content.split()
|
||||
# Target is the last part (hostname)
|
||||
return parts[-1] if parts else content
|
||||
else:
|
||||
# CNAME and NS are just hostnames
|
||||
return content
|
||||
|
||||
def _check_target_resolves(self, target: str) -> bool:
|
||||
"""Check if target hostname resolves to a valid address."""
|
||||
# Remove trailing dot if present
|
||||
target = target.rstrip(".")
|
||||
|
||||
try:
|
||||
# Attempt DNS resolution
|
||||
socket.getaddrinfo(target, None, socket.AF_UNSPEC)
|
||||
return True
|
||||
except socket.gaierror:
|
||||
# DNS resolution failed - potential dangling record
|
||||
return False
|
||||
except Exception:
|
||||
# On any other error, assume valid to avoid false positives
|
||||
return True
|
||||
+36
@@ -0,0 +1,36 @@
|
||||
{
|
||||
"Provider": "cloudflare",
|
||||
"CheckID": "dns_record_no_internal_ip",
|
||||
"CheckTitle": "DNS records do not expose internal IP addresses",
|
||||
"CheckType": [],
|
||||
"ServiceName": "dns",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "high",
|
||||
"ResourceType": "DNSRecord",
|
||||
"ResourceGroup": "network",
|
||||
"Description": "**Cloudflare DNS records** are assessed for **internal IP exposure** by checking if A or AAAA records point to private, loopback, or reserved IP addresses which could **leak internal network structure**.",
|
||||
"Risk": "DNS records exposing **internal IP addresses** leak sensitive network information.\n- **Confidentiality**: reveals internal network topology and addressing schemes to attackers\n- **Integrity**: provides reconnaissance data for targeted attacks on internal infrastructure\n- **Availability**: internal IPs in public DNS may indicate misconfiguration affecting service routing",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://developers.cloudflare.com/dns/manage-dns-records/how-to/create-dns-records/"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Log in to the Cloudflare dashboard and select your account and domain\n2. Go to DNS > Records\n3. Identify A/AAAA records pointing to internal IP addresses\n4. Update records to point to public IP addresses or remove if not needed\n5. Use split-horizon DNS if internal resolution is required",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Remove **internal IP addresses** from public DNS records.\n- Use split-horizon DNS for internal service resolution\n- Ensure DNS records only contain publicly routable IP addresses\n- Review DNS records after network changes or migrations\n- Consider using Cloudflare Access for secure internal service access",
|
||||
"Url": "https://hub.prowler.com/checks/cloudflare/dns_record_no_internal_ip"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"internet-exposed"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": "Internal IP ranges include: 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16 (IPv4), fc00::/7 (IPv6 ULA), and loopback addresses. These should not appear in public DNS records."
|
||||
}
|
||||
+73
@@ -0,0 +1,73 @@
|
||||
import ipaddress
|
||||
|
||||
from prowler.lib.check.models import Check, CheckReportCloudflare
|
||||
from prowler.providers.cloudflare.services.dns.dns_client import dns_client
|
||||
|
||||
|
||||
class dns_record_no_internal_ip(Check):
|
||||
"""Ensure that DNS records do not expose internal or private IP addresses.
|
||||
|
||||
Public DNS records should only contain publicly routable IP addresses.
|
||||
Exposing internal, private, loopback, or link-local addresses in DNS records
|
||||
can leak information about internal network infrastructure, potentially
|
||||
aiding attackers in reconnaissance and targeted attacks against internal
|
||||
systems.
|
||||
"""
|
||||
|
||||
def execute(self) -> list[CheckReportCloudflare]:
|
||||
"""Execute the internal IP address exposure check.
|
||||
|
||||
Iterates through all A and AAAA DNS records and checks if they contain
|
||||
private, loopback, link-local, or reserved IP addresses that should not
|
||||
be exposed publicly.
|
||||
|
||||
Returns:
|
||||
A list of CheckReportCloudflare objects with PASS status if the
|
||||
record points to a public IP address, or FAIL status if it exposes
|
||||
an internal IP address.
|
||||
"""
|
||||
findings = []
|
||||
|
||||
for record in dns_client.records:
|
||||
# Only check A and AAAA records
|
||||
if record.type not in ("A", "AAAA"):
|
||||
continue
|
||||
|
||||
report = CheckReportCloudflare(
|
||||
metadata=self.metadata(),
|
||||
resource=record,
|
||||
)
|
||||
|
||||
is_internal = self._is_internal_ip(record.content)
|
||||
|
||||
if not is_internal:
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
f"DNS record {record.name} ({record.type}) points to "
|
||||
f"public IP address {record.content}."
|
||||
)
|
||||
else:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
f"DNS record {record.name} ({record.type}) exposes "
|
||||
f"internal IP address {record.content} - information disclosure risk."
|
||||
)
|
||||
findings.append(report)
|
||||
|
||||
return findings
|
||||
|
||||
def _is_internal_ip(self, ip_str: str) -> bool:
|
||||
"""Check if IP address is internal/private."""
|
||||
try:
|
||||
ip = ipaddress.ip_address(ip_str)
|
||||
# Check for private, loopback, link-local, or reserved addresses
|
||||
return (
|
||||
ip.is_private
|
||||
or ip.is_loopback
|
||||
or ip.is_link_local
|
||||
or ip.is_reserved
|
||||
or ip.is_unspecified
|
||||
)
|
||||
except ValueError:
|
||||
# Invalid IP format, assume not internal
|
||||
return False
|
||||
+36
@@ -0,0 +1,36 @@
|
||||
{
|
||||
"Provider": "cloudflare",
|
||||
"CheckID": "dns_record_no_wildcard",
|
||||
"CheckTitle": "DNS records do not use wildcard entries",
|
||||
"CheckType": [],
|
||||
"ServiceName": "dns",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "DNSRecord",
|
||||
"ResourceGroup": "network",
|
||||
"Description": "**Cloudflare DNS records** are assessed for **wildcard usage** by checking if A, AAAA, CNAME, MX, or SRV records use wildcard entries (*.example.com) which can **increase attack surface**, expose unintended services, or allow mail interception.",
|
||||
"Risk": "**Wildcard DNS records** can expose unintended services and increase attack surface.\n- **Confidentiality**: any subdomain resolves, potentially exposing internal naming conventions; wildcard MX allows mail interception\n- **Integrity**: attackers can access unintended services via arbitrary subdomains\n- **Availability**: wildcard records may route traffic or services not designed for public access",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://developers.cloudflare.com/dns/manage-dns-records/how-to/create-dns-records/"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Log in to the Cloudflare dashboard and select your account and domain\n2. Go to DNS > Records\n3. Identify wildcard DNS records (starting with *.)\n4. Evaluate if the wildcard is necessary for your use case\n5. Replace wildcard records with specific subdomain records where possible",
|
||||
"Terraform": ""
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Avoid using **wildcard DNS records** unless absolutely necessary.\n- Use specific subdomain records instead of wildcards\n- If wildcards are required, ensure the target service handles unknown subdomains securely\n- Document the business justification for any wildcard records\n- Combine with proper web server configuration to reject unknown hosts",
|
||||
"Url": "https://hub.prowler.com/checks/cloudflare/dns_record_no_wildcard"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"internet-exposed"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": "Wildcard DNS records (*.example.com) cause any subdomain query to resolve. While useful for some applications, they can expose services unintentionally and make subdomain enumeration easier for attackers. Wildcard MX records can accept mail for any subdomain, and wildcard SRV records can expose services on arbitrary subdomains."
|
||||
}
|
||||
+60
@@ -0,0 +1,60 @@
|
||||
from prowler.lib.check.models import Check, CheckReportCloudflare
|
||||
from prowler.providers.cloudflare.services.dns.dns_client import dns_client
|
||||
|
||||
# Record types where wildcards pose security risks:
|
||||
# - A, AAAA: Wildcard resolves any subdomain to an IP, exposing services
|
||||
# - CNAME: Wildcard aliases any subdomain, potential for subdomain takeover
|
||||
# - MX: Wildcard accepts mail for any subdomain, potential mail interception
|
||||
# - SRV: Wildcard exposes services on any subdomain
|
||||
WILDCARD_RISK_TYPES = {"A", "AAAA", "CNAME", "MX", "SRV"}
|
||||
|
||||
|
||||
class dns_record_no_wildcard(Check):
|
||||
"""Ensure that wildcard DNS records are not configured for the zone.
|
||||
|
||||
Wildcard DNS records (*.domain.com) match any subdomain that doesn't have
|
||||
an explicit record, which can unintentionally expose services or create
|
||||
security risks. Attackers may discover hidden services, and wildcard
|
||||
certificates combined with wildcard DNS can increase the attack surface
|
||||
for subdomain takeover vulnerabilities. Wildcard MX records can allow
|
||||
mail interception for arbitrary subdomains.
|
||||
"""
|
||||
|
||||
def execute(self) -> list[CheckReportCloudflare]:
|
||||
"""Execute the wildcard DNS record check.
|
||||
|
||||
Iterates through all security-relevant DNS records (A, AAAA, CNAME, MX, SRV)
|
||||
and identifies those configured as wildcard records (starting with *.).
|
||||
Wildcard records may expose unintended services or create security risks.
|
||||
|
||||
Returns:
|
||||
A list of CheckReportCloudflare objects with PASS status if the
|
||||
record is not a wildcard, or FAIL status if it is a wildcard record.
|
||||
"""
|
||||
findings = []
|
||||
|
||||
for record in dns_client.records:
|
||||
# Check record types where wildcards pose security risks
|
||||
if record.type not in WILDCARD_RISK_TYPES:
|
||||
continue
|
||||
|
||||
report = CheckReportCloudflare(
|
||||
metadata=self.metadata(),
|
||||
resource=record,
|
||||
)
|
||||
|
||||
# Check if record name starts with wildcard
|
||||
is_wildcard = record.name.startswith("*.")
|
||||
|
||||
if not is_wildcard:
|
||||
report.status = "PASS"
|
||||
report.status_extended = f"DNS record {record.name} ({record.type}) is not a wildcard record."
|
||||
else:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
f"DNS record {record.name} ({record.type}) is a wildcard record - "
|
||||
f"may expose unintended services."
|
||||
)
|
||||
findings.append(report)
|
||||
|
||||
return findings
|
||||
+36
@@ -0,0 +1,36 @@
|
||||
{
|
||||
"Provider": "cloudflare",
|
||||
"CheckID": "dns_record_proxied",
|
||||
"CheckTitle": "Cloudflare proxy is enabled for applicable DNS records",
|
||||
"CheckType": [],
|
||||
"ServiceName": "dns",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "DNSRecord",
|
||||
"ResourceGroup": "network",
|
||||
"Description": "**Cloudflare DNS records** are assessed for **proxy configuration** by checking if A, AAAA, and CNAME records are proxied through Cloudflare to benefit from **DDoS protection**, **WAF**, and **caching** capabilities.",
|
||||
"Risk": "Unproxied **DNS records** expose origin server IP addresses directly to the internet.\n- **Confidentiality**: origin IP exposure enables targeted reconnaissance and attacks\n- **Integrity**: direct access to origin bypasses WAF and security controls\n- **Availability**: origin is exposed to DDoS attacks without Cloudflare protection",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://developers.cloudflare.com/dns/manage-dns-records/reference/proxied-dns-records/"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Log in to the Cloudflare dashboard and select your account and domain\n2. Go to DNS > Records\n3. For each A, AAAA, or CNAME record that should be protected\n4. Click Edit and toggle Proxy status to Proxied (orange cloud)\n5. Save the changes and verify traffic flows through Cloudflare",
|
||||
"Terraform": "```hcl\n# Enable Cloudflare proxy for DNS records\nresource \"cloudflare_record\" \"proxied_record\" {\n zone_id = \"<ZONE_ID>\"\n name = \"www\"\n content = \"192.0.2.1\"\n type = \"A\"\n proxied = true # Critical: enables DDoS protection, WAF, and caching\n}\n```"
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Enable the **Cloudflare proxy** (orange cloud) for DNS records that should be protected.\n- Proxied records benefit from DDoS protection, WAF, and caching\n- Origin server IP addresses are hidden from public DNS queries\n- Apply defense in depth by combining proxy protection with origin hardening\n- Some record types (MX, TXT) cannot be proxied by design",
|
||||
"Url": "https://hub.prowler.com/checks/cloudflare/dns_record_proxied"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"internet-exposed"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": "Only A, AAAA, and CNAME records can be proxied. MX, TXT, and other record types are always DNS-only. Some services may require DNS-only mode for specific use cases."
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
from prowler.lib.check.models import Check, CheckReportCloudflare
|
||||
from prowler.providers.cloudflare.services.dns.dns_client import dns_client
|
||||
|
||||
PROXYABLE_TYPES = {"A", "AAAA", "CNAME"}
|
||||
|
||||
|
||||
class dns_record_proxied(Check):
|
||||
"""Ensure that DNS records are proxied through Cloudflare.
|
||||
|
||||
Proxying DNS records through Cloudflare hides the origin server's IP address
|
||||
and provides DDoS protection, WAF capabilities, and performance optimizations.
|
||||
Non-proxied (DNS-only) records expose the origin IP directly, bypassing
|
||||
Cloudflare's security features and making the origin vulnerable to direct
|
||||
attacks.
|
||||
"""
|
||||
|
||||
def execute(self) -> list[CheckReportCloudflare]:
|
||||
"""Execute the DNS record proxy status check.
|
||||
|
||||
Iterates through all proxyable DNS records (A, AAAA, CNAME) and verifies
|
||||
that they are configured to be proxied through Cloudflare. Non-proxied
|
||||
records bypass Cloudflare's security and performance features.
|
||||
|
||||
Returns:
|
||||
A list of CheckReportCloudflare objects with PASS status if the
|
||||
record is proxied through Cloudflare, or FAIL status if it is
|
||||
DNS-only (not proxied).
|
||||
"""
|
||||
findings = []
|
||||
|
||||
for record in dns_client.records:
|
||||
# Only check proxyable record types
|
||||
if record.type not in PROXYABLE_TYPES:
|
||||
continue
|
||||
|
||||
report = CheckReportCloudflare(
|
||||
metadata=self.metadata(),
|
||||
resource=record,
|
||||
)
|
||||
|
||||
if record.proxied:
|
||||
report.status = "PASS"
|
||||
report.status_extended = f"DNS record {record.name} ({record.type}) is proxied through Cloudflare."
|
||||
else:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = f"DNS record {record.name} ({record.type}) is not proxied through Cloudflare."
|
||||
findings.append(report)
|
||||
|
||||
return findings
|
||||
@@ -4,7 +4,6 @@ from pydantic import BaseModel
|
||||
|
||||
from prowler.lib.logger import logger
|
||||
from prowler.providers.cloudflare.lib.service.service import CloudflareService
|
||||
from prowler.providers.cloudflare.services.zone.zone_client import zone_client
|
||||
|
||||
|
||||
class DNS(CloudflareService):
|
||||
@@ -19,10 +18,13 @@ class DNS(CloudflareService):
|
||||
"""List DNS records for all zones."""
|
||||
logger.info("DNS - Listing DNS records...")
|
||||
try:
|
||||
for zone in zone_client.zones.values():
|
||||
# Get zones directly from API to avoid circular dependency with zone_client
|
||||
zones = self._get_zones()
|
||||
|
||||
for zone_id, zone_name in zones.items():
|
||||
seen_record_ids: set[str] = set()
|
||||
try:
|
||||
for record in self.client.dns.records.list(zone_id=zone.id):
|
||||
for record in self.client.dns.records.list(zone_id=zone_id):
|
||||
record_id = getattr(record, "id", None)
|
||||
# Prevent infinite loop
|
||||
if record_id in seen_record_ids:
|
||||
@@ -32,8 +34,8 @@ class DNS(CloudflareService):
|
||||
self.records.append(
|
||||
CloudflareDNSRecord(
|
||||
id=record_id,
|
||||
zone_id=zone.id,
|
||||
zone_name=zone.name,
|
||||
zone_id=zone_id,
|
||||
zone_name=zone_name,
|
||||
name=getattr(record, "name", None),
|
||||
type=getattr(record, "type", None),
|
||||
content=getattr(record, "content", ""),
|
||||
@@ -43,13 +45,57 @@ class DNS(CloudflareService):
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{zone.id} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
f"{zone_id} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
def _get_zones(self) -> dict[str, str]:
|
||||
"""Get zones directly from Cloudflare API.
|
||||
|
||||
Returns:
|
||||
Dictionary mapping zone_id to zone_name.
|
||||
"""
|
||||
zones = {}
|
||||
audited_accounts = self.provider.identity.audited_accounts
|
||||
filter_zones = self.provider.filter_zones
|
||||
seen_zone_ids: set[str] = set()
|
||||
|
||||
try:
|
||||
for zone in self.client.zones.list():
|
||||
zone_id = getattr(zone, "id", None)
|
||||
# Prevent infinite loop - skip if we've seen this zone
|
||||
if zone_id in seen_zone_ids:
|
||||
break
|
||||
seen_zone_ids.add(zone_id)
|
||||
|
||||
zone_account = getattr(zone, "account", None)
|
||||
account_id = getattr(zone_account, "id", None) if zone_account else None
|
||||
|
||||
# Filter by audited accounts
|
||||
if audited_accounts and account_id not in audited_accounts:
|
||||
continue
|
||||
|
||||
zone_name = getattr(zone, "name", None)
|
||||
|
||||
# Apply zone filter if specified via --region
|
||||
if (
|
||||
filter_zones
|
||||
and zone_id not in filter_zones
|
||||
and zone_name not in filter_zones
|
||||
):
|
||||
continue
|
||||
|
||||
zones[zone_id] = zone_name
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
return zones
|
||||
|
||||
|
||||
class CloudflareDNSRecord(BaseModel):
|
||||
"""Cloudflare DNS record representation."""
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
from prowler.providers.cloudflare.services.firewall.firewall_service import Firewall
|
||||
from prowler.providers.common.provider import Provider
|
||||
|
||||
firewall_client = Firewall(Provider.get_global_provider())
|
||||
@@ -0,0 +1,123 @@
|
||||
from typing import Optional
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from prowler.lib.logger import logger
|
||||
from prowler.providers.cloudflare.lib.service.service import CloudflareService
|
||||
|
||||
|
||||
class Firewall(CloudflareService):
|
||||
"""Retrieve Cloudflare firewall rules for all zones."""
|
||||
|
||||
def __init__(self, provider):
|
||||
super().__init__(__class__.__name__, provider)
|
||||
self.rules: list["CloudflareFirewallRule"] = []
|
||||
self._list_rulesets()
|
||||
|
||||
def _list_rulesets(self) -> None:
|
||||
"""List firewall rulesets for all zones."""
|
||||
logger.info("Firewall - Listing firewall rulesets...")
|
||||
try:
|
||||
# Get zones directly from API to avoid circular dependency with zone_client
|
||||
zones = self._get_zones()
|
||||
|
||||
for zone_id, zone_name in zones.items():
|
||||
try:
|
||||
# Get all rulesets for the zone
|
||||
rulesets = self.client.rulesets.list(zone_id=zone_id)
|
||||
for ruleset in rulesets:
|
||||
ruleset_id = getattr(ruleset, "id", None)
|
||||
phase = getattr(ruleset, "phase", None)
|
||||
if not ruleset_id:
|
||||
continue
|
||||
|
||||
# Get rules within each ruleset
|
||||
try:
|
||||
ruleset_detail = self.client.rulesets.get(
|
||||
ruleset_id=ruleset_id, zone_id=zone_id
|
||||
)
|
||||
rules = getattr(ruleset_detail, "rules", []) or []
|
||||
for rule in rules:
|
||||
self.rules.append(
|
||||
CloudflareFirewallRule(
|
||||
id=getattr(rule, "id", None),
|
||||
zone_id=zone_id,
|
||||
zone_name=zone_name,
|
||||
ruleset_id=ruleset_id,
|
||||
phase=phase,
|
||||
action=getattr(rule, "action", None),
|
||||
expression=getattr(rule, "expression", None),
|
||||
description=getattr(rule, "description", None),
|
||||
enabled=getattr(rule, "enabled", True),
|
||||
)
|
||||
)
|
||||
except Exception as error:
|
||||
logger.debug(
|
||||
f"{zone_id} ruleset {ruleset_id} -- {error.__class__.__name__}: {error}"
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{zone_id} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
def _get_zones(self) -> dict[str, str]:
|
||||
"""Get zones directly from Cloudflare API.
|
||||
|
||||
Returns:
|
||||
Dictionary mapping zone_id to zone_name.
|
||||
"""
|
||||
zones = {}
|
||||
audited_accounts = self.provider.identity.audited_accounts
|
||||
filter_zones = self.provider.filter_zones
|
||||
seen_zone_ids: set[str] = set()
|
||||
|
||||
try:
|
||||
for zone in self.client.zones.list():
|
||||
zone_id = getattr(zone, "id", None)
|
||||
# Prevent infinite loop - skip if we've seen this zone
|
||||
if zone_id in seen_zone_ids:
|
||||
break
|
||||
seen_zone_ids.add(zone_id)
|
||||
|
||||
zone_account = getattr(zone, "account", None)
|
||||
account_id = getattr(zone_account, "id", None) if zone_account else None
|
||||
|
||||
# Filter by audited accounts
|
||||
if audited_accounts and account_id not in audited_accounts:
|
||||
continue
|
||||
|
||||
zone_name = getattr(zone, "name", None)
|
||||
|
||||
# Apply zone filter if specified via --region
|
||||
if (
|
||||
filter_zones
|
||||
and zone_id not in filter_zones
|
||||
and zone_name not in filter_zones
|
||||
):
|
||||
continue
|
||||
|
||||
zones[zone_id] = zone_name
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
return zones
|
||||
|
||||
|
||||
class CloudflareFirewallRule(BaseModel):
|
||||
"""Cloudflare firewall rule representation."""
|
||||
|
||||
id: Optional[str] = None
|
||||
zone_id: str
|
||||
zone_name: str
|
||||
ruleset_id: Optional[str] = None
|
||||
phase: Optional[str] = None
|
||||
action: Optional[str] = None
|
||||
expression: Optional[str] = None
|
||||
description: Optional[str] = None
|
||||
enabled: bool = True
|
||||
+36
@@ -0,0 +1,36 @@
|
||||
{
|
||||
"Provider": "cloudflare",
|
||||
"CheckID": "zone_firewall_blocking_rules_configured",
|
||||
"CheckTitle": "Cloudflare Zone Firewall Rules Use Blocking Actions to Protect Against Threats",
|
||||
"CheckType": [],
|
||||
"ServiceName": "zone",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "medium",
|
||||
"ResourceType": "Zone",
|
||||
"ResourceGroup": "network",
|
||||
"Description": "**Cloudflare zones** are assessed for **firewall blocking rules** by checking if custom rules use block, challenge, js_challenge, or managed_challenge actions to actively protect against threats rather than only logging.",
|
||||
"Risk": "Firewall rules configured only for **logging** provide visibility but no protection.\n- **Confidentiality**: malicious traffic can access and exfiltrate sensitive data\n- **Integrity**: application exploits can modify data without being blocked\n- **Availability**: credential stuffing and abuse attacks reach the origin unimpeded",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://developers.cloudflare.com/waf/custom-rules/"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Log in to the Cloudflare dashboard and select your account and domain\n2. Go to Security > WAF > Custom rules\n3. Review existing rules and their actions\n4. Update rules to use blocking actions (Block, Challenge, JS Challenge, Managed Challenge)\n5. Test rules in log mode first, then enable blocking actions",
|
||||
"Terraform": "```hcl\n# Configure firewall rule with blocking action\nresource \"cloudflare_ruleset\" \"blocking_rule\" {\n zone_id = \"<ZONE_ID>\"\n name = \"Block malicious requests\"\n kind = \"zone\"\n phase = \"http_request_firewall_custom\"\n rules {\n action = \"block\" # Actively blocks matching traffic\n expression = \"(ip.geoip.country eq \\\"XX\\\")\"\n description = \"Block traffic from high-risk country\"\n }\n}\n```"
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Configure **firewall rules** with blocking actions to enforce security policies.\n- Use challenge actions for suspicious traffic to verify human visitors\n- Use block actions for known malicious patterns and high-risk sources\n- Test rules in log mode before enabling blocking to avoid false positives\n- Follow the principle of least privilege in rule configuration",
|
||||
"Url": "https://hub.prowler.com/checks/cloudflare/zone_firewall_blocking_rules_configured"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"internet-exposed"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": "Blocking actions include: block, challenge, js_challenge, managed_challenge. Log-only rules provide visibility but do not prevent attacks."
|
||||
}
|
||||
+53
@@ -0,0 +1,53 @@
|
||||
from prowler.lib.check.models import Check, CheckReportCloudflare
|
||||
from prowler.providers.cloudflare.services.zone.zone_client import zone_client
|
||||
|
||||
BLOCKING_ACTIONS = {"block", "challenge", "js_challenge", "managed_challenge"}
|
||||
|
||||
|
||||
class zone_firewall_blocking_rules_configured(Check):
|
||||
"""Ensure that firewall rules with blocking actions are configured for Cloudflare zones.
|
||||
|
||||
Firewall rules should use blocking actions (block, challenge, js_challenge,
|
||||
managed_challenge) to actively protect against threats rather than only logging
|
||||
traffic. Without blocking actions, malicious requests can reach the origin server
|
||||
and potentially compromise the application's security.
|
||||
"""
|
||||
|
||||
def execute(self) -> list[CheckReportCloudflare]:
|
||||
"""Execute the firewall blocking rules configured check.
|
||||
|
||||
Iterates through all Cloudflare zones and verifies that at least one
|
||||
firewall rule exists with a blocking action. Blocking actions include
|
||||
block, challenge, js_challenge, and managed_challenge.
|
||||
|
||||
Returns:
|
||||
A list of CheckReportCloudflare objects with PASS status if blocking
|
||||
rules are configured, or FAIL status if no blocking rules exist.
|
||||
"""
|
||||
findings = []
|
||||
|
||||
for zone in zone_client.zones.values():
|
||||
report = CheckReportCloudflare(
|
||||
metadata=self.metadata(),
|
||||
resource=zone,
|
||||
)
|
||||
|
||||
# Find blocking rules for this zone
|
||||
blocking_rules = [
|
||||
rule for rule in zone.firewall_rules if rule.action in BLOCKING_ACTIONS
|
||||
]
|
||||
|
||||
if blocking_rules:
|
||||
report.status = "PASS"
|
||||
report.status_extended = (
|
||||
f"Zone {zone.name} has firewall rules with blocking actions "
|
||||
f"({len(blocking_rules)} rule(s))."
|
||||
)
|
||||
else:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
f"Zone {zone.name} has no firewall rules with blocking actions."
|
||||
)
|
||||
findings.append(report)
|
||||
|
||||
return findings
|
||||
@@ -17,6 +17,31 @@ class CloudflareRateLimitRule(BaseModel):
|
||||
expression: Optional[str] = None
|
||||
|
||||
|
||||
class CloudflareFirewallRule(BaseModel):
|
||||
"""Represents a firewall rule from custom rulesets."""
|
||||
|
||||
id: str
|
||||
name: str = ""
|
||||
description: Optional[str] = None
|
||||
action: Optional[str] = None
|
||||
enabled: bool = True
|
||||
expression: Optional[str] = None
|
||||
phase: Optional[str] = None
|
||||
|
||||
class Config:
|
||||
arbitrary_types_allowed = True
|
||||
|
||||
|
||||
class CloudflareWAFRuleset(BaseModel):
|
||||
"""Represents a WAF ruleset (managed rules) for a zone."""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
kind: Optional[str] = None
|
||||
phase: Optional[str] = None
|
||||
enabled: bool = True
|
||||
|
||||
|
||||
class Zone(CloudflareService):
|
||||
"""Retrieve Cloudflare zones with security-relevant settings."""
|
||||
|
||||
@@ -29,6 +54,8 @@ class Zone(CloudflareService):
|
||||
self._get_zones_universal_ssl()
|
||||
self._get_zones_rate_limit_rules()
|
||||
self._get_zones_bot_management()
|
||||
self._get_zones_firewall_rules()
|
||||
self._get_zones_waf_rulesets()
|
||||
|
||||
def _list_zones(self) -> None:
|
||||
"""List all Cloudflare zones with their basic information."""
|
||||
@@ -191,6 +218,109 @@ class Zone(CloudflareService):
|
||||
f"{zone.id} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
def _get_zones_firewall_rules(self) -> None:
|
||||
"""Get firewall rules for all zones."""
|
||||
logger.info("Zones - Getting firewall rules...")
|
||||
for zone in self.zones.values():
|
||||
try:
|
||||
self._get_zone_firewall_rules(zone)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{zone.id} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
def _get_zone_firewall_rules(self, zone: "CloudflareZone") -> None:
|
||||
"""List firewall rules from custom rulesets for a zone."""
|
||||
seen_ruleset_ids: set[str] = set()
|
||||
try:
|
||||
for ruleset in self.client.rulesets.list(zone_id=zone.id):
|
||||
ruleset_id = getattr(ruleset, "id", "")
|
||||
if ruleset_id in seen_ruleset_ids:
|
||||
break
|
||||
seen_ruleset_ids.add(ruleset_id)
|
||||
|
||||
ruleset_phase = getattr(ruleset, "phase", "")
|
||||
if ruleset_phase in [
|
||||
"http_request_firewall_custom",
|
||||
"http_ratelimit",
|
||||
"http_request_firewall_managed",
|
||||
]:
|
||||
try:
|
||||
ruleset_detail = self.client.rulesets.get(
|
||||
ruleset_id=ruleset_id, zone_id=zone.id
|
||||
)
|
||||
rules = getattr(ruleset_detail, "rules", []) or []
|
||||
seen_rule_ids: set[str] = set()
|
||||
for rule in rules:
|
||||
rule_id = getattr(rule, "id", "")
|
||||
if rule_id in seen_rule_ids:
|
||||
break
|
||||
seen_rule_ids.add(rule_id)
|
||||
try:
|
||||
zone.firewall_rules.append(
|
||||
CloudflareFirewallRule(
|
||||
id=rule_id,
|
||||
name=getattr(rule, "description", "")
|
||||
or rule_id,
|
||||
description=getattr(rule, "description", None),
|
||||
action=getattr(rule, "action", None),
|
||||
enabled=getattr(rule, "enabled", True),
|
||||
expression=getattr(rule, "expression", None),
|
||||
phase=ruleset_phase,
|
||||
)
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
def _get_zones_waf_rulesets(self) -> None:
|
||||
"""Get WAF rulesets for all zones."""
|
||||
logger.info("Zones - Getting WAF rulesets...")
|
||||
for zone in self.zones.values():
|
||||
try:
|
||||
self._get_zone_waf_rulesets(zone)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{zone.id} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
def _get_zone_waf_rulesets(self, zone: "CloudflareZone") -> None:
|
||||
"""List WAF rulesets for a zone using the rulesets API."""
|
||||
seen_ids: set[str] = set()
|
||||
try:
|
||||
for ruleset in self.client.rulesets.list(zone_id=zone.id):
|
||||
ruleset_id = getattr(ruleset, "id", "")
|
||||
if ruleset_id in seen_ids:
|
||||
break
|
||||
seen_ids.add(ruleset_id)
|
||||
try:
|
||||
zone.waf_rulesets.append(
|
||||
CloudflareWAFRuleset(
|
||||
id=ruleset_id,
|
||||
name=getattr(ruleset, "name", ""),
|
||||
kind=getattr(ruleset, "kind", None),
|
||||
phase=getattr(ruleset, "phase", None),
|
||||
enabled=True,
|
||||
)
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
except Exception as error:
|
||||
logger.error(
|
||||
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
|
||||
)
|
||||
|
||||
def _get_zone_setting(self, zone_id: str, setting_id: str):
|
||||
"""Get a single zone setting by ID."""
|
||||
try:
|
||||
@@ -326,3 +456,5 @@ class CloudflareZone(BaseModel):
|
||||
settings: CloudflareZoneSettings = Field(default_factory=CloudflareZoneSettings)
|
||||
dnssec_status: Optional[str] = None
|
||||
rate_limit_rules: list[CloudflareRateLimitRule] = Field(default_factory=list)
|
||||
firewall_rules: list[CloudflareFirewallRule] = Field(default_factory=list)
|
||||
waf_rulesets: list[CloudflareWAFRuleset] = Field(default_factory=list)
|
||||
|
||||
+36
@@ -0,0 +1,36 @@
|
||||
{
|
||||
"Provider": "cloudflare",
|
||||
"CheckID": "zone_waf_owasp_ruleset_enabled",
|
||||
"CheckTitle": "Cloudflare Zone OWASP Managed WAF Rulesets Are Enabled",
|
||||
"CheckType": [],
|
||||
"ServiceName": "zone",
|
||||
"SubServiceName": "",
|
||||
"ResourceIdTemplate": "",
|
||||
"Severity": "high",
|
||||
"ResourceType": "Zone",
|
||||
"ResourceGroup": "network",
|
||||
"Description": "**Cloudflare zones** are assessed for **OWASP managed rulesets** by checking if they are enabled to protect against common web application vulnerabilities including **SQL injection**, **XSS**, and other **OWASP Top 10** threats.",
|
||||
"Risk": "Without **OWASP managed rulesets**, web applications are exposed to well-known attack vectors.\n- **Confidentiality**: SQL injection attacks can exfiltrate sensitive database contents\n- **Integrity**: XSS attacks can modify page content and steal session tokens\n- **Availability**: remote code execution can compromise server availability",
|
||||
"RelatedUrl": "",
|
||||
"AdditionalURLs": [
|
||||
"https://developers.cloudflare.com/waf/managed-rules/"
|
||||
],
|
||||
"Remediation": {
|
||||
"Code": {
|
||||
"CLI": "",
|
||||
"NativeIaC": "",
|
||||
"Other": "1. Log in to the Cloudflare dashboard and select your account and domain\n2. Go to Security > WAF > Managed rules\n3. Enable the Cloudflare OWASP Core Ruleset\n4. Review and configure rule sensitivity based on your application\n5. Monitor WAF analytics to tune rules and reduce false positives",
|
||||
"Terraform": "```hcl\n# Enable OWASP managed WAF rulesets\nresource \"cloudflare_ruleset\" \"waf_owasp\" {\n zone_id = \"<ZONE_ID>\"\n name = \"OWASP Managed Rules\"\n kind = \"zone\"\n phase = \"http_request_firewall_managed\"\n rules {\n action = \"execute\"\n action_parameters {\n id = \"4814384a9e5d4991b9815dcfc25d2f1f\" # Cloudflare OWASP Core Ruleset\n }\n expression = \"true\"\n description = \"Execute Cloudflare OWASP Core Ruleset\"\n }\n}\n```"
|
||||
},
|
||||
"Recommendation": {
|
||||
"Text": "Enable **OWASP Core Ruleset** managed rules as part of a defense in depth strategy.\n- Protects against OWASP Top 10 vulnerabilities including SQLi and XSS\n- Regularly review and tune rule sensitivity based on application requirements\n- Monitor WAF analytics to identify and address false positives\n- Combine with custom rules for application-specific protection",
|
||||
"Url": "https://hub.prowler.com/checks/cloudflare/zone_waf_owasp_ruleset_enabled"
|
||||
}
|
||||
},
|
||||
"Categories": [
|
||||
"vulnerabilities"
|
||||
],
|
||||
"DependsOn": [],
|
||||
"RelatedTo": [],
|
||||
"Notes": "OWASP managed rulesets are available on Pro, Business, and Enterprise plans. The Cloudflare OWASP Core Ruleset provides protection against common web application vulnerabilities."
|
||||
}
|
||||
+58
@@ -0,0 +1,58 @@
|
||||
from prowler.lib.check.models import Check, CheckReportCloudflare
|
||||
from prowler.providers.cloudflare.services.zone.zone_client import zone_client
|
||||
|
||||
|
||||
class zone_waf_owasp_ruleset_enabled(Check):
|
||||
"""Ensure that OWASP managed WAF rulesets are enabled for Cloudflare zones.
|
||||
|
||||
The OWASP Core Ruleset provides protection against common web application
|
||||
vulnerabilities including SQL injection, cross-site scripting (XSS), and other
|
||||
OWASP Top 10 threats. These managed rulesets are essential for defense in depth
|
||||
and protecting applications from well-known attack vectors.
|
||||
"""
|
||||
|
||||
def execute(self) -> list[CheckReportCloudflare]:
|
||||
"""Execute the OWASP WAF ruleset enabled check.
|
||||
|
||||
Iterates through all Cloudflare zones and verifies that OWASP managed
|
||||
WAF rulesets are enabled. The check identifies OWASP rulesets by name
|
||||
containing "owasp" or by the http_request_firewall_managed phase.
|
||||
|
||||
Returns:
|
||||
A list of CheckReportCloudflare objects with PASS status if OWASP
|
||||
rulesets are enabled, or FAIL status if no OWASP protection exists.
|
||||
"""
|
||||
findings = []
|
||||
|
||||
for zone in zone_client.zones.values():
|
||||
report = CheckReportCloudflare(
|
||||
metadata=self.metadata(),
|
||||
resource=zone,
|
||||
)
|
||||
|
||||
# Find OWASP managed rulesets for this zone
|
||||
# Only match rulesets that explicitly contain "owasp" in the name
|
||||
# The phase check was too broad as it matched any managed ruleset
|
||||
owasp_rulesets = [
|
||||
ruleset
|
||||
for ruleset in zone.waf_rulesets
|
||||
if "owasp" in (ruleset.name or "").lower()
|
||||
]
|
||||
|
||||
if owasp_rulesets:
|
||||
report.status = "PASS"
|
||||
ruleset_descriptions = ", ".join(
|
||||
ruleset.name for ruleset in owasp_rulesets
|
||||
)
|
||||
report.status_extended = (
|
||||
f"Zone {zone.name} has OWASP managed WAF ruleset enabled: "
|
||||
f"{ruleset_descriptions}."
|
||||
)
|
||||
else:
|
||||
report.status = "FAIL"
|
||||
report.status_extended = (
|
||||
f"Zone {zone.name} does not have OWASP managed WAF ruleset enabled."
|
||||
)
|
||||
findings.append(report)
|
||||
|
||||
return findings
|
||||
+425
@@ -0,0 +1,425 @@
|
||||
from typing import Optional
|
||||
from unittest import mock
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from tests.providers.cloudflare.cloudflare_fixtures import (
|
||||
ZONE_ID,
|
||||
ZONE_NAME,
|
||||
set_mocked_cloudflare_provider,
|
||||
)
|
||||
|
||||
|
||||
class CloudflareDNSRecord(BaseModel):
|
||||
"""Cloudflare DNS record representation for testing."""
|
||||
|
||||
id: str
|
||||
zone_id: str
|
||||
zone_name: str
|
||||
name: Optional[str] = None
|
||||
type: Optional[str] = None
|
||||
content: str = ""
|
||||
ttl: Optional[int] = None
|
||||
proxied: bool = False
|
||||
|
||||
|
||||
class Test_dns_record_cname_target_valid:
|
||||
def test_no_records(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = []
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid import (
|
||||
dns_record_cname_target_valid,
|
||||
)
|
||||
|
||||
check = dns_record_cname_target_valid()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
def test_non_cname_record(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="www.example.com",
|
||||
type="A",
|
||||
content="192.0.2.1",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid import (
|
||||
dns_record_cname_target_valid,
|
||||
)
|
||||
|
||||
check = dns_record_cname_target_valid()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
def test_cname_record_valid_target(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="www.example.com",
|
||||
type="CNAME",
|
||||
content="example.com",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
mock.patch(
|
||||
"socket.getaddrinfo",
|
||||
return_value=[("", "", "", "", ("192.0.2.1", 0))],
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid import (
|
||||
dns_record_cname_target_valid,
|
||||
)
|
||||
|
||||
check = dns_record_cname_target_valid()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].resource_id == "record-1"
|
||||
assert result[0].resource_name == "www.example.com"
|
||||
assert result[0].status == "PASS"
|
||||
assert "points to valid target" in result[0].status_extended
|
||||
|
||||
def test_cname_record_dangling_target(self):
|
||||
import socket
|
||||
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="old.example.com",
|
||||
type="CNAME",
|
||||
content="nonexistent.example.com",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
mock.patch(
|
||||
"socket.getaddrinfo",
|
||||
side_effect=socket.gaierror("Name or service not known"),
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid import (
|
||||
dns_record_cname_target_valid,
|
||||
)
|
||||
|
||||
check = dns_record_cname_target_valid()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "potentially dangling target" in result[0].status_extended
|
||||
assert "subdomain takeover risk" in result[0].status_extended
|
||||
|
||||
def test_cname_record_with_trailing_dot(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="www.example.com",
|
||||
type="CNAME",
|
||||
content="example.com.",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
mock.patch(
|
||||
"socket.getaddrinfo",
|
||||
return_value=[("", "", "", "", ("192.0.2.1", 0))],
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid import (
|
||||
dns_record_cname_target_valid,
|
||||
)
|
||||
|
||||
check = dns_record_cname_target_valid()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
|
||||
def test_mx_record_valid_target(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="example.com",
|
||||
type="MX",
|
||||
content="10 mail.example.com",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
mock.patch(
|
||||
"socket.getaddrinfo",
|
||||
return_value=[("", "", "", "", ("192.0.2.1", 0))],
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid import (
|
||||
dns_record_cname_target_valid,
|
||||
)
|
||||
|
||||
check = dns_record_cname_target_valid()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert "MX record" in result[0].status_extended
|
||||
assert "points to valid target" in result[0].status_extended
|
||||
|
||||
def test_mx_record_dangling_target(self):
|
||||
import socket
|
||||
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="example.com",
|
||||
type="MX",
|
||||
content="10 nonexistent-mail.example.com",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
mock.patch(
|
||||
"socket.getaddrinfo",
|
||||
side_effect=socket.gaierror("Name or service not known"),
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid import (
|
||||
dns_record_cname_target_valid,
|
||||
)
|
||||
|
||||
check = dns_record_cname_target_valid()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "MX record" in result[0].status_extended
|
||||
assert "mail interception risk" in result[0].status_extended
|
||||
|
||||
def test_ns_record_valid_target(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="sub.example.com",
|
||||
type="NS",
|
||||
content="ns1.example.com",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
mock.patch(
|
||||
"socket.getaddrinfo",
|
||||
return_value=[("", "", "", "", ("192.0.2.1", 0))],
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid import (
|
||||
dns_record_cname_target_valid,
|
||||
)
|
||||
|
||||
check = dns_record_cname_target_valid()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert "NS record" in result[0].status_extended
|
||||
|
||||
def test_ns_record_dangling_target(self):
|
||||
import socket
|
||||
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="sub.example.com",
|
||||
type="NS",
|
||||
content="nonexistent-ns.example.com",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
mock.patch(
|
||||
"socket.getaddrinfo",
|
||||
side_effect=socket.gaierror("Name or service not known"),
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid import (
|
||||
dns_record_cname_target_valid,
|
||||
)
|
||||
|
||||
check = dns_record_cname_target_valid()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "NS record" in result[0].status_extended
|
||||
assert "subdomain delegation takeover risk" in result[0].status_extended
|
||||
|
||||
def test_srv_record_valid_target(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="_sip._tcp.example.com",
|
||||
type="SRV",
|
||||
content="10 5 5060 sip.example.com",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
mock.patch(
|
||||
"socket.getaddrinfo",
|
||||
return_value=[("", "", "", "", ("192.0.2.1", 0))],
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid import (
|
||||
dns_record_cname_target_valid,
|
||||
)
|
||||
|
||||
check = dns_record_cname_target_valid()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert "SRV record" in result[0].status_extended
|
||||
|
||||
def test_srv_record_dangling_target(self):
|
||||
import socket
|
||||
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="_sip._tcp.example.com",
|
||||
type="SRV",
|
||||
content="10 5 5060 nonexistent-sip.example.com",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
mock.patch(
|
||||
"socket.getaddrinfo",
|
||||
side_effect=socket.gaierror("Name or service not known"),
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_cname_target_valid.dns_record_cname_target_valid import (
|
||||
dns_record_cname_target_valid,
|
||||
)
|
||||
|
||||
check = dns_record_cname_target_valid()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "SRV record" in result[0].status_extended
|
||||
assert "service discovery vulnerability" in result[0].status_extended
|
||||
+312
@@ -0,0 +1,312 @@
|
||||
from typing import Optional
|
||||
from unittest import mock
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from tests.providers.cloudflare.cloudflare_fixtures import (
|
||||
ZONE_ID,
|
||||
ZONE_NAME,
|
||||
set_mocked_cloudflare_provider,
|
||||
)
|
||||
|
||||
|
||||
class CloudflareDNSRecord(BaseModel):
|
||||
"""Cloudflare DNS record representation for testing."""
|
||||
|
||||
id: str
|
||||
zone_id: str
|
||||
zone_name: str
|
||||
name: Optional[str] = None
|
||||
type: Optional[str] = None
|
||||
content: str = ""
|
||||
ttl: Optional[int] = None
|
||||
proxied: bool = False
|
||||
|
||||
|
||||
class Test_dns_record_no_internal_ip:
|
||||
def test_no_records(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = []
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip import (
|
||||
dns_record_no_internal_ip,
|
||||
)
|
||||
|
||||
check = dns_record_no_internal_ip()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
def test_non_ip_record(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="www.example.com",
|
||||
type="CNAME",
|
||||
content="example.com",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip import (
|
||||
dns_record_no_internal_ip,
|
||||
)
|
||||
|
||||
check = dns_record_no_internal_ip()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
def test_a_record_public_ip(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="www.example.com",
|
||||
type="A",
|
||||
content="8.8.8.8", # Google DNS - a truly public IP
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip import (
|
||||
dns_record_no_internal_ip,
|
||||
)
|
||||
|
||||
check = dns_record_no_internal_ip()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].resource_id == "record-1"
|
||||
assert result[0].resource_name == "www.example.com"
|
||||
assert result[0].status == "PASS"
|
||||
assert "public IP address" in result[0].status_extended
|
||||
|
||||
def test_a_record_private_ip_10(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="internal.example.com",
|
||||
type="A",
|
||||
content="10.0.0.1",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip import (
|
||||
dns_record_no_internal_ip,
|
||||
)
|
||||
|
||||
check = dns_record_no_internal_ip()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "internal IP address" in result[0].status_extended
|
||||
assert "information disclosure risk" in result[0].status_extended
|
||||
|
||||
def test_a_record_private_ip_172(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="internal.example.com",
|
||||
type="A",
|
||||
content="172.16.0.1",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip import (
|
||||
dns_record_no_internal_ip,
|
||||
)
|
||||
|
||||
check = dns_record_no_internal_ip()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "internal IP address" in result[0].status_extended
|
||||
|
||||
def test_a_record_private_ip_192(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="internal.example.com",
|
||||
type="A",
|
||||
content="192.168.1.1",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip import (
|
||||
dns_record_no_internal_ip,
|
||||
)
|
||||
|
||||
check = dns_record_no_internal_ip()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "internal IP address" in result[0].status_extended
|
||||
|
||||
def test_a_record_loopback(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="localhost.example.com",
|
||||
type="A",
|
||||
content="127.0.0.1",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip import (
|
||||
dns_record_no_internal_ip,
|
||||
)
|
||||
|
||||
check = dns_record_no_internal_ip()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "internal IP address" in result[0].status_extended
|
||||
|
||||
def test_aaaa_record_public_ip(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="www.example.com",
|
||||
type="AAAA",
|
||||
content="2001:db8::1",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip import (
|
||||
dns_record_no_internal_ip,
|
||||
)
|
||||
|
||||
check = dns_record_no_internal_ip()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
# 2001:db8:: is documentation prefix and is reserved
|
||||
assert result[0].status == "FAIL"
|
||||
|
||||
def test_aaaa_record_link_local(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="internal.example.com",
|
||||
type="AAAA",
|
||||
content="fe80::1",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_internal_ip.dns_record_no_internal_ip import (
|
||||
dns_record_no_internal_ip,
|
||||
)
|
||||
|
||||
check = dns_record_no_internal_ip()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "internal IP address" in result[0].status_extended
|
||||
+345
@@ -0,0 +1,345 @@
|
||||
from typing import Optional
|
||||
from unittest import mock
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from tests.providers.cloudflare.cloudflare_fixtures import (
|
||||
ZONE_ID,
|
||||
ZONE_NAME,
|
||||
set_mocked_cloudflare_provider,
|
||||
)
|
||||
|
||||
|
||||
class CloudflareDNSRecord(BaseModel):
|
||||
"""Cloudflare DNS record representation for testing."""
|
||||
|
||||
id: str
|
||||
zone_id: str
|
||||
zone_name: str
|
||||
name: Optional[str] = None
|
||||
type: Optional[str] = None
|
||||
content: str = ""
|
||||
ttl: Optional[int] = None
|
||||
proxied: bool = False
|
||||
|
||||
|
||||
class Test_dns_record_no_wildcard:
|
||||
def test_no_records(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = []
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard import (
|
||||
dns_record_no_wildcard,
|
||||
)
|
||||
|
||||
check = dns_record_no_wildcard()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
def test_non_ip_record(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="example.com",
|
||||
type="TXT",
|
||||
content="v=spf1 include:_spf.google.com ~all",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard import (
|
||||
dns_record_no_wildcard,
|
||||
)
|
||||
|
||||
check = dns_record_no_wildcard()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
def test_a_record_not_wildcard(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="www.example.com",
|
||||
type="A",
|
||||
content="8.8.8.8",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard import (
|
||||
dns_record_no_wildcard,
|
||||
)
|
||||
|
||||
check = dns_record_no_wildcard()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].resource_id == "record-1"
|
||||
assert result[0].resource_name == "www.example.com"
|
||||
assert result[0].status == "PASS"
|
||||
assert "is not a wildcard record" in result[0].status_extended
|
||||
|
||||
def test_a_record_wildcard(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="*.example.com",
|
||||
type="A",
|
||||
content="8.8.8.8",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard import (
|
||||
dns_record_no_wildcard,
|
||||
)
|
||||
|
||||
check = dns_record_no_wildcard()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "is a wildcard record" in result[0].status_extended
|
||||
assert "may expose unintended services" in result[0].status_extended
|
||||
|
||||
def test_aaaa_record_wildcard(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="*.example.com",
|
||||
type="AAAA",
|
||||
content="2001:db8::1",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard import (
|
||||
dns_record_no_wildcard,
|
||||
)
|
||||
|
||||
check = dns_record_no_wildcard()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "is a wildcard record" in result[0].status_extended
|
||||
|
||||
def test_cname_record_wildcard(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="*.example.com",
|
||||
type="CNAME",
|
||||
content="example.com",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard import (
|
||||
dns_record_no_wildcard,
|
||||
)
|
||||
|
||||
check = dns_record_no_wildcard()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "is a wildcard record" in result[0].status_extended
|
||||
|
||||
def test_cname_record_not_wildcard(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="www.example.com",
|
||||
type="CNAME",
|
||||
content="example.com",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard import (
|
||||
dns_record_no_wildcard,
|
||||
)
|
||||
|
||||
check = dns_record_no_wildcard()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert "is not a wildcard record" in result[0].status_extended
|
||||
|
||||
def test_mx_record_wildcard(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="*.example.com",
|
||||
type="MX",
|
||||
content="10 mail.example.com",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard import (
|
||||
dns_record_no_wildcard,
|
||||
)
|
||||
|
||||
check = dns_record_no_wildcard()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "is a wildcard record" in result[0].status_extended
|
||||
|
||||
def test_mx_record_not_wildcard(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="example.com",
|
||||
type="MX",
|
||||
content="10 mail.example.com",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard import (
|
||||
dns_record_no_wildcard,
|
||||
)
|
||||
|
||||
check = dns_record_no_wildcard()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert "is not a wildcard record" in result[0].status_extended
|
||||
|
||||
def test_srv_record_wildcard(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="*._tcp.example.com",
|
||||
type="SRV",
|
||||
content="10 5 5060 sip.example.com",
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_no_wildcard.dns_record_no_wildcard import (
|
||||
dns_record_no_wildcard,
|
||||
)
|
||||
|
||||
check = dns_record_no_wildcard()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "is a wildcard record" in result[0].status_extended
|
||||
+288
@@ -0,0 +1,288 @@
|
||||
from typing import Optional
|
||||
from unittest import mock
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from tests.providers.cloudflare.cloudflare_fixtures import (
|
||||
ZONE_ID,
|
||||
ZONE_NAME,
|
||||
set_mocked_cloudflare_provider,
|
||||
)
|
||||
|
||||
|
||||
class CloudflareDNSRecord(BaseModel):
|
||||
"""Cloudflare DNS record representation for testing."""
|
||||
|
||||
id: str
|
||||
zone_id: str
|
||||
zone_name: str
|
||||
name: Optional[str] = None
|
||||
type: Optional[str] = None
|
||||
content: str = ""
|
||||
ttl: Optional[int] = None
|
||||
proxied: bool = False
|
||||
|
||||
|
||||
class Test_dns_record_proxied:
|
||||
def test_no_records(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = []
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_proxied.dns_record_proxied.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_proxied.dns_record_proxied import (
|
||||
dns_record_proxied,
|
||||
)
|
||||
|
||||
check = dns_record_proxied()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
def test_non_proxyable_record(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="example.com",
|
||||
type="TXT",
|
||||
content="v=spf1 include:_spf.google.com ~all",
|
||||
proxied=False,
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_proxied.dns_record_proxied.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_proxied.dns_record_proxied import (
|
||||
dns_record_proxied,
|
||||
)
|
||||
|
||||
check = dns_record_proxied()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
def test_a_record_proxied(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="www.example.com",
|
||||
type="A",
|
||||
content="8.8.8.8",
|
||||
proxied=True,
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_proxied.dns_record_proxied.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_proxied.dns_record_proxied import (
|
||||
dns_record_proxied,
|
||||
)
|
||||
|
||||
check = dns_record_proxied()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].resource_id == "record-1"
|
||||
assert result[0].resource_name == "www.example.com"
|
||||
assert result[0].status == "PASS"
|
||||
assert "is proxied through Cloudflare" in result[0].status_extended
|
||||
# DNS records should have zone_name as region
|
||||
assert result[0].region == ZONE_NAME
|
||||
assert result[0].zone_name == ZONE_NAME
|
||||
|
||||
def test_a_record_not_proxied(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="www.example.com",
|
||||
type="A",
|
||||
content="8.8.8.8",
|
||||
proxied=False,
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_proxied.dns_record_proxied.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_proxied.dns_record_proxied import (
|
||||
dns_record_proxied,
|
||||
)
|
||||
|
||||
check = dns_record_proxied()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "is not proxied through Cloudflare" in result[0].status_extended
|
||||
|
||||
def test_aaaa_record_proxied(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="www.example.com",
|
||||
type="AAAA",
|
||||
content="2001:db8::1",
|
||||
proxied=True,
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_proxied.dns_record_proxied.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_proxied.dns_record_proxied import (
|
||||
dns_record_proxied,
|
||||
)
|
||||
|
||||
check = dns_record_proxied()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert "is proxied through Cloudflare" in result[0].status_extended
|
||||
|
||||
def test_aaaa_record_not_proxied(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="www.example.com",
|
||||
type="AAAA",
|
||||
content="2001:db8::1",
|
||||
proxied=False,
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_proxied.dns_record_proxied.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_proxied.dns_record_proxied import (
|
||||
dns_record_proxied,
|
||||
)
|
||||
|
||||
check = dns_record_proxied()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "is not proxied through Cloudflare" in result[0].status_extended
|
||||
|
||||
def test_cname_record_proxied(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="www.example.com",
|
||||
type="CNAME",
|
||||
content="example.com",
|
||||
proxied=True,
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_proxied.dns_record_proxied.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_proxied.dns_record_proxied import (
|
||||
dns_record_proxied,
|
||||
)
|
||||
|
||||
check = dns_record_proxied()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert "is proxied through Cloudflare" in result[0].status_extended
|
||||
|
||||
def test_cname_record_not_proxied(self):
|
||||
dns_client = mock.MagicMock
|
||||
dns_client.records = [
|
||||
CloudflareDNSRecord(
|
||||
id="record-1",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
name="www.example.com",
|
||||
type="CNAME",
|
||||
content="example.com",
|
||||
proxied=False,
|
||||
)
|
||||
]
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.dns.dns_record_proxied.dns_record_proxied.dns_client",
|
||||
new=dns_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.dns.dns_record_proxied.dns_record_proxied import (
|
||||
dns_record_proxied,
|
||||
)
|
||||
|
||||
check = dns_record_proxied()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert "is not proxied through Cloudflare" in result[0].status_extended
|
||||
@@ -0,0 +1,84 @@
|
||||
from typing import Optional
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from tests.providers.cloudflare.cloudflare_fixtures import ZONE_ID, ZONE_NAME
|
||||
|
||||
|
||||
class CloudflareFirewallRule(BaseModel):
|
||||
"""Cloudflare firewall rule representation for testing."""
|
||||
|
||||
id: Optional[str] = None
|
||||
zone_id: str
|
||||
zone_name: str
|
||||
ruleset_id: Optional[str] = None
|
||||
phase: Optional[str] = None
|
||||
action: Optional[str] = None
|
||||
expression: Optional[str] = None
|
||||
description: Optional[str] = None
|
||||
enabled: bool = True
|
||||
|
||||
|
||||
class TestFirewallService:
|
||||
def test_cloudflare_firewall_rule_model(self):
|
||||
rule = CloudflareFirewallRule(
|
||||
id="rule-123",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
ruleset_id="ruleset-456",
|
||||
phase="http_ratelimit",
|
||||
action="block",
|
||||
expression="(http.request.uri.path contains '/api/')",
|
||||
description="Rate limit API requests",
|
||||
enabled=True,
|
||||
)
|
||||
|
||||
assert rule.id == "rule-123"
|
||||
assert rule.zone_id == ZONE_ID
|
||||
assert rule.zone_name == ZONE_NAME
|
||||
assert rule.ruleset_id == "ruleset-456"
|
||||
assert rule.phase == "http_ratelimit"
|
||||
assert rule.action == "block"
|
||||
assert rule.expression == "(http.request.uri.path contains '/api/')"
|
||||
assert rule.description == "Rate limit API requests"
|
||||
assert rule.enabled is True
|
||||
|
||||
def test_cloudflare_firewall_rule_defaults(self):
|
||||
rule = CloudflareFirewallRule(
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
)
|
||||
|
||||
assert rule.id is None
|
||||
assert rule.zone_id == ZONE_ID
|
||||
assert rule.zone_name == ZONE_NAME
|
||||
assert rule.ruleset_id is None
|
||||
assert rule.phase is None
|
||||
assert rule.action is None
|
||||
assert rule.expression is None
|
||||
assert rule.description is None
|
||||
assert rule.enabled is True
|
||||
|
||||
def test_cloudflare_firewall_rule_disabled(self):
|
||||
rule = CloudflareFirewallRule(
|
||||
id="rule-disabled",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
phase="http_ratelimit",
|
||||
enabled=False,
|
||||
)
|
||||
|
||||
assert rule.enabled is False
|
||||
|
||||
def test_cloudflare_firewall_rule_custom_phase(self):
|
||||
rule = CloudflareFirewallRule(
|
||||
id="rule-custom",
|
||||
zone_id=ZONE_ID,
|
||||
zone_name=ZONE_NAME,
|
||||
phase="http_request_firewall_custom",
|
||||
action="challenge",
|
||||
expression="(cf.threat_score > 10)",
|
||||
)
|
||||
|
||||
assert rule.phase == "http_request_firewall_custom"
|
||||
assert rule.action == "challenge"
|
||||
+250
@@ -0,0 +1,250 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.cloudflare.services.zone.zone_service import (
|
||||
CloudflareFirewallRule,
|
||||
CloudflareZone,
|
||||
CloudflareZoneSettings,
|
||||
)
|
||||
from tests.providers.cloudflare.cloudflare_fixtures import (
|
||||
ZONE_ID,
|
||||
ZONE_NAME,
|
||||
set_mocked_cloudflare_provider,
|
||||
)
|
||||
|
||||
|
||||
class Test_zone_firewall_blocking_rules_configured:
|
||||
def test_no_zones(self):
|
||||
zone_client = mock.MagicMock
|
||||
zone_client.zones = {}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.zone.zone_firewall_blocking_rules_configured.zone_firewall_blocking_rules_configured.zone_client",
|
||||
new=zone_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.zone.zone_firewall_blocking_rules_configured.zone_firewall_blocking_rules_configured import (
|
||||
zone_firewall_blocking_rules_configured,
|
||||
)
|
||||
|
||||
check = zone_firewall_blocking_rules_configured()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
def test_zone_with_blocking_rules(self):
|
||||
zone_client = mock.MagicMock
|
||||
zone_client.zones = {
|
||||
ZONE_ID: CloudflareZone(
|
||||
id=ZONE_ID,
|
||||
name=ZONE_NAME,
|
||||
status="active",
|
||||
paused=False,
|
||||
settings=CloudflareZoneSettings(),
|
||||
firewall_rules=[
|
||||
CloudflareFirewallRule(
|
||||
id="rule-1",
|
||||
name="Block bad actors",
|
||||
action="block",
|
||||
enabled=True,
|
||||
),
|
||||
CloudflareFirewallRule(
|
||||
id="rule-2",
|
||||
name="Challenge suspicious",
|
||||
action="challenge",
|
||||
enabled=True,
|
||||
),
|
||||
],
|
||||
)
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.zone.zone_firewall_blocking_rules_configured.zone_firewall_blocking_rules_configured.zone_client",
|
||||
new=zone_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.zone.zone_firewall_blocking_rules_configured.zone_firewall_blocking_rules_configured import (
|
||||
zone_firewall_blocking_rules_configured,
|
||||
)
|
||||
|
||||
check = zone_firewall_blocking_rules_configured()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].resource_id == ZONE_ID
|
||||
assert result[0].resource_name == ZONE_NAME
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
"has firewall rules with blocking actions" in result[0].status_extended
|
||||
)
|
||||
assert "2 rule(s)" in result[0].status_extended
|
||||
|
||||
def test_zone_without_blocking_rules(self):
|
||||
zone_client = mock.MagicMock
|
||||
zone_client.zones = {
|
||||
ZONE_ID: CloudflareZone(
|
||||
id=ZONE_ID,
|
||||
name=ZONE_NAME,
|
||||
status="active",
|
||||
paused=False,
|
||||
settings=CloudflareZoneSettings(),
|
||||
firewall_rules=[
|
||||
CloudflareFirewallRule(
|
||||
id="rule-1",
|
||||
name="Log traffic",
|
||||
action="log",
|
||||
enabled=True,
|
||||
),
|
||||
],
|
||||
)
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.zone.zone_firewall_blocking_rules_configured.zone_firewall_blocking_rules_configured.zone_client",
|
||||
new=zone_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.zone.zone_firewall_blocking_rules_configured.zone_firewall_blocking_rules_configured import (
|
||||
zone_firewall_blocking_rules_configured,
|
||||
)
|
||||
|
||||
check = zone_firewall_blocking_rules_configured()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
"has no firewall rules with blocking actions"
|
||||
in result[0].status_extended
|
||||
)
|
||||
|
||||
def test_zone_with_no_firewall_rules(self):
|
||||
zone_client = mock.MagicMock
|
||||
zone_client.zones = {
|
||||
ZONE_ID: CloudflareZone(
|
||||
id=ZONE_ID,
|
||||
name=ZONE_NAME,
|
||||
status="active",
|
||||
paused=False,
|
||||
settings=CloudflareZoneSettings(),
|
||||
firewall_rules=[],
|
||||
)
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.zone.zone_firewall_blocking_rules_configured.zone_firewall_blocking_rules_configured.zone_client",
|
||||
new=zone_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.zone.zone_firewall_blocking_rules_configured.zone_firewall_blocking_rules_configured import (
|
||||
zone_firewall_blocking_rules_configured,
|
||||
)
|
||||
|
||||
check = zone_firewall_blocking_rules_configured()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
"has no firewall rules with blocking actions"
|
||||
in result[0].status_extended
|
||||
)
|
||||
|
||||
def test_zone_with_js_challenge_rule(self):
|
||||
zone_client = mock.MagicMock
|
||||
zone_client.zones = {
|
||||
ZONE_ID: CloudflareZone(
|
||||
id=ZONE_ID,
|
||||
name=ZONE_NAME,
|
||||
status="active",
|
||||
paused=False,
|
||||
settings=CloudflareZoneSettings(),
|
||||
firewall_rules=[
|
||||
CloudflareFirewallRule(
|
||||
id="rule-1",
|
||||
name="JS Challenge",
|
||||
action="js_challenge",
|
||||
enabled=True,
|
||||
),
|
||||
],
|
||||
)
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.zone.zone_firewall_blocking_rules_configured.zone_firewall_blocking_rules_configured.zone_client",
|
||||
new=zone_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.zone.zone_firewall_blocking_rules_configured.zone_firewall_blocking_rules_configured import (
|
||||
zone_firewall_blocking_rules_configured,
|
||||
)
|
||||
|
||||
check = zone_firewall_blocking_rules_configured()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
"has firewall rules with blocking actions" in result[0].status_extended
|
||||
)
|
||||
|
||||
def test_zone_with_managed_challenge_rule(self):
|
||||
zone_client = mock.MagicMock
|
||||
zone_client.zones = {
|
||||
ZONE_ID: CloudflareZone(
|
||||
id=ZONE_ID,
|
||||
name=ZONE_NAME,
|
||||
status="active",
|
||||
paused=False,
|
||||
settings=CloudflareZoneSettings(),
|
||||
firewall_rules=[
|
||||
CloudflareFirewallRule(
|
||||
id="rule-1",
|
||||
name="Managed Challenge",
|
||||
action="managed_challenge",
|
||||
enabled=True,
|
||||
),
|
||||
],
|
||||
)
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.zone.zone_firewall_blocking_rules_configured.zone_firewall_blocking_rules_configured.zone_client",
|
||||
new=zone_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.zone.zone_firewall_blocking_rules_configured.zone_firewall_blocking_rules_configured import (
|
||||
zone_firewall_blocking_rules_configured,
|
||||
)
|
||||
|
||||
check = zone_firewall_blocking_rules_configured()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert (
|
||||
"has firewall rules with blocking actions" in result[0].status_extended
|
||||
)
|
||||
+254
@@ -0,0 +1,254 @@
|
||||
from unittest import mock
|
||||
|
||||
from prowler.providers.cloudflare.services.zone.zone_service import (
|
||||
CloudflareWAFRuleset,
|
||||
CloudflareZone,
|
||||
CloudflareZoneSettings,
|
||||
)
|
||||
from tests.providers.cloudflare.cloudflare_fixtures import (
|
||||
ZONE_ID,
|
||||
ZONE_NAME,
|
||||
set_mocked_cloudflare_provider,
|
||||
)
|
||||
|
||||
|
||||
class Test_zone_waf_owasp_ruleset_enabled:
|
||||
def test_no_zones(self):
|
||||
zone_client = mock.MagicMock
|
||||
zone_client.zones = {}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.zone.zone_waf_owasp_ruleset_enabled.zone_waf_owasp_ruleset_enabled.zone_client",
|
||||
new=zone_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.zone.zone_waf_owasp_ruleset_enabled.zone_waf_owasp_ruleset_enabled import (
|
||||
zone_waf_owasp_ruleset_enabled,
|
||||
)
|
||||
|
||||
check = zone_waf_owasp_ruleset_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 0
|
||||
|
||||
def test_zone_with_owasp_ruleset_by_name(self):
|
||||
zone_client = mock.MagicMock
|
||||
zone_client.zones = {
|
||||
ZONE_ID: CloudflareZone(
|
||||
id=ZONE_ID,
|
||||
name=ZONE_NAME,
|
||||
status="active",
|
||||
paused=False,
|
||||
settings=CloudflareZoneSettings(),
|
||||
waf_rulesets=[
|
||||
CloudflareWAFRuleset(
|
||||
id="ruleset-1",
|
||||
name="Cloudflare OWASP Core Ruleset",
|
||||
kind="managed",
|
||||
phase="http_request_firewall_managed",
|
||||
enabled=True,
|
||||
),
|
||||
],
|
||||
)
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.zone.zone_waf_owasp_ruleset_enabled.zone_waf_owasp_ruleset_enabled.zone_client",
|
||||
new=zone_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.zone.zone_waf_owasp_ruleset_enabled.zone_waf_owasp_ruleset_enabled import (
|
||||
zone_waf_owasp_ruleset_enabled,
|
||||
)
|
||||
|
||||
check = zone_waf_owasp_ruleset_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].resource_id == ZONE_ID
|
||||
assert result[0].resource_name == ZONE_NAME
|
||||
assert result[0].status == "PASS"
|
||||
assert "has OWASP managed WAF ruleset enabled" in result[0].status_extended
|
||||
assert "Cloudflare OWASP Core Ruleset" in result[0].status_extended
|
||||
|
||||
def test_zone_with_managed_ruleset_without_owasp_name(self):
|
||||
"""Test that a managed ruleset without 'owasp' in name does NOT pass."""
|
||||
zone_client = mock.MagicMock
|
||||
zone_client.zones = {
|
||||
ZONE_ID: CloudflareZone(
|
||||
id=ZONE_ID,
|
||||
name=ZONE_NAME,
|
||||
status="active",
|
||||
paused=False,
|
||||
settings=CloudflareZoneSettings(),
|
||||
waf_rulesets=[
|
||||
CloudflareWAFRuleset(
|
||||
id="ruleset-1",
|
||||
name="Managed Rules",
|
||||
kind="managed",
|
||||
phase="http_request_firewall_managed",
|
||||
enabled=True,
|
||||
),
|
||||
],
|
||||
)
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.zone.zone_waf_owasp_ruleset_enabled.zone_waf_owasp_ruleset_enabled.zone_client",
|
||||
new=zone_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.zone.zone_waf_owasp_ruleset_enabled.zone_waf_owasp_ruleset_enabled import (
|
||||
zone_waf_owasp_ruleset_enabled,
|
||||
)
|
||||
|
||||
check = zone_waf_owasp_ruleset_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
"does not have OWASP managed WAF ruleset enabled"
|
||||
in result[0].status_extended
|
||||
)
|
||||
|
||||
def test_zone_without_owasp_ruleset(self):
|
||||
zone_client = mock.MagicMock
|
||||
zone_client.zones = {
|
||||
ZONE_ID: CloudflareZone(
|
||||
id=ZONE_ID,
|
||||
name=ZONE_NAME,
|
||||
status="active",
|
||||
paused=False,
|
||||
settings=CloudflareZoneSettings(),
|
||||
waf_rulesets=[
|
||||
CloudflareWAFRuleset(
|
||||
id="ruleset-1",
|
||||
name="Custom Rules",
|
||||
kind="custom",
|
||||
phase="http_request_firewall_custom",
|
||||
enabled=True,
|
||||
),
|
||||
],
|
||||
)
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.zone.zone_waf_owasp_ruleset_enabled.zone_waf_owasp_ruleset_enabled.zone_client",
|
||||
new=zone_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.zone.zone_waf_owasp_ruleset_enabled.zone_waf_owasp_ruleset_enabled import (
|
||||
zone_waf_owasp_ruleset_enabled,
|
||||
)
|
||||
|
||||
check = zone_waf_owasp_ruleset_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
"does not have OWASP managed WAF ruleset enabled"
|
||||
in result[0].status_extended
|
||||
)
|
||||
|
||||
def test_zone_with_no_waf_rulesets(self):
|
||||
zone_client = mock.MagicMock
|
||||
zone_client.zones = {
|
||||
ZONE_ID: CloudflareZone(
|
||||
id=ZONE_ID,
|
||||
name=ZONE_NAME,
|
||||
status="active",
|
||||
paused=False,
|
||||
settings=CloudflareZoneSettings(),
|
||||
waf_rulesets=[],
|
||||
)
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.zone.zone_waf_owasp_ruleset_enabled.zone_waf_owasp_ruleset_enabled.zone_client",
|
||||
new=zone_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.zone.zone_waf_owasp_ruleset_enabled.zone_waf_owasp_ruleset_enabled import (
|
||||
zone_waf_owasp_ruleset_enabled,
|
||||
)
|
||||
|
||||
check = zone_waf_owasp_ruleset_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "FAIL"
|
||||
assert (
|
||||
"does not have OWASP managed WAF ruleset enabled"
|
||||
in result[0].status_extended
|
||||
)
|
||||
|
||||
def test_zone_with_multiple_owasp_rulesets(self):
|
||||
zone_client = mock.MagicMock
|
||||
zone_client.zones = {
|
||||
ZONE_ID: CloudflareZone(
|
||||
id=ZONE_ID,
|
||||
name=ZONE_NAME,
|
||||
status="active",
|
||||
paused=False,
|
||||
settings=CloudflareZoneSettings(),
|
||||
waf_rulesets=[
|
||||
CloudflareWAFRuleset(
|
||||
id="ruleset-1",
|
||||
name="Cloudflare OWASP Core Ruleset",
|
||||
kind="managed",
|
||||
phase="http_request_firewall_managed",
|
||||
enabled=True,
|
||||
),
|
||||
CloudflareWAFRuleset(
|
||||
id="ruleset-2",
|
||||
name="Custom OWASP Rules",
|
||||
kind="managed",
|
||||
phase="http_request_firewall_managed",
|
||||
enabled=True,
|
||||
),
|
||||
],
|
||||
)
|
||||
}
|
||||
|
||||
with (
|
||||
mock.patch(
|
||||
"prowler.providers.common.provider.Provider.get_global_provider",
|
||||
return_value=set_mocked_cloudflare_provider(),
|
||||
),
|
||||
mock.patch(
|
||||
"prowler.providers.cloudflare.services.zone.zone_waf_owasp_ruleset_enabled.zone_waf_owasp_ruleset_enabled.zone_client",
|
||||
new=zone_client,
|
||||
),
|
||||
):
|
||||
from prowler.providers.cloudflare.services.zone.zone_waf_owasp_ruleset_enabled.zone_waf_owasp_ruleset_enabled import (
|
||||
zone_waf_owasp_ruleset_enabled,
|
||||
)
|
||||
|
||||
check = zone_waf_owasp_ruleset_enabled()
|
||||
result = check.execute()
|
||||
assert len(result) == 1
|
||||
assert result[0].status == "PASS"
|
||||
assert "Cloudflare OWASP Core Ruleset" in result[0].status_extended
|
||||
assert "Custom OWASP Rules" in result[0].status_extended
|
||||
Reference in New Issue
Block a user