Add isolated physical network access for apps

supervisor/api/apps.py

@@ -2,6 +2,7 @@
 
 import asyncio
 from collections.abc import Awaitable
+from contextlib import suppress
 import logging
 from typing import Any, TypedDict
 
@@ -11,6 +12,7 @@
 
 from ..apps.app import App
 from ..apps.utils import rating_security
+from ..apps.validate import SCHEMA_NETWORK_ISOLATION
 from ..const import (
     ATTR_ADDONS,
     ATTR_ADVANCED,
@@ -37,6 +39,7 @@
     ATTR_DNS,
     ATTR_DOCKER_API,
     ATTR_DOCUMENTATION,
+    ATTR_DRIVER,
     ATTR_FORCE,
     ATTR_FULL_ACCESS,
     ATTR_GPIO,
@@ -67,6 +70,7 @@
     ATTR_NAME,
     ATTR_NETWORK,
     ATTR_NETWORK_DESCRIPTION,
+    ATTR_NETWORK_ISOLATION,
     ATTR_NETWORK_RX,
     ATTR_NETWORK_TX,
     ATTR_OPTIONS,
@@ -99,6 +103,11 @@
     AppBootConfig,
 )
 from ..coresys import CoreSysAttributes
+from ..docker.const import ExternalNetworkDriver, NetworkIsolationConfig
+from ..docker.external_network import (
+    MIN_EXTERNAL_NETWORK_DOCKER,
+    DockerExternalNetworks,
+)
 from ..docker.stats import DockerStats
 from ..exceptions import (
     APIAppNotInstalled,
@@ -107,23 +116,47 @@
     APINotFound,
     AppBootConfigCannotChangeError,
     AppConfigurationInvalidError,
+    AppNetworkIsolationDockerVersionError,
+    AppNetworkIsolationInvalidAddressError,
+    AppNetworkIsolationInvalidInterfaceError,
+    AppNetworkIsolationNotSupportedError,
     AppNotSupportedWriteStdinError,
+    DockerError,
+    HostNetworkNotFound,
     PwnedError,
     PwnedSecret,
 )
 from ..validate import docker_ports
-from .const import ATTR_BOOT_CONFIG, ATTR_REMOVE_CONFIG, ATTR_SIGNED
+from .const import (
+    ATTR_BOOT_CONFIG,
+    ATTR_NETWORK_ISOLATION_AVAILABLE,
+    ATTR_NETWORK_ISOLATION_MAC,
+    ATTR_REMOVE_CONFIG,
+    ATTR_SIGNED,
+)
 from .utils import api_process, api_validate, json_loads
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 
 SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): str})
 
