Compare commits

...
Author SHA1 Message Date
Hugo P.Brito 8fcb8e53b7 fix(huaweicloud): preserve FunctionGraph region attribution
- Carry regional client keys through function discovery
- Cover regionless SDK clients with a regression test
- Document FunctionGraph initialization behavior
2026-08-20 13:13:55 +01:00
Hugo P.Brito 84e9a91186 fix(huaweicloud): clarify FunctionGraph VPC guidance
- Separate VPC attachment from public network access controls

- Add supported remediation steps and API documentation
2026-08-20 12:49:36 +01:00
Hugo P.Brito c1eceb9b5a fix(huaweicloud): enable FunctionGraph runtime discovery
- Add the pinned FunctionGraph SDK and regional session client

- Paginate discovery without publishing partial API results

- Preserve canonical function identities and VPC state in reports
2026-08-20 12:48:44 +01:00
tomitobio 082915cad9 Add changelog fragment for huaweicloud_functiongraph_function_vpc_configured 2026-08-12 04:41:20 +08:00
tomitobio ec6878c824 feat(providers/huaweicloud): add functiongraph_function_vpc_configured check 2026-07-29 04:11:52 +08:00
13 changed files with 569 additions and 0 deletions
@@ -0,0 +1 @@
`functiongraph_function_vpc_configured` check for Huawei Cloud provider: FunctionGraph functions are configured within a VPC
+15
View File
@@ -384,6 +384,21 @@ class HuaweiCloudSession:
.build()
)
elif service == "functiongraph":
from huaweicloudsdkfunctiongraph.v2 import FunctionGraphClient
from huaweicloudsdkfunctiongraph.v2.region.functiongraph_region import (
FunctionGraphRegion,
)
client_region = region or self._region
return (
FunctionGraphClient.new_builder()
.with_credentials(self._get_basic_credentials(client_region))
.with_http_config(self._http_config())
.with_region(_aligned_region(FunctionGraphRegion, client_region))
.build()
)
else:
raise HuaweiCloudServiceError(
message=f"Huawei Cloud service '{service}' is not supported"
@@ -0,0 +1,6 @@
from prowler.providers.common.provider import Provider
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_service import (
FunctionGraph,
)
functiongraph_client = FunctionGraph(Provider.get_global_provider())
@@ -0,0 +1,36 @@
{
"Provider": "huaweicloud",
"CheckID": "functiongraph_function_vpc_configured",
"CheckTitle": "FunctionGraph functions are configured within a VPC",
"CheckType": [],
"ServiceName": "functiongraph",
"SubServiceName": "",
"ResourceIdTemplate": "",
"Severity": "medium",
"ResourceType": "FunctionGraphFunction",
"ResourceGroup": "serverless",
"Description": "Ensure FunctionGraph functions are attached to a VPC so workloads that require private connectivity can use VPC network controls.",
"Risk": "Functions without a VPC attachment cannot privately access VPC resources such as databases or internal services. They may depend on public service endpoints, reducing network isolation and increasing exposure to public-network paths.",
"RelatedUrl": "",
"AdditionalURLs": [
"https://support.huaweicloud.com/intl/en-us/api-functiongraph/functiongraph_06_0111.html"
],
"Remediation": {
"Code": {
"CLI": "",
"NativeIaC": "",
"Other": "In the FunctionGraph console, open the function, edit its network configuration, select the required VPC, subnet, and security groups, and save the configuration.",
"Terraform": ""
},
"Recommendation": {
"Text": "Attach functions that require private resource access to an appropriate VPC, subnet, and least-privilege security groups.",
"Url": "https://hub.prowler.com/check/functiongraph_function_vpc_configured"
}
},
"Categories": [
"trust-boundaries"
],
"DependsOn": [],
"RelatedTo": [],
"Notes": ""
}
@@ -0,0 +1,29 @@
from prowler.lib.check.models import Check, CheckReportHuaweiCloud
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_client import (
functiongraph_client,
)
class functiongraph_function_vpc_configured(Check):
"""Check if FunctionGraph functions are configured within a VPC."""
def execute(self) -> list[CheckReportHuaweiCloud]:
findings = []
for function in functiongraph_client.functions:
report = CheckReportHuaweiCloud(
metadata=self.metadata(),
resource=function,
)
if function.vpc_id:
report.status = "PASS"
report.status_extended = f"Function '{function.name}' is configured within VPC '{function.vpc_id}'."
else:
report.status = "FAIL"
report.status_extended = (
f"Function '{function.name}' is not configured within a VPC."
)
findings.append(report)
return findings
@@ -0,0 +1,118 @@
from typing import List, Optional
from prowler.lib.logger import logger
from prowler.lib.scan_filters.scan_filters import is_resource_filtered
from prowler.providers.huaweicloud.lib.service.service import HuaweiCloudService
from prowler.providers.huaweicloud.models import HuaweiCloudBaseModel
class FunctionGraph(HuaweiCloudService):
"""
FunctionGraph service class for Huawei Cloud.
This class provides methods to interact with Huawei Cloud FunctionGraph service
to retrieve serverless functions and their security configuration.
"""
def __init__(self, provider):
"""Initialize FunctionGraph and discover functions in enabled regions.
Args:
provider: The Huawei Cloud provider used to create regional clients.
"""
super().__init__(__class__.__name__, provider)
self.functions: List[FunctionGraphFunction] = []
if getattr(self.session, "is_mock", False):
self._load_mock_data()
return
self.__threading_call__(
self._list_functions, iterator=self.regional_clients.items()
)
def _load_mock_data(self):
"""Load mock data for testing."""
region = "la-south-2"
self.functions = [
FunctionGraphFunction(
id="fg-mock-001",
name="function-secure",
arn="urn:fss:la-south-2:project:function:default:function-secure:latest",
runtime="Python3.9",
timeout=30,
memory_size=128,
vpc_id="vpc-12345",
region=region,
),
FunctionGraphFunction(
id="fg-mock-002",
name="function-insecure",
arn="urn:fss:la-south-2:project:function:default:function-insecure:latest",
runtime="Python3.9",
timeout=30,
memory_size=128,
vpc_id=None,
region=region,
),
]
def _list_functions(self, regional_client_item):
"""List every FunctionGraph function in one region."""
region, regional_client = regional_client_item
logger.info(f"FunctionGraph - Listing Functions in {region}...")
discovered_functions = []
marker = None
try:
from huaweicloudsdkfunctiongraph.v2 import ListFunctionsRequest
while True:
request = ListFunctionsRequest(marker=marker, maxitems="400")
response = self._call_with_retries(
regional_client.list_functions, request
)
for function in getattr(response, "functions", None) or []:
resource_id = getattr(function, "resource_id", None) or ""
if self.audit_resources and not is_resource_filtered(
resource_id, self.audit_resources
):
continue
discovered_functions.append(
FunctionGraphFunction(
id=resource_id,
name=getattr(function, "func_name", None) or resource_id,
arn=getattr(function, "func_urn", None) or "",
runtime=getattr(function, "runtime", None) or "",
timeout=getattr(function, "timeout", None) or 0,
memory_size=getattr(function, "memory_size", None) or 0,
vpc_id=getattr(function, "func_vpc_id", None),
region=region,
)
)
next_marker = getattr(response, "next_marker", None)
if not next_marker or str(next_marker) == marker:
break
marker = str(next_marker)
self.functions.extend(discovered_functions)
except Exception as error:
logger.error(
f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
class FunctionGraphFunction(HuaweiCloudBaseModel):
"""FunctionGraph function model."""
id: str
name: str = ""
arn: str = ""
runtime: str = ""
timeout: int = 0
memory_size: int = 0
vpc_id: Optional[str] = None
region: str = ""
+2
View File
@@ -123,6 +123,7 @@ dependencies = [
"huaweicloudsdkecs==3.1.204",
"huaweicloudsdkelb==3.1.204",
"huaweicloudsdkevs==3.1.204",
"huaweicloudsdkfunctiongraph==3.1.204",
"huaweicloudsdkiam==3.1.204",
"huaweicloudsdkkms==3.1.204",
"huaweicloudsdkobs==3.1.204",
@@ -260,6 +261,7 @@ constraint-dependencies = [
"huaweicloudsdkecs==3.1.204",
"huaweicloudsdkelb==3.1.204",
"huaweicloudsdkevs==3.1.204",
"huaweicloudsdkfunctiongraph==3.1.204",
"huaweicloudsdkiam==3.1.204",
"huaweicloudsdkkms==3.1.204",
"huaweicloudsdkobs==3.1.204",
@@ -50,6 +50,31 @@ class TestHuaweiCloudProviderSetupSession:
session = HuaweicloudProvider.setup_session()
assert session.get_credentials().ak == ACCESS_KEY
def test_creates_functiongraph_client_for_europe(self):
session = HuaweiCloudSession(
HuaweiCloudCredentials(ak=ACCESS_KEY, sk=SECRET_KEY)
)
functiongraph_client = mock.MagicMock()
builder = mock.MagicMock()
builder.with_credentials.return_value.with_http_config.return_value.with_region.return_value.build.return_value = (
functiongraph_client
)
with mock.patch(
"huaweicloudsdkfunctiongraph.v2.FunctionGraphClient.new_builder",
return_value=builder,
):
client = session.client("functiongraph", "eu-west-101")
assert client is functiongraph_client
region = builder.with_credentials.return_value.with_http_config.return_value.with_region.call_args.args[
0
]
assert region.id == "eu-west-101"
assert region.endpoints == [
"https://functiongraph.eu-west-101.myhuaweicloud.eu"
]
class TestHuaweiCloudProviderValidateCredentials:
def test_resolves_caller_identity_from_iam(self):
@@ -0,0 +1,184 @@
from unittest import mock
from prowler.lib.check.models import CheckMetadata
from tests.providers.huaweicloud.huaweicloud_fixtures import (
set_mocked_huaweicloud_provider,
)
class Test_functiongraph_function_vpc_configured:
def test_metadata_describes_vpc_attachment_accurately(self):
metadata = CheckMetadata.parse_file(
"prowler/providers/huaweicloud/services/functiongraph/"
"functiongraph_function_vpc_configured/"
"functiongraph_function_vpc_configured.metadata.json"
)
assert metadata.ResourceIdTemplate == ""
assert "direct internet access" not in metadata.Risk
assert metadata.AdditionalURLs == [
"https://support.huaweicloud.com/intl/en-us/api-functiongraph/functiongraph_06_0111.html"
]
assert metadata.Remediation.Code.Other
def test_functiongraph_vpc_configured_pass(self):
functiongraph_client = mock.MagicMock()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_huaweicloud_provider(),
),
mock.patch(
"prowler.providers.huaweicloud.services.functiongraph.functiongraph_function_vpc_configured.functiongraph_function_vpc_configured.functiongraph_client",
new=functiongraph_client,
),
):
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_function_vpc_configured.functiongraph_function_vpc_configured import (
functiongraph_function_vpc_configured,
)
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_service import (
FunctionGraphFunction,
)
functiongraph_client.functions = [
FunctionGraphFunction(
id="fg-001",
name="function-secure",
arn="urn:fss:la-south-2:project:function:default:function-secure:latest",
runtime="Python3.9",
timeout=30,
memory_size=128,
vpc_id="vpc-12345",
region="la-south-2",
),
]
functiongraph_client.audited_account = "123456789012"
check = functiongraph_function_vpc_configured()
results = check.execute()
assert len(results) == 1
assert results[0].status == "PASS"
assert results[0].resource_id == "fg-001"
assert results[0].resource_name == "function-secure"
assert results[0].resource_arn.endswith("function-secure:latest")
assert "configured within VPC" in results[0].status_extended
def test_functiongraph_vpc_configured_fail(self):
functiongraph_client = mock.MagicMock()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_huaweicloud_provider(),
),
mock.patch(
"prowler.providers.huaweicloud.services.functiongraph.functiongraph_function_vpc_configured.functiongraph_function_vpc_configured.functiongraph_client",
new=functiongraph_client,
),
):
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_function_vpc_configured.functiongraph_function_vpc_configured import (
functiongraph_function_vpc_configured,
)
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_service import (
FunctionGraphFunction,
)
functiongraph_client.functions = [
FunctionGraphFunction(
id="fg-002",
name="function-insecure",
arn="urn:fss:la-south-2:project:function:default:function-insecure:latest",
runtime="Python3.9",
timeout=30,
memory_size=128,
vpc_id=None,
region="la-south-2",
),
]
functiongraph_client.audited_account = "123456789012"
check = functiongraph_function_vpc_configured()
results = check.execute()
assert len(results) == 1
assert results[0].status == "FAIL"
assert results[0].resource_id == "fg-002"
assert "not configured within a VPC" in results[0].status_extended
def test_functiongraph_vpc_configured_mixed(self):
functiongraph_client = mock.MagicMock()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_huaweicloud_provider(),
),
mock.patch(
"prowler.providers.huaweicloud.services.functiongraph.functiongraph_function_vpc_configured.functiongraph_function_vpc_configured.functiongraph_client",
new=functiongraph_client,
),
):
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_function_vpc_configured.functiongraph_function_vpc_configured import (
functiongraph_function_vpc_configured,
)
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_service import (
FunctionGraphFunction,
)
functiongraph_client.functions = [
FunctionGraphFunction(
id="fg-001",
name="function-secure",
arn="urn:fss:la-south-2:project:function:default:function-secure:latest",
runtime="Python3.9",
timeout=30,
memory_size=128,
vpc_id="vpc-12345",
region="la-south-2",
),
FunctionGraphFunction(
id="fg-002",
name="function-insecure",
arn="urn:fss:la-south-2:project:function:default:function-insecure:latest",
runtime="Python3.9",
timeout=30,
memory_size=128,
vpc_id=None,
region="la-south-2",
),
]
functiongraph_client.audited_account = "123456789012"
check = functiongraph_function_vpc_configured()
results = check.execute()
assert len(results) == 2
assert results[0].status == "PASS"
assert results[1].status == "FAIL"
def test_functiongraph_vpc_configured_empty(self):
functiongraph_client = mock.MagicMock()
with (
mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=set_mocked_huaweicloud_provider(),
),
mock.patch(
"prowler.providers.huaweicloud.services.functiongraph.functiongraph_function_vpc_configured.functiongraph_function_vpc_configured.functiongraph_client",
new=functiongraph_client,
),
):
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_function_vpc_configured.functiongraph_function_vpc_configured import (
functiongraph_function_vpc_configured,
)
functiongraph_client.functions = []
functiongraph_client.audited_account = "123456789012"
check = functiongraph_function_vpc_configured()
results = check.execute()
assert len(results) == 0
@@ -0,0 +1,139 @@
from types import SimpleNamespace
from unittest import mock
from prowler.providers.huaweicloud.services.functiongraph.functiongraph_service import (
FunctionGraph,
)
from tests.providers.huaweicloud.huaweicloud_fixtures import (
set_mocked_huaweicloud_provider,
)
class TestFunctionGraphService:
def test_real_session_lists_functions(self):
function = SimpleNamespace(
resource_id="resource-1",
func_urn="urn:fss:eu-west-101:project-1:function:default:first:latest",
func_name="first",
runtime="Python3.9",
timeout=30,
memory_size=128,
func_vpc_id=None,
)
regional_client = SimpleNamespace(list_functions=mock.MagicMock())
regional_client.list_functions.return_value = SimpleNamespace(
functions=[function], next_marker=None
)
provider = set_mocked_huaweicloud_provider()
provider.session = SimpleNamespace(client=mock.MagicMock())
provider.generate_regional_clients = mock.MagicMock(
return_value={"eu-west-101": regional_client}
)
provider.get_default_region = mock.MagicMock(return_value="eu-west-101")
service = FunctionGraph(provider)
assert len(service.functions) == 1
assert service.functions[0].region == "eu-west-101"
regional_client.list_functions.assert_called_once()
def test_lists_every_page_and_maps_sdk_fields(self):
first_function = SimpleNamespace(
resource_id="resource-1",
func_urn="urn:fss:eu-west-101:project-1:function:default:first:latest",
func_name="first",
runtime=None,
timeout=None,
memory_size=None,
func_vpc_id=None,
)
second_function = SimpleNamespace(
resource_id="resource-2",
func_urn="urn:fss:eu-west-101:project-1:function:default:second:latest",
func_name="second",
runtime="Python3.9",
timeout=30,
memory_size=128,
func_vpc_id="vpc-1",
)
regional_client = mock.MagicMock(region="eu-west-101")
regional_client.list_functions.side_effect = [
SimpleNamespace(functions=[first_function], next_marker=400),
SimpleNamespace(functions=[second_function], next_marker=None),
]
service = FunctionGraph.__new__(FunctionGraph)
service.functions = []
service.audit_resources = []
service._call_with_retries = mock.MagicMock(
side_effect=regional_client.list_functions.side_effect
)
service._list_functions(("eu-west-101", regional_client))
assert [function.id for function in service.functions] == [
"resource-1",
"resource-2",
]
assert service.functions[0].arn == first_function.func_urn
assert service.functions[0].runtime == ""
assert service.functions[0].timeout == 0
assert service.functions[0].memory_size == 0
assert service.functions[0].vpc_id is None
assert service.functions[1].vpc_id == "vpc-1"
assert service._call_with_retries.call_count == 2
first_request = service._call_with_retries.call_args_list[0].args[1]
second_request = service._call_with_retries.call_args_list[1].args[1]
assert first_request.maxitems == "400"
assert first_request.marker is None
assert second_request.marker == "400"
def test_api_error_does_not_fabricate_functions(self):
function = SimpleNamespace(
resource_id="resource-1",
func_urn="urn:fss:eu-west-101:project-1:function:default:first:latest",
func_name="first",
runtime="Python3.9",
timeout=30,
memory_size=128,
func_vpc_id=None,
)
regional_client = mock.MagicMock(region="eu-west-101")
service = FunctionGraph.__new__(FunctionGraph)
service.functions = []
service.audit_resources = []
service._call_with_retries = mock.MagicMock(
side_effect=[
SimpleNamespace(functions=[function], next_marker=400),
Exception("boom"),
]
)
service._list_functions(("eu-west-101", regional_client))
assert service.functions == []
def test_repeated_marker_stops_pagination(self):
function = SimpleNamespace(
resource_id="resource-1",
func_urn="urn:fss:eu-west-101:project-1:function:default:first:latest",
func_name="first",
runtime="Python3.9",
timeout=30,
memory_size=128,
func_vpc_id=None,
)
regional_client = mock.MagicMock(region="eu-west-101")
service = FunctionGraph.__new__(FunctionGraph)
service.functions = []
service.audit_resources = []
service._call_with_retries = mock.MagicMock(
side_effect=[
SimpleNamespace(functions=[function], next_marker=400),
SimpleNamespace(functions=[], next_marker=400),
]
)
service._list_functions(("eu-west-101", regional_client))
assert [item.id for item in service.functions] == ["resource-1"]
assert service._call_with_retries.call_count == 2
Generated
+14
View File
@@ -100,6 +100,7 @@ constraints = [
{ name = "huaweicloudsdkecs", specifier = "==3.1.204" },
{ name = "huaweicloudsdkelb", specifier = "==3.1.204" },
{ name = "huaweicloudsdkevs", specifier = "==3.1.204" },
{ name = "huaweicloudsdkfunctiongraph", specifier = "==3.1.204" },
{ name = "huaweicloudsdkiam", specifier = "==3.1.204" },
{ name = "huaweicloudsdkkms", specifier = "==3.1.204" },
{ name = "huaweicloudsdkobs", specifier = "==3.1.204" },
@@ -2310,6 +2311,17 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/f7/cf/531dc55fd9d0f3bbd3eef24c7e4d78c6a1ba8eb80fa506e3574d72bcc98a/huaweicloudsdkevs-3.1.204-py3-none-any.whl", hash = "sha256:9118ac4c576e54aa7eaa926949e2b6824c5f038a2274b51d9a304d37fc0d7e2f", size = 251404, upload-time = "2026-07-09T09:02:50.05Z" },
]
[[package]]
name = "huaweicloudsdkfunctiongraph"
version = "3.1.204"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "huaweicloudsdkcore" },
]
wheels = [
{ url = "https://files.pythonhosted.org/packages/1a/10/19e2f8a0f15aa6c3c1b2f67eb58e5d06aac0b3eef527283624acfd946b0a/huaweicloudsdkfunctiongraph-3.1.204-py3-none-any.whl", hash = "sha256:0999e57f56b157817c9d8d68940279feff6c1170693e2f41eef21a1fd12a7de9", size = 556873, upload-time = "2026-07-09T09:02:52.402Z" },
]
[[package]]
name = "huaweicloudsdkiam"
version = "3.1.204"
@@ -3753,6 +3765,7 @@ dependencies = [
{ name = "huaweicloudsdkecs" },
{ name = "huaweicloudsdkelb" },
{ name = "huaweicloudsdkevs" },
{ name = "huaweicloudsdkfunctiongraph" },
{ name = "huaweicloudsdkiam" },
{ name = "huaweicloudsdkkms" },
{ name = "huaweicloudsdkobs" },
@@ -3873,6 +3886,7 @@ requires-dist = [
{ name = "huaweicloudsdkecs", specifier = "==3.1.204" },
{ name = "huaweicloudsdkelb", specifier = "==3.1.204" },
{ name = "huaweicloudsdkevs", specifier = "==3.1.204" },
{ name = "huaweicloudsdkfunctiongraph", specifier = "==3.1.204" },
{ name = "huaweicloudsdkiam", specifier = "==3.1.204" },
{ name = "huaweicloudsdkkms", specifier = "==3.1.204" },
{ name = "huaweicloudsdkobs", specifier = "==3.1.204" },