feat: add support for VPC direct connect

IzaakGough · IzaakGough · commit 4eb5e9eb23bf · 2026-06-24T11:37:35.000+01:00
diff --git 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
@@ -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
@@ -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
