feat(azure): add Application Gateway WAF and HTTP/2 checks (#13160)

Co-authored-by: pedrooot <pedromarting3@gmail.com>
This commit is contained in:
Babafemi Bulugbeandpedrooot authored and GitHub committed 2026-10-29 10:49:38 +01:00
1 parent ca6ed48664
commit 8e83de67bc
13 files changed
+1395 -6

No files matched your search

@@ -0,0 +1 @@
`network_application_gateway_waf_enabled` and `network_application_gateway_http2_enabled` checks for Azure provider, with Application Gateway and WAF policy collection in the network service, mapped to CIS 5.0 and 6.0
+6 -2
View File
@@ -1954,7 +1954,9 @@
{
"Id": "7.10",
"Description": "Ensure Azure Web Application Firewall (WAF) is enabled on Azure Application Gateway",
"Checks": [],
"Checks": [
"network_application_gateway_waf_enabled"
],
"Attributes": [
{
"Section": "7 Networking Services",
@@ -2016,7 +2018,9 @@
{
"Id": "7.13",
"Description": "Ensure 'HTTP2' is set to 'Enabled' on Azure Application Gateway",
"Checks": [],
"Checks": [
"network_application_gateway_http2_enabled"
],
"Attributes": [
{
"Section": "7 Networking Services",
+6 -2
View File
@@ -1358,7 +1358,9 @@
{
"Id": "7.10",
"Description": "Ensure Azure Web Application Firewall (WAF) is Enabled on Azure Application Gateway",
"Checks": [],
"Checks": [
"network_application_gateway_waf_enabled"
],
"Attributes": [
{
"Section": "7 Networking Services",
@@ -1420,7 +1422,9 @@
{
"Id": "7.13",
"Description": "Ensure 'HTTP2' is Set to 'Enabled' on Azure Application Gateway",
"Checks": [],
"Checks": [
"network_application_gateway_http2_enabled"
],
"Attributes": [
{
"Section": "7 Networking Services",
@@ -0,0 +1,34 @@
{
"Provider": "azure",
"CheckID": "network_application_gateway_http2_enabled",
"CheckTitle": "Application Gateway has HTTP/2 enabled",
"CheckType": [],
"ServiceName": "network",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "low",
"ResourceType": "microsoft.network/applicationgateways",
"ResourceGroup": "network",
"Description": "**Azure Application Gateways** are evaluated for **HTTP/2** support on client connections (`enableHttp2`). The gateway offers HTTP/2 only over TLS on HTTPS listeners, and traffic to backend pools always uses HTTP/1.1.",
"Risk": "Clients fall back to **HTTP/1.1**, opening more TLS connections and losing HTTP/2 multiplexing and header compression. The direct security impact is low, so treat this as protocol hygiene and performance rather than an exposure.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://learn.microsoft.com/en-us/azure/application-gateway/configuration-listeners"
],
"Remediation": {
"Code": {
"CLI": "az network application-gateway update --resource-group <resource_group> --name <gateway_name> --http2 Enabled",
"NativeIaC": "```bicep\nresource appGw 'Microsoft.Network/applicationGateways@2024-05-01' = {\n name: '<example_resource_name>'\n location: '<example_location>'\n properties: {\n enableHttp2: true // Critical: enables HTTP/2 for clients\n // ...existing gateway configuration\n }\n}\n```",
"Other": "1. In the Azure portal, open the Application Gateway\n2. Under Settings, select Configuration\n3. Set HTTP2 to Enabled and click Save",
"Terraform": "```hcl\nresource \"azurerm_application_gateway\" \"example_resource\" {\n # ...existing gateway configuration\n http2_enabled = true # Critical: enables HTTP/2 for clients\n}\n```"
},
"Recommendation": {
"Text": "Enable **HTTP/2** on every gateway so clients that support it negotiate the modern protocol over TLS. HTTP/1.1 clients keep working because the gateway supports both.",
"Url": "https://hub.prowler.com/check/network_application_gateway_http2_enabled"
}
},
"Categories": [],
"DependsOn": [],
"RelatedTo": [],
"Notes": "The Azure API leaves HTTP/2 disabled unless it is set, while the Azure portal enables it by default."
}
@@ -0,0 +1,39 @@
from prowler.lib.check.models import Check, Check_Report_Azure
from prowler.providers.azure.services.network.network_client import network_client
class network_application_gateway_http2_enabled(Check):
"""Application Gateway has HTTP/2 enabled for client connections."""
def execute(self) -> list[Check_Report_Azure]:
"""Evaluate the HTTP/2 setting of every Application Gateway.
Returns:
One finding per Application Gateway, plus one MANUAL finding for
each subscription whose gateways could not be listed completely.
"""
findings = []
for subscription_id, gateways in network_client.application_gateways.items():
subscription_name = network_client.subscriptions.get(
subscription_id, subscription_id
)
if subscription_id in network_client.application_gateways_lookup_failed:
report = Check_Report_Azure(metadata=self.metadata(), resource={})
report.subscription = subscription_id
report.resource_name = subscription_id
report.resource_id = f"/subscriptions/{subscription_id}"
report.status = "MANUAL"
report.status_extended = f"Cannot evaluate HTTP/2 on the Application Gateways in subscription {subscription_name} ({subscription_id}): they could not be listed completely. Check the scan log for the error, and verify that Microsoft.Network/applicationGateways/read (included in Reader) is granted to the scanning identity."
findings.append(report)
for gateway in gateways.values():
report = Check_Report_Azure(metadata=self.metadata(), resource=gateway)
report.subscription = subscription_id
prefix = f"Application Gateway {gateway.name} in subscription {subscription_name} ({subscription_id})"
if gateway.http2_enabled:
report.status = "PASS"
report.status_extended = f"{prefix} has HTTP/2 enabled."
else:
report.status = "FAIL"
report.status_extended = f"{prefix} does not have HTTP/2 enabled."
findings.append(report)
return findings
@@ -0,0 +1,37 @@
{
"Provider": "azure",
"CheckID": "network_application_gateway_waf_enabled",
"CheckTitle": "Application Gateway has an enabled Web Application Firewall policy associated",
"CheckType": [],
"ServiceName": "network",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "microsoft.network/applicationgateways",
"ResourceGroup": "network",
"Description": "**Azure Application Gateways** are evaluated for an enabled **Web Application Firewall (WAF) policy** associated at the gateway level, which requires the `WAF_v2` tier. The legacy WAF configuration does not count as a policy.",
"Risk": "Without a WAF, web applications behind the gateway receive **SQL injection**, **cross-site scripting** and other common exploits unfiltered, risking **data theft**, **account takeover** and **defacement** of internet-facing services.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://learn.microsoft.com/en-us/azure/web-application-firewall/ag/ag-overview",
"https://learn.microsoft.com/en-us/azure/web-application-firewall/ag/upgrade-ag-waf-policy"
],
"Remediation": {
"Code": {
"CLI": "az network application-gateway update --resource-group <resource_group> --name <gateway_name> --set sku.name=WAF_v2 sku.tier=WAF_v2 firewallPolicy.id=<waf_policy_id> forceFirewallPolicyAssociation=true",
"NativeIaC": "```bicep\nresource appGw 'Microsoft.Network/applicationGateways@2024-05-01' = {\n name: '<example_resource_name>'\n location: '<example_location>'\n properties: {\n sku: { name: 'WAF_v2', tier: 'WAF_v2' } // Critical: WAF tier\n firewallPolicy: { id: '<example_resource_id>' } // Critical: associates the WAF policy\n forceFirewallPolicyAssociation: true // Replaces a legacy WAF configuration with the policy\n // ...existing gateway configuration, without webApplicationFirewallConfiguration\n }\n}\n```",
"Other": "1. In the Azure portal, open the Application Gateway\n2. If the tier is Standard V2, select Configuration under Settings, set Tier to WAF V2 and click Save. A V1 gateway must first be migrated to V2, and a Basic gateway cannot change tier and must be redeployed on WAF V2\n3. Select Web Application Firewall. If the gateway uses the legacy WAF configuration, select Upgrade from WAF configuration, name the new policy and click Upgrade. This copies the legacy exclusions and rule settings into the policy, whereas the CLI and IaC examples force the association and apply the policy's own settings instead\n4. Otherwise associate a new or existing WAF policy and click Save\n5. If the associated policy is disabled, enable it, for example with `az network application-gateway waf-policy policy-setting update --resource-group <waf_policy_resource_group> --policy-name <waf_policy_name> --state Enabled --request-body-check true`. Keep `--request-body-check true`: without it, this command turns request body inspection off",
"Terraform": "```hcl\nresource \"azurerm_application_gateway\" \"example_resource\" {\n # ...existing gateway configuration, without a waf_configuration block\n sku {\n name = \"WAF_v2\" # Critical: WAF tier\n tier = \"WAF_v2\"\n }\n firewall_policy_id = azurerm_web_application_firewall_policy.example_resource.id # Critical: associates the WAF policy\n force_firewall_policy_association = true # Replaces a legacy WAF configuration with the policy\n}\n```"
},
"Recommendation": {
"Text": "Put an **enabled WAF policy in Prevention mode** in front of every internet-facing gateway, using the latest Default Rule Set. Migrate gateways that still use the legacy WAF configuration to a policy before it retires.",
"Url": "https://hub.prowler.com/check/network_application_gateway_waf_enabled"
}
},
"Categories": [
"internet-exposed"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": "Gateways using the legacy WAF configuration are reported as FAIL because Microsoft retires it on 15 March 2027 and newer WAF features, such as bot protection, require a policy. Only the gateway-level policy is evaluated, as in the CIS audit; policies attached to listeners or path rules are not. The policy state comes from the WAF policies collected in the same scan, or is read from the policy itself when it sits outside the scanned resource groups, and a disabled policy is reported as FAIL. The policy mode is not evaluated, so an enabled policy in Detection mode, which only logs requests, is reported as PASS."
}
@@ -0,0 +1,85 @@
from prowler.lib.check.models import Check, Check_Report_Azure
from prowler.providers.azure.services.network.network_client import network_client
class network_application_gateway_waf_enabled(Check):
"""Application Gateway has an enabled Web Application Firewall policy associated."""
def execute(self) -> list[Check_Report_Azure]:
"""Evaluate the WAF policy association of every Application Gateway.
Returns:
One finding per Application Gateway, plus one MANUAL finding for
each subscription whose gateways, or whose WAF policies with a
gateway's policy state unknown, could not be listed or read
completely. A gateway covered by that finding gets none of its
own. Any other gateway whose policy state cannot be determined is
MANUAL.
"""
findings = []
for subscription_id, gateways in network_client.application_gateways.items():
subscription_name = network_client.subscriptions.get(
subscription_id, subscription_id
)
gateways_unlisted = (
subscription_id in network_client.application_gateways_lookup_failed
)
# A policy state left unknown by a failed listing or read is reported
# once for the subscription; a missing policy keeps its own finding.
policies_unlisted = (
subscription_id in network_client.waf_policies_lookup_failed
)
unknown_policy_state = {
gateway.id
for gateway in gateways.values()
if gateway.firewall_policy_id
and gateway.firewall_policy_enabled is None
and (
policies_unlisted
or gateway.firewall_policy_id.casefold()
in network_client.waf_policy_reads_failed
)
}
policies_unreadable = bool(unknown_policy_state)
if gateways_unlisted or policies_unreadable:
report = Check_Report_Azure(metadata=self.metadata(), resource={})
report.subscription = subscription_id
report.resource_name = subscription_id
report.resource_id = f"/subscriptions/{subscription_id}"
report.status = "MANUAL"
scope = f"Cannot evaluate the WAF policy association of the Application Gateways in subscription {subscription_name} ({subscription_id})"
if gateways_unlisted and policies_unreadable:
report.status_extended = f"{scope}: they could not be listed completely, and the WAF policies associated with them could not be listed or read completely. Check the scan log for the errors, and verify that Microsoft.Network/applicationGateways/read and Microsoft.Network/ApplicationGatewayWebApplicationFirewallPolicies/read (both included in Reader) are granted to the scanning identity."
elif gateways_unlisted:
report.status_extended = f"{scope}: they could not be listed completely. Check the scan log for the error, and verify that Microsoft.Network/applicationGateways/read (included in Reader) is granted to the scanning identity."
else:
report.status_extended = f"{scope}: the WAF policies associated with them could not be listed or read completely. Check the scan log for the error, and verify that Microsoft.Network/ApplicationGatewayWebApplicationFirewallPolicies/read (included in Reader) is granted to the scanning identity."
findings.append(report)
for gateway in gateways.values():
if policies_unreadable and gateway.id in unknown_policy_state:
continue
report = Check_Report_Azure(metadata=self.metadata(), resource=gateway)
report.subscription = subscription_id
prefix = f"Application Gateway {gateway.name} in subscription {subscription_name} ({subscription_id})"
if gateway.firewall_policy_id:
if gateway.firewall_policy_enabled:
report.status = "PASS"
report.status_extended = (
f"{prefix} has an enabled WAF policy associated."
)
elif gateway.firewall_policy_enabled is False:
report.status = "FAIL"
report.status_extended = f"{prefix} has a WAF policy associated, but the policy is disabled."
else:
report.status = "MANUAL"
report.status_extended = f"{prefix} has a WAF policy associated, but the state of that policy could not be determined. Check the scan log for the error."
elif gateway.waf_configuration_enabled:
report.status = "FAIL"
report.status_extended = f"{prefix} uses the legacy WAF configuration instead of a WAF policy."
else:
report.status = "FAIL"
report.status_extended = (
f"{prefix} has no gateway-level WAF policy associated."
)
findings.append(report)
return findings
@@ -1,14 +1,26 @@
import re
from dataclasses import dataclass
from typing import List, Optional
from typing import Callable, Iterable, List, Optional
from azure.core.exceptions import ResourceNotFoundError
from azure.mgmt.network import NetworkManagementClient
from azure.mgmt.network import models as network_models
from pydantic.v1 import BaseModel
from prowler.lib.logger import logger
from prowler.providers.azure.azure_provider import AzureProvider
from prowler.providers.azure.lib.service.service import AzureService
WAF_POLICY_ID = re.compile(
r"^/subscriptions/(?P<subscription>[^/]+)/resourceGroups/(?P<resource_group>[^/]+)"
r"/providers/Microsoft\.Network/ApplicationGatewayWebApplicationFirewallPolicies"
r"/(?P<name>[^/]+)$",
re.IGNORECASE,
)
# Whether a WAF policy in each state, matched without regard to case, is enabled.
WAF_POLICY_STATES = {"enabled": True, "disabled": False}
class Network(AzureService):
def __init__(self, provider: AzureProvider):
@@ -18,6 +30,14 @@ class Network(AzureService):
self.network_watchers = self._get_network_watchers()
self.public_ip_addresses = self._get_public_ip_addresses()
self.virtual_networks = self._get_virtual_networks()
# Subscriptions whose WAF policies could not be listed completely.
self.waf_policies_lookup_failed: set[str] = set()
self.waf_policies = self._get_waf_policies()
# Casefolded IDs of WAF policies outside that listing that could not be read.
self.waf_policy_reads_failed: set[str] = set()
# Subscriptions whose Application Gateways could not be listed completely.
self.application_gateways_lookup_failed: set[str] = set()
self.application_gateways = self._get_application_gateways()
def _get_security_groups(self):
logger.info("Network - Getting Network Security Groups...")
@@ -258,6 +278,109 @@ class Network(AzureService):
)
return virtual_networks
def _get_waf_policies(self) -> dict:
"""Get the Application Gateway WAF policies of every subscription.
A subscription whose WAF policies cannot be listed or mapped completely
is added to waf_policies_lookup_failed. A policy that cannot be mapped
is logged and skipped, so the others are still collected.
Returns:
A dictionary keyed by subscription ID whose values map each WAF
policy ID to a WafPolicy model.
"""
logger.info("Network - Getting Application Gateway WAF policies...")
waf_policies = {}
for subscription_id, client in self.clients.items():
try:
waf_policies[subscription_id] = {}
waf_policies_list = self.list_with_rg_scope(
subscription_id,
client.web_application_firewall_policies.list_all,
mark_failed_lookups(
client.web_application_firewall_policies.list,
self.waf_policies_lookup_failed,
subscription_id,
),
)
for waf_policy in waf_policies_list:
try:
policy = build_waf_policy(waf_policy)
waf_policies[subscription_id][policy.id] = policy
except Exception as error:
self.waf_policies_lookup_failed.add(subscription_id)
logger.error(
f"Subscription ID: {subscription_id} -- WAF policy: {getattr(waf_policy, 'id', None)} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
self.waf_policies_lookup_failed.add(subscription_id)
logger.error(
f"Subscription ID: {subscription_id} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return waf_policies
def _get_application_gateways(self) -> dict:
"""Get the Application Gateways of every subscription.
The state of each gateway's WAF policy comes from the policies that
_get_waf_policies collected, so that method must run first, and a
policy that was not listed is read once by its ID. A subscription whose
gateways cannot be listed or mapped completely is added to
application_gateways_lookup_failed. A gateway that cannot be mapped is
logged and skipped, so the others are still collected.
Returns:
A dictionary keyed by subscription ID whose values map each
Application Gateway ID to an ApplicationGateway model.
"""
logger.info("Network - Getting Application Gateways...")
application_gateways = {}
for subscription_id, client in self.clients.items():
try:
application_gateways[subscription_id] = {}
application_gateways_list = self.list_with_rg_scope(
subscription_id,
client.application_gateways.list_all,
mark_failed_lookups(
client.application_gateways.list,
self.application_gateways_lookup_failed,
subscription_id,
),
)
listed_policies = {
policy_id.casefold(): policy
for policy_id, policy in self.waf_policies.get(
subscription_id, {}
).items()
}
policy_states = {}
for application_gateway in application_gateways_list:
try:
gateway = build_application_gateway(application_gateway)
if gateway.firewall_policy_id:
key = gateway.firewall_policy_id.casefold()
if key not in policy_states:
policy_states[key] = get_waf_policy_enabled(
client,
subscription_id,
gateway.firewall_policy_id,
listed_policies,
self.waf_policy_reads_failed,
)
gateway.firewall_policy_enabled = policy_states[key]
application_gateways[subscription_id][gateway.id] = gateway
except Exception as error:
self.application_gateways_lookup_failed.add(subscription_id)
logger.error(
f"Subscription ID: {subscription_id} -- Application Gateway: {getattr(application_gateway, 'id', None)} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
self.application_gateways_lookup_failed.add(subscription_id)
logger.error(
f"Subscription ID: {subscription_id} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return application_gateways
@dataclass
class BastionHost:
@@ -336,3 +459,192 @@ class VirtualNetwork:
def __post_init__(self):
if self.subnets is None:
self.subnets = []
class ApplicationGateway(BaseModel):
"""Application Gateway settings used by the network checks."""
id: str
name: str
location: str
firewall_policy_id: Optional[str] = None
firewall_policy_enabled: Optional[bool] = None
waf_configuration_enabled: bool = False
http2_enabled: bool = False
class WafPolicy(BaseModel):
"""Application Gateway WAF policy settings used by the network checks."""
id: str
name: str
location: str
state: Optional[str] = None
def mark_failed_lookups(
list_by_rg_fn: Callable[..., Iterable[object]],
lookup_failed: set[str],
subscription_id: str,
) -> Callable[..., List[object]]:
"""Wrap a per-resource-group list call so that a failure marks the subscription.
list_with_rg_scope logs and skips a resource group whose list call fails,
so the subscription is recorded here before the error reaches it. A
resource group that no longer exists (not found) is not a failed lookup.
Args:
list_by_rg_fn: Azure SDK list operation for one resource group.
lookup_failed: Set of subscriptions whose listing was incomplete.
subscription_id: Subscription the list call belongs to.
Returns:
A callable with the same keyword arguments as list_by_rg_fn.
"""
def list_by_resource_group(**kwargs) -> List[object]:
try:
return list(list_by_rg_fn(**kwargs))
except ResourceNotFoundError:
raise
except Exception:
lookup_failed.add(subscription_id)
raise
return list_by_resource_group
def build_application_gateway(
application_gateway: network_models.ApplicationGateway,
) -> ApplicationGateway:
"""Convert an Azure SDK Application Gateway into an ApplicationGateway model.
The state of the associated WAF policy is not part of the gateway, so
firewall_policy_enabled is set afterwards from get_waf_policy_enabled.
Args:
application_gateway: ApplicationGateway from the Azure SDK.
Returns:
The ApplicationGateway model.
"""
waf_configuration = getattr(
application_gateway, "web_application_firewall_configuration", None
)
return ApplicationGateway(
id=application_gateway.id,
name=application_gateway.name,
location=application_gateway.location,
firewall_policy_id=getattr(
getattr(application_gateway, "firewall_policy", None), "id", None
),
waf_configuration_enabled=bool(getattr(waf_configuration, "enabled", False)),
http2_enabled=bool(getattr(application_gateway, "enable_http2", False)),
)
def build_waf_policy(
waf_policy: network_models.WebApplicationFirewallPolicy,
) -> WafPolicy:
"""Convert an Azure SDK WAF policy into a WafPolicy model.
Args:
waf_policy: WebApplicationFirewallPolicy from the Azure SDK.
Returns:
The WafPolicy model.
"""
settings = getattr(waf_policy, "policy_settings", None)
return WafPolicy(
id=waf_policy.id,
name=waf_policy.name,
location=waf_policy.location,
state=getattr(settings, "state", None),
)
def read_waf_policy(
client: NetworkManagementClient,
subscription_id: str,
policy_id: str,
reads_failed: set[str],
) -> Optional[WafPolicy]:
"""Read one Application Gateway WAF policy by its resource ID.
A read that fails for any reason other than the policy not existing adds
the casefolded policy ID to reads_failed. Every failure is logged.
Args:
client: NetworkManagementClient of the gateway's subscription.
subscription_id: Subscription of the gateway.
policy_id: Resource ID of the WAF policy.
reads_failed: Casefolded IDs of WAF policies that could not be read.
Returns:
The WafPolicy model, or None when the ID is not a WAF policy ID in the
subscription or the policy cannot be read.
"""
match = WAF_POLICY_ID.match(policy_id)
if (
not match
or match.group("subscription").casefold() != subscription_id.casefold()
):
logger.warning(
f"Subscription ID: {subscription_id} -- WAF policy: {policy_id} -- not an Application Gateway WAF policy ID in this subscription, so the policy was not read."
)
return None
try:
return build_waf_policy(
client.web_application_firewall_policies.get(
resource_group_name=match.group("resource_group"),
policy_name=match.group("name"),
)
)
except ResourceNotFoundError as error:
logger.warning(
f"Subscription ID: {subscription_id} -- WAF policy: {policy_id} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
reads_failed.add(policy_id.casefold())
logger.error(
f"Subscription ID: {subscription_id} -- WAF policy: {policy_id} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return None
def get_waf_policy_enabled(
client: NetworkManagementClient,
subscription_id: str,
policy_id: str,
listed_policies: dict,
reads_failed: set[str],
) -> Optional[bool]:
"""Return whether the WAF policy associated with a gateway is enabled.
The policy is looked up among the WAF policies listed for the
subscription, and read by its ID only when it was not listed, such as a
policy outside the scanned resource groups.
Args:
client: NetworkManagementClient of the gateway's subscription.
subscription_id: Subscription of the gateway.
policy_id: Resource ID of the associated WAF policy.
listed_policies: WafPolicy models listed for the subscription, keyed by
casefolded policy ID.
reads_failed: Casefolded IDs of WAF policies that could not be read.
Returns:
True when the policy is enabled, False when it is disabled, and None
when its state cannot be determined.
"""
policy = listed_policies.get(policy_id.casefold())
if policy is None:
policy = read_waf_policy(client, subscription_id, policy_id, reads_failed)
if policy is None:
return None
enabled = WAF_POLICY_STATES.get((policy.state or "").casefold())
if enabled is None:
logger.warning(
f"Subscription ID: {subscription_id} -- WAF policy: {policy_id} -- unrecognised policy state: {policy.state}"
)
return enabled
@@ -0,0 +1,101 @@
from unittest import mock
from prowler.providers.azure.services.network.network_service import (
ApplicationGateway,
)
from tests.providers.azure.azure_fixtures import (
AZURE_SUBSCRIPTION_DISPLAY,
AZURE_SUBSCRIPTION_ID,
AZURE_SUBSCRIPTION_NAME,
RESOURCE_GROUP,
set_mocked_azure_provider,
)
CHECK = "network_application_gateway_http2_enabled"
CHECK_PATH = f"prowler.providers.azure.services.network.{CHECK}.{CHECK}"
GATEWAY_ID = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourceGroups/{RESOURCE_GROUP}/providers/Microsoft.Network/applicationGateways/agw1"
PREFIX = f"Application Gateway agw1 in subscription {AZURE_SUBSCRIPTION_DISPLAY}"
def run_check(gateways, lookup_failed=()):
network_client = mock.MagicMock()
network_client.subscriptions = {AZURE_SUBSCRIPTION_ID: AZURE_SUBSCRIPTION_NAME}
network_client.application_gateways = gateways
network_client.application_gateways_lookup_failed = set(lookup_failed)
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_azure_provider(),
),
mock.patch(f"{CHECK_PATH}.network_client", new=network_client),
):
from prowler.providers.azure.services.network.network_application_gateway_http2_enabled.network_application_gateway_http2_enabled import (
network_application_gateway_http2_enabled,
)
return network_application_gateway_http2_enabled().execute()
def single(http2_enabled: bool):
gateway = ApplicationGateway(
id=GATEWAY_ID, name="agw1", location="westeurope", http2_enabled=http2_enabled
)
return run_check({AZURE_SUBSCRIPTION_ID: {GATEWAY_ID: gateway}})
UNLISTED = f"Cannot evaluate HTTP/2 on the Application Gateways in subscription {AZURE_SUBSCRIPTION_DISPLAY}: they could not be listed completely. Check the scan log for the error, and verify that Microsoft.Network/applicationGateways/read (included in Reader) is granted to the scanning identity."
def assert_gateway_finding(finding, status, status_extended):
assert finding.status == status
assert finding.status_extended == status_extended
assert finding.subscription == AZURE_SUBSCRIPTION_ID
assert finding.resource_id == GATEWAY_ID
assert finding.resource_name == "agw1"
assert finding.location == "westeurope"
def assert_subscription_finding(finding):
assert finding.status == "MANUAL"
assert finding.status_extended == UNLISTED
assert finding.subscription == AZURE_SUBSCRIPTION_ID
assert finding.resource_id == f"/subscriptions/{AZURE_SUBSCRIPTION_ID}"
assert finding.resource_name == AZURE_SUBSCRIPTION_ID
assert finding.location == "global"
class Test_network_application_gateway_http2_enabled:
def test_no_gateways(self):
assert run_check({}) == []
assert run_check({AZURE_SUBSCRIPTION_ID: {}}) == []
def test_http2_enabled(self):
result = single(True)
assert len(result) == 1
assert_gateway_finding(result[0], "PASS", f"{PREFIX} has HTTP/2 enabled.")
def test_http2_disabled(self):
result = single(False)
assert len(result) == 1
assert_gateway_finding(
result[0], "FAIL", f"{PREFIX} does not have HTTP/2 enabled."
)
def test_gateways_lookup_failed_is_manual(self):
result = run_check(
{AZURE_SUBSCRIPTION_ID: {}}, lookup_failed={AZURE_SUBSCRIPTION_ID}
)
assert len(result) == 1
assert_subscription_finding(result[0])
def test_partial_lookup_still_reports_listed_gateways(self):
gateway = ApplicationGateway(
id=GATEWAY_ID, name="agw1", location="westeurope", http2_enabled=True
)
result = run_check(
{AZURE_SUBSCRIPTION_ID: {GATEWAY_ID: gateway}},
lookup_failed={AZURE_SUBSCRIPTION_ID},
)
assert len(result) == 2
assert_subscription_finding(result[0])
assert_gateway_finding(result[1], "PASS", f"{PREFIX} has HTTP/2 enabled.")
@@ -0,0 +1,260 @@
from unittest import mock
from prowler.providers.azure.services.network.network_service import (
ApplicationGateway,
)
from tests.providers.azure.azure_fixtures import (
AZURE_SUBSCRIPTION_DISPLAY,
AZURE_SUBSCRIPTION_ID,
AZURE_SUBSCRIPTION_NAME,
RESOURCE_GROUP,
set_mocked_azure_provider,
)
CHECK = "network_application_gateway_waf_enabled"
CHECK_PATH = f"prowler.providers.azure.services.network.{CHECK}.{CHECK}"
GATEWAY_ID = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourceGroups/{RESOURCE_GROUP}/providers/Microsoft.Network/applicationGateways/agw1"
POLICY_ID = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourceGroups/{RESOURCE_GROUP}/providers/Microsoft.Network/ApplicationGatewayWebApplicationFirewallPolicies/waf1"
PREFIX = f"Application Gateway agw1 in subscription {AZURE_SUBSCRIPTION_DISPLAY}"
GATEWAYS_UNLISTED = f"Cannot evaluate the WAF policy association of the Application Gateways in subscription {AZURE_SUBSCRIPTION_DISPLAY}: they could not be listed completely. Check the scan log for the error, and verify that Microsoft.Network/applicationGateways/read (included in Reader) is granted to the scanning identity."
POLICIES_UNLISTED = f"Cannot evaluate the WAF policy association of the Application Gateways in subscription {AZURE_SUBSCRIPTION_DISPLAY}: the WAF policies associated with them could not be listed or read completely. Check the scan log for the error, and verify that Microsoft.Network/ApplicationGatewayWebApplicationFirewallPolicies/read (included in Reader) is granted to the scanning identity."
BOTH_UNLISTED = f"Cannot evaluate the WAF policy association of the Application Gateways in subscription {AZURE_SUBSCRIPTION_DISPLAY}: they could not be listed completely, and the WAF policies associated with them could not be listed or read completely. Check the scan log for the errors, and verify that Microsoft.Network/applicationGateways/read and Microsoft.Network/ApplicationGatewayWebApplicationFirewallPolicies/read (both included in Reader) are granted to the scanning identity."
def build_gateway(**settings) -> ApplicationGateway:
return ApplicationGateway(
id=GATEWAY_ID, name="agw1", location="westeurope", **settings
)
def run_check(
gateways,
gateways_lookup_failed=(),
policies_lookup_failed=(),
policy_reads_failed=(),
):
network_client = mock.MagicMock()
network_client.subscriptions = {AZURE_SUBSCRIPTION_ID: AZURE_SUBSCRIPTION_NAME}
network_client.application_gateways = gateways
network_client.application_gateways_lookup_failed = set(gateways_lookup_failed)
network_client.waf_policies_lookup_failed = set(policies_lookup_failed)
network_client.waf_policy_reads_failed = set(policy_reads_failed)
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_azure_provider(),
),
mock.patch(f"{CHECK_PATH}.network_client", new=network_client),
):
from prowler.providers.azure.services.network.network_application_gateway_waf_enabled.network_application_gateway_waf_enabled import (
network_application_gateway_waf_enabled,
)
return network_application_gateway_waf_enabled().execute()
def single(gateway):
return run_check({AZURE_SUBSCRIPTION_ID: {gateway.id: gateway}})
def assert_gateway_finding(finding, status, status_extended):
assert finding.status == status
assert finding.status_extended == status_extended
assert finding.subscription == AZURE_SUBSCRIPTION_ID
assert finding.resource_id == GATEWAY_ID
assert finding.resource_name == "agw1"
assert finding.location == "westeurope"
def assert_subscription_finding(finding, status_extended):
assert finding.status == "MANUAL"
assert finding.status_extended == status_extended
assert finding.subscription == AZURE_SUBSCRIPTION_ID
assert finding.resource_id == f"/subscriptions/{AZURE_SUBSCRIPTION_ID}"
assert finding.resource_name == AZURE_SUBSCRIPTION_ID
assert finding.location == "global"
class Test_network_application_gateway_waf_enabled:
def test_no_gateways(self):
assert run_check({}) == []
assert run_check({AZURE_SUBSCRIPTION_ID: {}}) == []
def test_enabled_waf_policy_associated(self):
result = single(
build_gateway(firewall_policy_id=POLICY_ID, firewall_policy_enabled=True)
)
assert len(result) == 1
assert_gateway_finding(
result[0], "PASS", f"{PREFIX} has an enabled WAF policy associated."
)
def test_disabled_waf_policy_associated(self):
result = single(
build_gateway(firewall_policy_id=POLICY_ID, firewall_policy_enabled=False)
)
assert len(result) == 1
assert_gateway_finding(
result[0],
"FAIL",
f"{PREFIX} has a WAF policy associated, but the policy is disabled.",
)
def test_waf_policy_state_unknown(self):
result = single(build_gateway(firewall_policy_id=POLICY_ID))
assert len(result) == 1
assert_gateway_finding(
result[0],
"MANUAL",
f"{PREFIX} has a WAF policy associated, but the state of that policy could not be determined. Check the scan log for the error.",
)
def test_legacy_waf_configuration(self):
result = single(build_gateway(waf_configuration_enabled=True))
assert len(result) == 1
assert_gateway_finding(
result[0],
"FAIL",
f"{PREFIX} uses the legacy WAF configuration instead of a WAF policy.",
)
def test_no_waf(self):
result = single(build_gateway())
assert len(result) == 1
assert_gateway_finding(
result[0], "FAIL", f"{PREFIX} has no gateway-level WAF policy associated."
)
def test_gateways_lookup_failed_is_manual(self):
result = run_check(
{AZURE_SUBSCRIPTION_ID: {}}, gateways_lookup_failed={AZURE_SUBSCRIPTION_ID}
)
assert len(result) == 1
assert_subscription_finding(result[0], GATEWAYS_UNLISTED)
def test_partial_gateways_lookup_still_reports_listed_gateways(self):
gateway = build_gateway(
firewall_policy_id=POLICY_ID, firewall_policy_enabled=True
)
result = run_check(
{AZURE_SUBSCRIPTION_ID: {GATEWAY_ID: gateway}},
gateways_lookup_failed={AZURE_SUBSCRIPTION_ID},
)
assert len(result) == 2
assert_subscription_finding(result[0], GATEWAYS_UNLISTED)
assert_gateway_finding(
result[1], "PASS", f"{PREFIX} has an enabled WAF policy associated."
)
def test_policies_lookup_failed_reports_unknown_states_once(self):
other_id = GATEWAY_ID.replace("agw1", "agw2")
gateways = {
GATEWAY_ID: build_gateway(firewall_policy_id=POLICY_ID),
other_id: ApplicationGateway(
id=other_id,
name="agw2",
location="westeurope",
firewall_policy_id=POLICY_ID,
),
}
result = run_check(
{AZURE_SUBSCRIPTION_ID: gateways},
policies_lookup_failed={AZURE_SUBSCRIPTION_ID},
)
assert len(result) == 1
assert_subscription_finding(result[0], POLICIES_UNLISTED)
def test_policies_lookup_failed_keeps_gateways_with_known_states(self):
other_id = GATEWAY_ID.replace("agw1", "agw2")
gateways = {
GATEWAY_ID: build_gateway(
firewall_policy_id=POLICY_ID, firewall_policy_enabled=False
),
other_id: ApplicationGateway(
id=other_id,
name="agw2",
location="westeurope",
firewall_policy_id=POLICY_ID.replace("waf1", "waf2"),
),
}
result = run_check(
{AZURE_SUBSCRIPTION_ID: gateways},
policies_lookup_failed={AZURE_SUBSCRIPTION_ID},
)
assert len(result) == 2
assert_subscription_finding(result[0], POLICIES_UNLISTED)
assert_gateway_finding(
result[1],
"FAIL",
f"{PREFIX} has a WAF policy associated, but the policy is disabled.",
)
def test_policies_lookup_failed_without_unknown_states_adds_nothing(self):
result = run_check(
{
AZURE_SUBSCRIPTION_ID: {
GATEWAY_ID: build_gateway(
firewall_policy_id=POLICY_ID, firewall_policy_enabled=True
)
}
},
policies_lookup_failed={AZURE_SUBSCRIPTION_ID},
)
assert len(result) == 1
assert_gateway_finding(
result[0], "PASS", f"{PREFIX} has an enabled WAF policy associated."
)
def test_failed_policy_read_reports_unknown_states_once(self):
other_id = GATEWAY_ID.replace("agw1", "agw2")
gateways = {
GATEWAY_ID: build_gateway(firewall_policy_id=POLICY_ID),
other_id: ApplicationGateway(
id=other_id,
name="agw2",
location="westeurope",
firewall_policy_id=POLICY_ID,
),
}
result = run_check(
{AZURE_SUBSCRIPTION_ID: gateways},
policy_reads_failed={POLICY_ID.casefold()},
)
assert len(result) == 1
assert_subscription_finding(result[0], POLICIES_UNLISTED)
def test_failed_policy_read_keeps_gateway_with_missing_policy(self):
other_id = GATEWAY_ID.replace("agw1", "agw2")
gateways = {
GATEWAY_ID: build_gateway(firewall_policy_id=POLICY_ID),
other_id: ApplicationGateway(
id=other_id,
name="agw2",
location="westeurope",
firewall_policy_id=POLICY_ID.replace("waf1", "waf2"),
),
}
result = run_check(
{AZURE_SUBSCRIPTION_ID: gateways},
policy_reads_failed={POLICY_ID.upper().casefold()},
)
assert len(result) == 2
assert_subscription_finding(result[0], POLICIES_UNLISTED)
assert result[1].resource_id == other_id
assert result[1].status == "MANUAL"
assert result[1].status_extended == (
f"Application Gateway agw2 in subscription {AZURE_SUBSCRIPTION_DISPLAY} has a WAF policy associated, but the state of that policy could not be determined. Check the scan log for the error."
)
def test_unlisted_gateways_and_policies_give_one_finding(self):
result = run_check(
{
AZURE_SUBSCRIPTION_ID: {
GATEWAY_ID: build_gateway(firewall_policy_id=POLICY_ID)
}
},
gateways_lookup_failed={AZURE_SUBSCRIPTION_ID},
policies_lookup_failed={AZURE_SUBSCRIPTION_ID},
)
assert len(result) == 1
assert_subscription_finding(result[0], BOTH_UNLISTED)
@@ -1,14 +1,23 @@
from unittest.mock import MagicMock, patch
from azure.mgmt.network.models import FlowLog
from azure.core.exceptions import ResourceNotFoundError
from azure.mgmt.network import models as azure_network_models
from azure.mgmt.network.models import (
ApplicationGatewaySku,
ApplicationGatewayWebApplicationFirewallConfiguration,
FlowLog,
SubResource,
)
from prowler.providers.azure.services.network.network_service import (
ApplicationGateway,
BastionHost,
Network,
NetworkWatcher,
PublicIp,
SecurityGroup,
VirtualNetwork,
WafPolicy,
)
from tests.providers.azure.azure_fixtures import (
AZURE_SUBSCRIPTION_ID,
@@ -83,6 +92,30 @@ def mock_network_get_virtual_networks(_):
}
def mock_network_get_waf_policies(_):
return {
AZURE_SUBSCRIPTION_ID: {
"waf_id": WafPolicy(
id="waf_id", name="waf_name", location="location", state="Enabled"
)
}
}
def mock_network_get_application_gateways(_):
return {
AZURE_SUBSCRIPTION_ID: {
"id": ApplicationGateway(
id="id",
name="name",
location="location",
firewall_policy_id="policy_id",
http2_enabled=True,
)
}
}
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_security_groups",
new=mock_network_get_security_groups,
@@ -103,6 +136,14 @@ def mock_network_get_virtual_networks(_):
"prowler.providers.azure.services.network.network_service.Network._get_virtual_networks",
new=mock_network_get_virtual_networks,
)
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_waf_policies",
new=mock_network_get_waf_policies,
)
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_application_gateways",
new=mock_network_get_application_gateways,
)
class Test_Network_Service:
def test_get_client(self):
network = Network(set_mocked_azure_provider())
@@ -122,6 +163,21 @@ class Test_Network_Service:
assert network.security_groups[AZURE_SUBSCRIPTION_ID][0].location == "location"
assert network.security_groups[AZURE_SUBSCRIPTION_ID][0].security_rules == []
def test_get_waf_policies(self):
network = Network(set_mocked_azure_provider())
policy = network.waf_policies[AZURE_SUBSCRIPTION_ID]["waf_id"]
assert policy.__class__.__name__ == "WafPolicy"
assert policy.name == "waf_name"
assert policy.state == "Enabled"
def test_get_application_gateways(self):
network = Network(set_mocked_azure_provider())
gateway = network.application_gateways[AZURE_SUBSCRIPTION_ID]["id"]
assert gateway.__class__.__name__ == "ApplicationGateway"
assert gateway.name == "name"
assert gateway.firewall_policy_id == "policy_id"
assert gateway.http2_enabled is True
def test_get_network_watchers(self):
network = Network(set_mocked_azure_provider())
assert (
@@ -185,6 +241,14 @@ class Test_Network_Service:
)
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_waf_policies",
new=mock_network_get_waf_policies,
)
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_application_gateways",
new=mock_network_get_application_gateways,
)
class Test_Network_get_security_groups:
def test_get_security_groups_no_resource_groups(self):
mock_client = MagicMock()
@@ -354,6 +418,14 @@ class Test_Network_get_security_groups:
)
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_waf_policies",
new=mock_network_get_waf_policies,
)
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_application_gateways",
new=mock_network_get_application_gateways,
)
class Test_Network_get_network_watchers:
def test_get_network_watchers_no_resource_groups(self):
mock_client = MagicMock()
@@ -458,6 +530,14 @@ class Test_Network_get_network_watchers:
assert AZURE_SUBSCRIPTION_ID in result
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_waf_policies",
new=mock_network_get_waf_policies,
)
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_application_gateways",
new=mock_network_get_application_gateways,
)
class Test_Network_get_bastion_hosts:
def test_get_bastion_hosts_no_resource_groups(self):
mock_client = MagicMock()
@@ -563,6 +643,14 @@ class Test_Network_get_bastion_hosts:
assert result[AZURE_SUBSCRIPTION_ID] == []
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_waf_policies",
new=mock_network_get_waf_policies,
)
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_application_gateways",
new=mock_network_get_application_gateways,
)
class Test_Network_get_public_ip_addresses:
def test_get_public_ip_addresses_no_resource_groups(self):
mock_client = MagicMock()
@@ -668,6 +756,14 @@ class Test_Network_get_public_ip_addresses:
assert result[AZURE_SUBSCRIPTION_ID] == []
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_waf_policies",
new=mock_network_get_waf_policies,
)
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_application_gateways",
new=mock_network_get_application_gateways,
)
class Test_Network_get_network_watchers_extra:
def test_get_network_watchers_with_multiple_resource_groups(self):
mock_client = MagicMock()
@@ -737,6 +833,14 @@ class Test_Network_get_network_watchers_extra:
mock_client.network_watchers.list.assert_not_called()
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_waf_policies",
new=mock_network_get_waf_policies,
)
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_application_gateways",
new=mock_network_get_application_gateways,
)
class Test_Network_get_bastion_hosts_extra:
def test_get_bastion_hosts_with_multiple_resource_groups(self):
mock_client = MagicMock()
@@ -806,6 +910,14 @@ class Test_Network_get_bastion_hosts_extra:
)
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_waf_policies",
new=mock_network_get_waf_policies,
)
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_application_gateways",
new=mock_network_get_application_gateways,
)
class Test_Network_get_public_ip_addresses_extra:
def test_get_public_ip_addresses_with_multiple_resource_groups(self):
mock_client = MagicMock()
@@ -875,6 +987,14 @@ class Test_Network_get_public_ip_addresses_extra:
)
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_waf_policies",
new=mock_network_get_waf_policies,
)
@patch(
"prowler.providers.azure.services.network.network_service.Network._get_application_gateways",
new=mock_network_get_application_gateways,
)
class Test_Network_get_virtual_networks_extra:
def _ctx(self):
return (
@@ -1085,3 +1205,395 @@ class Test_Network_get_virtual_networks_extra:
mock_client.virtual_networks.list.assert_called_once_with(
resource_group_name="RG"
)
NETWORK = "prowler.providers.azure.services.network.network_service.Network"
GATEWAY_ID = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourceGroups/{RESOURCE_GROUP}/providers/Microsoft.Network/applicationGateways/agw1"
GATEWAY_POLICY_ID = f"/subscriptions/{AZURE_SUBSCRIPTION_ID}/resourceGroups/{RESOURCE_GROUP}/providers/Microsoft.Network/ApplicationGatewayWebApplicationFirewallPolicies/waf1"
def build_network_with_client(mock_client, resource_groups=None):
with (
patch(f"{NETWORK}._get_security_groups", new=mock_network_get_security_groups),
patch(f"{NETWORK}._get_bastion_hosts", new=mock_network_get_bastion_hosts),
patch(
f"{NETWORK}._get_network_watchers", new=mock_network_get_network_watchers
),
patch(
f"{NETWORK}._get_public_ip_addresses",
new=mock_network_get_public_ip_addresses,
),
patch(
f"{NETWORK}._get_virtual_networks", new=mock_network_get_virtual_networks
),
patch(f"{NETWORK}._get_waf_policies", new=mock_network_get_waf_policies),
patch(
f"{NETWORK}._get_application_gateways",
new=mock_network_get_application_gateways,
),
):
network = Network(set_mocked_azure_provider())
network.clients = {AZURE_SUBSCRIPTION_ID: mock_client}
network.resource_groups = resource_groups
return network
def build_sdk_waf_policy(state, policy_id=GATEWAY_POLICY_ID, name="waf1"):
waf_policy = azure_network_models.WebApplicationFirewallPolicy(
id=policy_id,
location="westeurope",
policy_settings=azure_network_models.PolicySettings(state=state),
)
waf_policy.name = name
return waf_policy
def build_waf_policy_model(state, policy_id=GATEWAY_POLICY_ID):
return WafPolicy(id=policy_id, name="waf1", location="westeurope", state=state)
def build_sdk_application_gateway(name="agw1", policy_id=None):
gateway = azure_network_models.ApplicationGateway(
id=GATEWAY_ID.replace("agw1", name),
location="westeurope",
sku=ApplicationGatewaySku(name="WAF_v2", tier="WAF_v2"),
firewall_policy=SubResource(id=policy_id) if policy_id else None,
)
gateway.name = name
return gateway
def get_gateways(mock_client, listed_policies=(), resource_groups=None):
"""Collect gateways with the given WAF policies already listed."""
network = build_network_with_client(mock_client, resource_groups)
network.waf_policies = {
AZURE_SUBSCRIPTION_ID: {policy.id: policy for policy in listed_policies}
}
return network, network._get_application_gateways()
class Test_Network_get_waf_policies:
def test_maps_policy_state(self):
mock_client = MagicMock()
mock_client.web_application_firewall_policies.list_all.return_value = [
build_sdk_waf_policy("Enabled")
]
network = build_network_with_client(mock_client)
result = network._get_waf_policies()
assert network.waf_policies_lookup_failed == set()
policy = result[AZURE_SUBSCRIPTION_ID][GATEWAY_POLICY_ID]
assert policy.id == GATEWAY_POLICY_ID
assert policy.name == "waf1"
assert policy.location == "westeurope"
assert policy.state == "Enabled"
def test_uses_resource_group_scope(self):
mock_client = MagicMock()
mock_client.web_application_firewall_policies.list.return_value = []
network = build_network_with_client(
mock_client, {AZURE_SUBSCRIPTION_ID: [RESOURCE_GROUP]}
)
result = network._get_waf_policies()
mock_client.web_application_firewall_policies.list.assert_called_once_with(
resource_group_name=RESOURCE_GROUP
)
mock_client.web_application_firewall_policies.list_all.assert_not_called()
assert result == {AZURE_SUBSCRIPTION_ID: {}}
def test_api_error_marks_subscription_lookup_failed(self):
mock_client = MagicMock()
mock_client.web_application_firewall_policies.list_all.side_effect = Exception(
"boom"
)
network = build_network_with_client(mock_client)
result = network._get_waf_policies()
assert result == {AZURE_SUBSCRIPTION_ID: {}}
assert network.waf_policies_lookup_failed == {AZURE_SUBSCRIPTION_ID}
def test_resource_group_error_marks_subscription_lookup_failed(self):
mock_client = MagicMock()
mock_client.web_application_firewall_policies.list.side_effect = Exception(
"boom"
)
network = build_network_with_client(
mock_client, {AZURE_SUBSCRIPTION_ID: [RESOURCE_GROUP]}
)
result = network._get_waf_policies()
assert result == {AZURE_SUBSCRIPTION_ID: {}}
assert network.waf_policies_lookup_failed == {AZURE_SUBSCRIPTION_ID}
def test_missing_resource_group_is_not_a_failed_lookup(self):
mock_client = MagicMock()
mock_client.web_application_firewall_policies.list.side_effect = (
ResourceNotFoundError("gone")
)
network = build_network_with_client(
mock_client, {AZURE_SUBSCRIPTION_ID: [RESOURCE_GROUP]}
)
result = network._get_waf_policies()
assert result == {AZURE_SUBSCRIPTION_ID: {}}
assert network.waf_policies_lookup_failed == set()
def test_unmappable_policy_is_skipped_and_marks_subscription(self):
broken = azure_network_models.WebApplicationFirewallPolicy(
location="westeurope"
)
mock_client = MagicMock()
mock_client.web_application_firewall_policies.list_all.return_value = [
broken,
build_sdk_waf_policy("Enabled"),
]
network = build_network_with_client(mock_client)
result = network._get_waf_policies()
assert list(result[AZURE_SUBSCRIPTION_ID]) == [GATEWAY_POLICY_ID]
assert network.waf_policies_lookup_failed == {AZURE_SUBSCRIPTION_ID}
class Test_Network_get_application_gateways:
def test_maps_gateway_settings(self):
gateway = azure_network_models.ApplicationGateway(
id=GATEWAY_ID,
location="westeurope",
sku=ApplicationGatewaySku(name="WAF_v2", tier="WAF_v2"),
firewall_policy=SubResource(id=GATEWAY_POLICY_ID),
enable_http2=True,
)
gateway.name = "agw1"
mock_client = MagicMock()
mock_client.application_gateways.list_all.return_value = [gateway]
network, result = get_gateways(mock_client, [build_waf_policy_model("Enabled")])
assert network.application_gateways_lookup_failed == set()
mock_client.web_application_firewall_policies.get.assert_not_called()
mapped = result[AZURE_SUBSCRIPTION_ID][GATEWAY_ID]
assert mapped.name == "agw1"
assert mapped.location == "westeurope"
assert mapped.firewall_policy_id == GATEWAY_POLICY_ID
assert mapped.firewall_policy_enabled is True
assert mapped.waf_configuration_enabled is False
assert mapped.http2_enabled is True
def test_maps_legacy_waf_configuration(self):
waf_configuration = ApplicationGatewayWebApplicationFirewallConfiguration(
enabled=True,
firewall_mode="Prevention",
rule_set_type="OWASP",
rule_set_version="3.2",
)
gateway = azure_network_models.ApplicationGateway(
id=GATEWAY_ID,
location="westeurope",
sku=ApplicationGatewaySku(name="WAF_v2", tier="WAF_v2"),
web_application_firewall_configuration=waf_configuration,
)
gateway.name = "agw1"
mock_client = MagicMock()
mock_client.application_gateways.list_all.return_value = [gateway]
_, result = get_gateways(mock_client)
mapped = result[AZURE_SUBSCRIPTION_ID][GATEWAY_ID]
assert mapped.firewall_policy_id is None
assert mapped.firewall_policy_enabled is None
assert mapped.waf_configuration_enabled is True
assert mapped.http2_enabled is False
def test_uses_resource_group_scope(self):
mock_client = MagicMock()
mock_client.application_gateways.list.return_value = []
_, result = get_gateways(
mock_client, resource_groups={AZURE_SUBSCRIPTION_ID: [RESOURCE_GROUP]}
)
mock_client.application_gateways.list.assert_called_once_with(
resource_group_name=RESOURCE_GROUP
)
mock_client.application_gateways.list_all.assert_not_called()
assert result[AZURE_SUBSCRIPTION_ID] == {}
def test_api_error_marks_subscription_lookup_failed(self):
mock_client = MagicMock()
mock_client.application_gateways.list_all.side_effect = Exception("boom")
network, result = get_gateways(mock_client)
assert result == {AZURE_SUBSCRIPTION_ID: {}}
assert network.application_gateways_lookup_failed == {AZURE_SUBSCRIPTION_ID}
def test_resource_group_error_marks_subscription_lookup_failed(self):
mock_client = MagicMock()
mock_client.application_gateways.list.side_effect = Exception("boom")
network, result = get_gateways(
mock_client, resource_groups={AZURE_SUBSCRIPTION_ID: [RESOURCE_GROUP]}
)
assert result == {AZURE_SUBSCRIPTION_ID: {}}
assert network.application_gateways_lookup_failed == {AZURE_SUBSCRIPTION_ID}
def test_maps_disabled_waf_policy_whatever_its_casing(self):
mock_client = MagicMock()
mock_client.application_gateways.list_all.return_value = [
build_sdk_application_gateway(policy_id=GATEWAY_POLICY_ID.upper())
]
_, result = get_gateways(mock_client, [build_waf_policy_model("disabled")])
assert (
result[AZURE_SUBSCRIPTION_ID][GATEWAY_ID].firewall_policy_enabled is False
)
def test_unlisted_waf_policy_is_read_by_id(self):
mock_client = MagicMock()
mock_client.application_gateways.list_all.return_value = [
build_sdk_application_gateway(policy_id=GATEWAY_POLICY_ID)
]
mock_client.web_application_firewall_policies.get.return_value = (
build_sdk_waf_policy("Enabled")
)
_, result = get_gateways(mock_client)
mock_client.web_application_firewall_policies.get.assert_called_once_with(
resource_group_name=RESOURCE_GROUP, policy_name="waf1"
)
assert result[AZURE_SUBSCRIPTION_ID][GATEWAY_ID].firewall_policy_enabled is True
def test_waf_policy_read_error_is_recorded_and_leaves_state_unknown(self):
mock_client = MagicMock()
mock_client.application_gateways.list_all.return_value = [
build_sdk_application_gateway(policy_id=GATEWAY_POLICY_ID)
]
mock_client.web_application_firewall_policies.get.side_effect = Exception(
"boom"
)
network, result = get_gateways(mock_client)
assert result[AZURE_SUBSCRIPTION_ID][GATEWAY_ID].firewall_policy_enabled is None
assert network.waf_policy_reads_failed == {GATEWAY_POLICY_ID.casefold()}
assert network.application_gateways_lookup_failed == set()
def test_missing_waf_policy_is_not_a_failed_read(self):
mock_client = MagicMock()
mock_client.application_gateways.list_all.return_value = [
build_sdk_application_gateway(policy_id=GATEWAY_POLICY_ID)
]
mock_client.web_application_firewall_policies.get.side_effect = (
ResourceNotFoundError("gone")
)
network, result = get_gateways(mock_client)
assert result[AZURE_SUBSCRIPTION_ID][GATEWAY_ID].firewall_policy_enabled is None
assert network.waf_policy_reads_failed == set()
def test_waf_policy_in_another_subscription_is_not_read(self):
other_subscription_policy_id = GATEWAY_POLICY_ID.replace(
AZURE_SUBSCRIPTION_ID, "00000000-0000-0000-0000-000000000000"
)
mock_client = MagicMock()
mock_client.application_gateways.list_all.return_value = [
build_sdk_application_gateway(policy_id=other_subscription_policy_id)
]
network, result = get_gateways(mock_client)
mock_client.web_application_firewall_policies.get.assert_not_called()
assert result[AZURE_SUBSCRIPTION_ID][GATEWAY_ID].firewall_policy_enabled is None
assert network.waf_policy_reads_failed == set()
def test_waf_policy_without_known_state_leaves_state_unknown(self):
mock_client = MagicMock()
mock_client.application_gateways.list_all.return_value = [
build_sdk_application_gateway(policy_id=GATEWAY_POLICY_ID)
]
_, result = get_gateways(mock_client, [build_waf_policy_model(None)])
mock_client.web_application_firewall_policies.get.assert_not_called()
assert result[AZURE_SUBSCRIPTION_ID][GATEWAY_ID].firewall_policy_enabled is None
def test_policy_reference_that_is_not_a_waf_policy_id_is_not_read(self):
mock_client = MagicMock()
mock_client.application_gateways.list_all.return_value = [
build_sdk_application_gateway(policy_id=f"{GATEWAY_POLICY_ID}/extra")
]
network, result = get_gateways(mock_client)
mock_client.web_application_firewall_policies.get.assert_not_called()
assert result[AZURE_SUBSCRIPTION_ID][GATEWAY_ID].firewall_policy_enabled is None
assert network.waf_policy_reads_failed == set()
def test_shared_unlisted_waf_policy_is_read_once(self):
mock_client = MagicMock()
mock_client.application_gateways.list_all.return_value = [
build_sdk_application_gateway(policy_id=GATEWAY_POLICY_ID),
build_sdk_application_gateway(
name="agw2", policy_id=GATEWAY_POLICY_ID.upper()
),
]
mock_client.web_application_firewall_policies.get.return_value = (
build_sdk_waf_policy("Enabled")
)
_, result = get_gateways(mock_client)
assert len(result[AZURE_SUBSCRIPTION_ID]) == 2
mock_client.web_application_firewall_policies.get.assert_called_once()
def test_unmappable_gateway_is_skipped_and_marks_subscription(self):
broken = azure_network_models.ApplicationGateway(location="westeurope")
mock_client = MagicMock()
mock_client.application_gateways.list_all.return_value = [
broken,
build_sdk_application_gateway(),
]
network, result = get_gateways(mock_client)
assert list(result[AZURE_SUBSCRIPTION_ID]) == [GATEWAY_ID]
assert network.application_gateways_lookup_failed == {AZURE_SUBSCRIPTION_ID}
def test_failed_resource_group_keeps_gateways_from_the_others(self):
mock_client = MagicMock()
mock_client.application_gateways.list.side_effect = [
[build_sdk_application_gateway()],
Exception("boom"),
]
network, result = get_gateways(
mock_client,
resource_groups={AZURE_SUBSCRIPTION_ID: [RESOURCE_GROUP, "rg2"]},
)
assert list(result[AZURE_SUBSCRIPTION_ID]) == [GATEWAY_ID]
assert network.application_gateways_lookup_failed == {AZURE_SUBSCRIPTION_ID}
def test_missing_resource_group_is_not_a_failed_lookup(self):
mock_client = MagicMock()
mock_client.application_gateways.list.side_effect = ResourceNotFoundError(
"gone"
)
network, result = get_gateways(
mock_client, resource_groups={AZURE_SUBSCRIPTION_ID: [RESOURCE_GROUP]}
)
assert result == {AZURE_SUBSCRIPTION_ID: {}}
assert network.application_gateways_lookup_failed == set()