+# Only the macvlan driver is supported for now
+SCHEMA_NETWORK_ISOLATION_OPTIONS = SCHEMA_NETWORK_ISOLATION.extend(
+    {
+        vol.Optional(ATTR_DRIVER, default=ExternalNetworkDriver.MACVLAN): vol.All(
+            vol.Coerce(ExternalNetworkDriver), vol.In([ExternalNetworkDriver.MACVLAN])
+        ),
+    }
+)
+
 # pylint: disable=no-value-for-parameter
 SCHEMA_OPTIONS = vol.Schema(
     {
         vol.Optional(ATTR_BOOT): vol.Coerce(AppBoot),
         vol.Optional(ATTR_NETWORK): vol.Maybe(docker_ports),
+        vol.Optional(ATTR_NETWORK_ISOLATION): vol.Maybe(
+            SCHEMA_NETWORK_ISOLATION_OPTIONS
+        ),
         vol.Optional(ATTR_AUTO_UPDATE): vol.Boolean(),
         vol.Optional(ATTR_AUDIO_OUTPUT): vol.Maybe(str),
         vol.Optional(ATTR_AUDIO_INPUT): vol.Maybe(str),
@@ -249,6 +282,12 @@ async def info_data(self, app: App) -> dict[str, Any]:
             ATTR_BUILD: app.need_build,
             ATTR_NETWORK: app.ports,
             ATTR_NETWORK_DESCRIPTION: app.ports_description,
+            ATTR_NETWORK_ISOLATION: config.to_dict()
+            if (config := app.network_isolation)
+            else None,
+            ATTR_NETWORK_ISOLATION_AVAILABLE: app.host_network
+            and self.sys_docker.external_networks.available,
+            ATTR_NETWORK_ISOLATION_MAC: app.external_mac_address,
             ATTR_HOST_NETWORK: app.host_network,
             ATTR_HOST_PID: app.host_pid,
             ATTR_HOST_IPC: app.host_ipc,
@@ -337,6 +376,13 @@ async def options(self, request: web.Request) -> None:
             app.auto_update = body[ATTR_AUTO_UPDATE]
         if ATTR_NETWORK in body:
             app.ports = body[ATTR_NETWORK]
+        if ATTR_NETWORK_ISOLATION in body:
+            if (isolation := body[ATTR_NETWORK_ISOLATION]) is None:
+                app.network_isolation = None
+            else:
+                config = NetworkIsolationConfig.from_dict(isolation)
+                self._validate_network_isolation(app, config)
+                app.network_isolation = config
         if ATTR_AUDIO_INPUT in body:
             app.audio_input = body[ATTR_AUDIO_INPUT]
         if ATTR_AUDIO_OUTPUT in body:
@@ -349,6 +395,72 @@ async def options(self, request: web.Request) -> None:
 
         await app.save_persist()
 
+        # Clean up external networks no longer referenced by any app
+        if ATTR_NETWORK_ISOLATION in body:
+            with suppress(DockerError):
+                await self.sys_docker.external_networks.gc()
+
+    def _validate_network_isolation(
+        self, app: App, config: NetworkIsolationConfig
+    ) -> None:
+        """Validate a network isolation config against app and host state."""
+        if not app.host_network:
+            raise AppNetworkIsolationNotSupportedError(app=app)
+        if not self.sys_docker.external_networks.available:
+            raise AppNetworkIsolationDockerVersionError(
+                app=app, minimum_version=str(MIN_EXTERNAL_NETWORK_DOCKER)
+            )
+
+        try:
+            interface = self.sys_host.network.get(config.interface)
+        except HostNetworkNotFound:
+            raise AppNetworkIsolationInvalidInterfaceError(
+                app=app, interface=config.interface, reason="interface not found"
+            ) from None
+        if not DockerExternalNetworks.capable_interface(interface):
+            raise AppNetworkIsolationInvalidInterfaceError(
+                app=app,
+                interface=config.interface,
+                reason="must be a connected ethernet interface with IPv4",
+            )
+
+        subnet = DockerExternalNetworks.interface_subnet(interface)
+        if (
+            subnet is None
+            or config.ipv4 not in subnet
+            or config.ipv4 in {subnet.network_address, subnet.broadcast_address}
+        ):
+            raise AppNetworkIsolationInvalidAddressError(
+                app=app,
+                address=str(config.ipv4),
+                reason=f"not a usable address within subnet {subnet}",
+            )
+        if interface.ipv4 and (
+            config.ipv4 == interface.ipv4.gateway
+            or any(
+                config.ipv4 == address.ip
+                for address in interface.ipv4.address
+                if address.version == 4
+            )
+        ):
+            raise AppNetworkIsolationInvalidAddressError(
+                app=app,
+                address=str(config.ipv4),
+                reason="already used by the host or its gateway",
+            )
+
+        for other in self.sys_apps.installed:
+            if other.slug == app.slug:
+                continue
+            if (
+                other_config := other.network_isolation
+            ) and other_config.ipv4 == config.ipv4:
+                raise AppNetworkIsolationInvalidAddressError(
+                    app=app,
+                    address=str(config.ipv4),
+                    reason=f"already assigned to app {other.slug}",
+                )
+
     @api_process
     async def sys_options(self, request: web.Request) -> None:
         """Store system options for an app."""

supervisor/api/const.py

@@ -54,6 +54,9 @@
 ATTR_MODEL = "model"
 ATTR_MOUNTS = "mounts"
 ATTR_MOUNT_POINTS = "mount_points"
+ATTR_NETWORK_ISOLATION_AVAILABLE = "network_isolation_available"
+ATTR_NETWORK_ISOLATION_CAPABLE = "network_isolation_capable"
+ATTR_NETWORK_ISOLATION_MAC = "network_isolation_mac"
 ATTR_PANEL_PATH = "panel_path"
 ATTR_REMOVABLE = "removable"
 ATTR_REMOVE_CONFIG = "remove_config"

supervisor/api/network.py

@@ -47,6 +47,7 @@
     DOCKER_NETWORK,
 )
 from ..coresys import CoreSysAttributes
+from ..docker.external_network import DockerExternalNetworks
 from ..exceptions import APIError, APINotFound, HostNetworkNotFound
 from ..host.configuration import (
     AccessPoint,
@@ -62,6 +63,7 @@
     WifiConfig,
 )
 from ..host.const import AuthMethod, InterfaceType, WifiMode
+from .const import ATTR_NETWORK_ISOLATION_CAPABLE
 from .utils import api_process, api_validate
 
 _SCHEMA_IPV4_CONFIG = vol.Schema(
@@ -172,6 +174,9 @@ def interface_struct(interface: Interface) -> dict[str, Any]:
         ATTR_VLAN: vlan_struct(interface.vlan) if interface.vlan else None,
         ATTR_MDNS: interface.mdns,
         ATTR_LLMNR: interface.llmnr,
+        ATTR_NETWORK_ISOLATION_CAPABLE: DockerExternalNetworks.capable_interface(
+            interface
+        ),
     }
 
 

supervisor/apps/app.py

@@ -37,6 +37,7 @@
     ATTR_INGRESS_TOKEN,
     ATTR_LOCATION,
     ATTR_NETWORK,
+    ATTR_NETWORK_ISOLATION,
     ATTR_OPTIONS,
     ATTR_PORTS,
     ATTR_PROTECTED,
@@ -60,7 +61,12 @@
 )
 from ..coresys import CoreSys
 from ..docker.app import DockerApp
