From 4eb5e9eb23bf2b720c11373b2befcf2b14cc3fca Mon Sep 17 00:00:00 2001 From: Izaak Gough Date: Wed, 24 Jun 2026 11:37:35 +0100 Subject: [PATCH 1/2] feat: add support for VPC direct connect --- src/firebase_functions/options.py | 128 ++++++++++++++++++--- src/firebase_functions/private/manifest.py | 13 ++- tests/test_options.py | 100 ++++++++++++++++ 3 files changed, 225 insertions(+), 16 deletions(-) diff --git a/src/firebase_functions/options.py b/src/firebase_functions/options.py index 3d432739..238a3834 100644 --- a/src/firebase_functions/options.py +++ b/src/firebase_functions/options.py @@ -58,6 +58,29 @@ def __str__(self) -> str: return self.value +@_dataclasses.dataclass(frozen=True) +class NetworkInterface: + """ + Interface for a direct VPC network connection. + At least one of ``network`` or ``subnetwork`` must be specified. + """ + + network: str | Expression[str] | _util.Sentinel | None = None + """ + Network to use for Direct VPC Egress. + """ + + subnetwork: str | Expression[str] | _util.Sentinel | None = None + """ + Subnetwork to use for Direct VPC Egress. + """ + + tags: str | list[str] | Expression[str] | Expression[list] | _util.Sentinel | None = None + """ + Optional tags to apply to Direct VPC Egress traffic. + """ + + @_dataclasses.dataclass(frozen=True) class CorsOptions: """ @@ -257,6 +280,7 @@ class RuntimeOptions: vpc_connector: str | _util.Sentinel | None = None """ Connect function to specified VPC connector. + Mutually exclusive with ``network_interface``. A value of ``RESET_VALUE`` removes the VPC connector. """ @@ -266,6 +290,18 @@ class RuntimeOptions: A value of ``RESET_VALUE`` turns off VPC connector egress settings. """ + vpc_egress: VpcEgressSetting | _util.Sentinel | None = None + """ + Alias for ``vpc_connector_egress_settings``. + """ + + network_interface: NetworkInterface | _util.Sentinel | None = None + """ + Direct VPC Egress network interface settings. + Mutually exclusive with ``vpc_connector``. + A value of ``RESET_VALUE`` removes Direct VPC Egress settings. + """ + service_account: str | _util.Sentinel | None = None """ Specific service account for the function to run as. @@ -339,6 +375,8 @@ def _asdict_with_global_options(self) -> dict: "service_account", "vpc_connector", "vpc_connector_egress_settings", + "vpc_egress", + "network_interface", ] if not preserve_external_changes: for option in resettable_options: @@ -380,18 +418,76 @@ def convert_secret(secret) -> _manifest.SecretEnvironmentVariable: elif options.region is not None: region = [_typing.cast(str, options.region)] - vpc: _manifest.VpcSettings | None = None - if isinstance(options.vpc_connector, str): - vpc = ( - { - "connector": options.vpc_connector, - "egressSettings": options.vpc_connector_egress_settings.value - if isinstance(options.vpc_connector_egress_settings, VpcEgressSetting) - else options.vpc_connector_egress_settings, - } - if options.vpc_connector_egress_settings is not None - else {"connector": options.vpc_connector} - ) + vpc: _manifest.VpcSettings | _util.Sentinel | None = None + vpc_egress = options.vpc_connector_egress_settings + if options.vpc_egress is not None and ( + not isinstance(options.vpc_egress, _util.Sentinel) or vpc_egress is None + ): + vpc_egress = options.vpc_egress + + connector = options.vpc_connector + has_connector = isinstance(connector, str) + reset_connector = isinstance(connector, _util.Sentinel) + + raw_network_interface = options_dict.get("network_interface") + has_network_interface = isinstance(raw_network_interface, dict) + reset_network_interface = isinstance(raw_network_interface, _util.Sentinel) + + if has_network_interface: + network_interface_dict = _typing.cast(dict[str, object], raw_network_interface) + if ( + network_interface_dict.get("network") is None + and network_interface_dict.get("subnetwork") is None + ): + raise ValueError( + "At least one of network or subnetwork must be specified in network_interface." + ) + + if has_connector and has_network_interface: + raise ValueError("Cannot set both vpc_connector and network_interface") + + if ( + has_connector + or vpc_egress is not None + or has_network_interface + or reset_connector + or reset_network_interface + ): + if (reset_connector and not has_network_interface) or ( + reset_network_interface and not has_connector + ): + vpc = RESET_VALUE + else: + vpc = {} + if has_connector: + vpc["connector"] = _typing.cast(str, connector) + if vpc_egress is not None: + vpc["egressSettings"] = ( + vpc_egress.value if isinstance(vpc_egress, VpcEgressSetting) else vpc_egress + ) + if has_network_interface: + network_interface_dict = _typing.cast(dict[str, object], raw_network_interface) + network_interface_spec: _manifest.VpcNetworkInterface = {} + if network_interface_dict.get("network") is not None: + network_interface_spec["network"] = _typing.cast( + str | Expression[str] | _util.Sentinel, + network_interface_dict["network"], + ) + if network_interface_dict.get("subnetwork") is not None: + network_interface_spec["subnetwork"] = _typing.cast( + str | Expression[str] | _util.Sentinel, + network_interface_dict["subnetwork"], + ) + if network_interface_dict.get("tags") is not None: + network_interface_spec["tags"] = _typing.cast( + str + | list[str] + | Expression[str] + | Expression[list] + | _util.Sentinel, + network_interface_dict["tags"], + ) + vpc["networkInterfaces"] = [network_interface_spec] endpoint = _manifest.ManifestEndpoint( entryPoint=kwargs["func_name"], @@ -1254,8 +1350,10 @@ def set_global_options( max_instances: int | Expression[int] | _util.Sentinel | None = None, concurrency: int | Expression[int] | _util.Sentinel | None = None, cpu: int | _typing.Literal["gcf_gen1"] | _util.Sentinel = "gcf_gen1", - vpc_connector: str | None = None, - vpc_connector_egress_settings: VpcEgressSetting | None = None, + vpc_connector: str | _util.Sentinel | None = None, + vpc_connector_egress_settings: VpcEgressSetting | _util.Sentinel | None = None, + vpc_egress: VpcEgressSetting | _util.Sentinel | None = None, + network_interface: NetworkInterface | _util.Sentinel | None = None, service_account: str | _util.Sentinel | None = None, ingress: IngressSetting | _util.Sentinel | None = None, labels: dict[str, str] | None = None, @@ -1277,6 +1375,8 @@ def set_global_options( cpu=cpu, vpc_connector=vpc_connector, vpc_connector_egress_settings=vpc_connector_egress_settings, + vpc_egress=vpc_egress, + network_interface=network_interface, service_account=service_account, ingress=ingress, labels=labels, diff --git a/src/firebase_functions/private/manifest.py b/src/firebase_functions/private/manifest.py index 7672a9f5..502f4e75 100644 --- a/src/firebase_functions/private/manifest.py +++ b/src/firebase_functions/private/manifest.py @@ -138,9 +138,18 @@ class BlockingTrigger(_typing.TypedDict): options: _typing_extensions.NotRequired[BlockingTriggerOptions] +class VpcNetworkInterface(_typing.TypedDict): + network: _typing_extensions.NotRequired[str | _params.Expression[str] | _util.Sentinel] + subnetwork: _typing_extensions.NotRequired[str | _params.Expression[str] | _util.Sentinel] + tags: _typing_extensions.NotRequired[ + str | list[str] | _params.Expression[str] | _params.Expression[list] | _util.Sentinel + ] + + class VpcSettings(_typing.TypedDict): - connector: _typing_extensions.Required[str] + connector: _typing_extensions.NotRequired[str] egressSettings: _typing_extensions.NotRequired[str | _util.Sentinel] + networkInterfaces: _typing_extensions.NotRequired[list[VpcNetworkInterface] | _util.Sentinel] @_dataclasses.dataclass(frozen=True) @@ -157,7 +166,7 @@ class ManifestEndpoint: serviceAccountEmail: str | _util.Sentinel | None = None timeoutSeconds: int | _params.Expression[int] | _util.Sentinel | None = None cpu: int | str | _util.Sentinel | None = None - vpc: VpcSettings | None = None + vpc: VpcSettings | _util.Sentinel | None = None labels: dict[str, str] | None = None ingressSettings: str | None | _util.Sentinel = None secretEnvironmentVariables: list[SecretEnvironmentVariable] | _util.Sentinel | None = ( diff --git a/tests/test_options.py b/tests/test_options.py index 3e9cb523..8865b8cb 100644 --- a/tests/test_options.py +++ b/tests/test_options.py @@ -86,6 +86,69 @@ def test_options_asdict_uses_cel_representation(): ) +def test_vpc_connector_endpoint_unchanged(): + endpoint = options.HttpsOptions( + vpc_connector="projects/test/locations/us-central1/connectors/test", + vpc_connector_egress_settings=options.VpcEgressSetting.ALL_TRAFFIC, + )._endpoint(func_name="test") + + assert endpoint.vpc == { + "connector": "projects/test/locations/us-central1/connectors/test", + "egressSettings": "ALL_TRAFFIC", + } + + +def test_network_interface_endpoint(): + endpoint = options.HttpsOptions( + network_interface=options.NetworkInterface( + network="default", + tags=["internal", "egress"], + ), + vpc_egress=options.VpcEgressSetting.PRIVATE_RANGES_ONLY, + )._endpoint(func_name="test") + + assert endpoint.vpc == { + "networkInterfaces": [ + { + "network": "default", + "tags": ["internal", "egress"], + } + ], + "egressSettings": "PRIVATE_RANGES_ONLY", + } + + +def test_vpc_egress_alias_takes_precedence(): + endpoint = options.HttpsOptions( + vpc_connector="projects/test/locations/us-central1/connectors/test", + vpc_connector_egress_settings=options.VpcEgressSetting.ALL_TRAFFIC, + vpc_egress=options.VpcEgressSetting.PRIVATE_RANGES_ONLY, + )._endpoint(func_name="test") + + assert endpoint.vpc == { + "connector": "projects/test/locations/us-central1/connectors/test", + "egressSettings": "PRIVATE_RANGES_ONLY", + } + + +def test_network_interface_requires_network_or_subnetwork(): + with raises( + ValueError, + match="At least one of network or subnetwork must be specified in network_interface.", + ): + options.HttpsOptions( + network_interface=options.NetworkInterface(), + )._endpoint(func_name="test") + + +def test_vpc_connector_and_network_interface_are_mutually_exclusive(): + with raises(ValueError, match="Cannot set both vpc_connector and network_interface"): + options.HttpsOptions( + vpc_connector="projects/test/locations/us-central1/connectors/test", + network_interface=options.NetworkInterface(network="default"), + )._endpoint(func_name="test") + + def test_options_preserve_external_changes(): """ Testing if setting a global option internally change the values. @@ -100,6 +163,10 @@ def test_options_preserve_external_changes(): options_asdict = options._GLOBAL_OPTIONS._asdict_with_global_options() assert options_asdict["max_instances"] is options.RESET_VALUE, "option should be RESET_VALUE" assert options_asdict["min_instances"] == 5, "option should be set" + assert options_asdict["network_interface"] is options.RESET_VALUE, ( + "network_interface should be RESET_VALUE" + ) + assert options_asdict["vpc_egress"] is options.RESET_VALUE, "vpc_egress should be RESET_VALUE" firebase_functions = { "asamplefunction": asamplefunction, @@ -109,6 +176,7 @@ def test_options_preserve_external_changes(): # where we expect. assert " availableMemoryMb: null\n" in yaml, "availableMemoryMb not in yaml" assert " serviceAccountEmail: null\n" in yaml, "serviceAccountEmail not in yaml" + assert " vpc: null\n" in yaml, "vpc not in yaml" firebase_functions2 = { "asamplefunctionpreserved": asamplefunctionpreserved, @@ -118,6 +186,38 @@ def test_options_preserve_external_changes(): assert " serviceAccountEmail: null\n" not in yaml, "serviceAccountEmail found in yaml" +def test_network_interface_reset_sets_vpc_reset(): + endpoint = options.HttpsOptions( + network_interface=options.RESET_VALUE, + )._endpoint(func_name="test") + + assert endpoint.vpc == options.RESET_VALUE, "vpc should be RESET_VALUE" + + +def test_global_options_support_network_interface(): + previous_options = options._GLOBAL_OPTIONS + try: + options.set_global_options( + network_interface=options.NetworkInterface( + subnetwork="projects/test/regions/us-central1/subnetworks/test" + ), + vpc_egress=options.VpcEgressSetting.ALL_TRAFFIC, + ) + + endpoint = options.HttpsOptions()._endpoint(func_name="test") + + assert endpoint.vpc == { + "networkInterfaces": [ + { + "subnetwork": "projects/test/regions/us-central1/subnetworks/test", + } + ], + "egressSettings": "ALL_TRAFFIC", + } + finally: + options._GLOBAL_OPTIONS = previous_options + + def test_merge_apis_empty_input(): """ This test checks the behavior of the merge_required_apis function From b1089c32f65fda8c28aec55320b5bca03345f606 Mon Sep 17 00:00:00 2001 From: Izaak Gough Date: Wed, 24 Jun 2026 11:39:24 +0100 Subject: [PATCH 2/2] chore: fix formatting --- src/firebase_functions/options.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/src/firebase_functions/options.py b/src/firebase_functions/options.py index 238a3834..33ca774a 100644 --- a/src/firebase_functions/options.py +++ b/src/firebase_functions/options.py @@ -480,11 +480,7 @@ def convert_secret(secret) -> _manifest.SecretEnvironmentVariable: ) if network_interface_dict.get("tags") is not None: network_interface_spec["tags"] = _typing.cast( - str - | list[str] - | Expression[str] - | Expression[list] - | _util.Sentinel, + str | list[str] | Expression[str] | Expression[list] | _util.Sentinel, network_interface_dict["tags"], ) vpc["networkInterfaces"] = [network_interface_spec]