-from ..docker.const import EXIT_CODE_SIGTERM_DEFAULT, ContainerState
+from ..docker.const import (
+    EXIT_CODE_SIGTERM_DEFAULT,
+    ContainerState,
+    NetworkIsolationConfig,
+)
+from ..docker.external_network import DockerExternalNetworks
 from ..docker.manager import ExecReturn
 from ..docker.monitor import DockerContainerStateEvent
 from ..docker.stats import DockerStats
@@ -164,6 +170,9 @@ def __init__(self, coresys: CoreSys, slug: str):
         self._device_access_missing_issue = Issue(
             IssueType.DEVICE_ACCESS_MISSING, ContextType.ADDON, reference=self.slug
         )
+        self._network_isolation_failed_issue = Issue(
+            IssueType.NETWORK_ISOLATION_FAILED, ContextType.ADDON, reference=self.slug
+        )
 
     def __repr__(self) -> str:
         """Return internal representation."""
@@ -179,6 +188,11 @@ def device_access_missing_issue(self) -> Issue:
         """Get issue used if device access is missing and can't be automatically added."""
         return self._device_access_missing_issue
 
+    @property
+    def network_isolation_failed_issue(self) -> Issue:
+        """Get issue used if the isolated network endpoint can't be set up."""
+        return self._network_isolation_failed_issue
+
     @property
     def state(self) -> AppState:
         """Return current state of the app."""
@@ -352,6 +366,17 @@ def ip_address(self) -> IPv4Address:
         """Return IP of app instance."""
         return self.instance.ip_address
 
+    @property
+    def external_mac_address(self) -> str | None:
+        """Return MAC address of the isolated network endpoint, if assigned.
+
+        Derived from the static IP, so it is known without a running
+        container (the endpoint is connected with exactly this MAC).
+        """
+        if not (config := self.network_isolation):
+            return None
+        return DockerExternalNetworks.mac_from_ip(config.ipv4)
+
     @property
     def data(self) -> Data:
         """Return app data/config."""
@@ -595,6 +620,22 @@ def ports(self, value: dict[str, int | None] | None) -> None:
 
         self.persist[ATTR_NETWORK] = new_ports
 
+    @property
+    def network_isolation(self) -> NetworkIsolationConfig | None:
+        """Return isolated physical network endpoint assigned by the user."""
+        if not (data := self.persist.get(ATTR_NETWORK_ISOLATION)):
+            return None
+        return NetworkIsolationConfig.from_dict(data)
+
+    @network_isolation.setter
+    def network_isolation(self, value: NetworkIsolationConfig | None) -> None:
+        """Assign or clear the isolated physical network endpoint."""
+        if value is None:
+            self.persist.pop(ATTR_NETWORK_ISOLATION, None)
+            return
+
+        self.persist[ATTR_NETWORK_ISOLATION] = value.to_dict()
+
     @property
     def ingress_url(self) -> str | None:
         """Return URL to ingress url."""
@@ -993,9 +1034,15 @@ def cleanup_config_and_audio():
             await service.del_service_data(self)
 
         # Remove from app manager
+        had_network_isolation = self.network_isolation is not None
         self.sys_apps.local.pop(self.slug)
         await self.sys_apps.data.uninstall(self)
 
+        # Clean up external networks no longer referenced by any app
+        if had_network_isolation:
+            with suppress(DockerError):
+                await self.sys_docker.external_networks.gc()
+
         # Cleanup Ingress tokens
         if need_ingress_token_cleanup:
             await self.sys_ingress.reload()

supervisor/apps/manager.py

@@ -126,8 +126,11 @@ async def boot(self, stage: AppStartup) -> None:
         for app in self.installed:
             if app.boot != AppBoot.AUTO or app.startup != stage:
                 continue
+            # Apps with an isolated network endpoint do not run in the host
+            # network namespace, their traffic never passes the Docker gateway
             if (
                 app.host_network
+                and not app.network_isolation
                 and UnhealthyReason.DOCKER_GATEWAY_UNPROTECTED
                 in self.sys_resolution.unhealthy
             ):

supervisor/apps/model.py

@@ -89,7 +89,7 @@
     CpuArch,
 )
 from ..coresys import CoreSys
-from ..docker.const import Capabilities
+from ..docker.const import Capabilities, NetworkIsolationConfig
 from ..exceptions import (
     AppFileReadError,
     AppNotSupportedArchitectureError,
@@ -333,6 +333,14 @@ def host_network(self) -> bool:
         """Return True if app run on host network."""
         return self.data[ATTR_HOST_NETWORK]
 
+    @property
+    def network_isolation(self) -> NetworkIsolationConfig | None:
+        """Return isolated physical network endpoint assigned to the app.
+
+        Only installed apps can have one assigned (user setting).
+        """
+        return None
+
     @property
     def host_pid(self) -> bool:
         """Return True if app run on host PID namespace."""

supervisor/apps/utils.py

@@ -67,8 +67,9 @@ def rating_security(app: AppModel) -> int:
     elif app.hassio_role == ROLE_ADMIN:
         rating += -2
 
-    # Not secure Networking
-    if app.host_network:
+    # Not secure Networking. With an isolated physical network endpoint
+    # assigned the app does not run in the host network namespace.
+    if app.host_network and not app.network_isolation:
         rating += -1
 
     # Insecure PID namespace