From ef03b257622599a95f603cd278f1aa08bca4a3ca Mon Sep 17 00:00:00 2001 From: Peter Adams <18162810+Maxteabag@users.noreply.github.com> Date: Mon, 3 Aug 2026 21:36:26 +0200 Subject: [PATCH] fix(docker): preserve published address family --- .../connections/discovery/docker_detector.py | 57 ++++++++++++++----- tests/test_docker_detector.py | 22 +++++-- 2 files changed, 59 insertions(+), 20 deletions(-) diff --git a/sqlit/domains/connections/discovery/docker_detector.py b/sqlit/domains/connections/discovery/docker_detector.py index ffad336e..0b7937b4 100644 --- a/sqlit/domains/connections/discovery/docker_detector.py +++ b/sqlit/domains/connections/discovery/docker_detector.py @@ -8,6 +8,7 @@ from dataclasses import dataclass from enum import Enum +from ipaddress import ip_address from typing import TYPE_CHECKING, Any, Protocol, cast if TYPE_CHECKING: @@ -121,15 +122,15 @@ def _get_db_type_from_image(image_name: str) -> str | None: return None -def _get_host_port(container: Any, container_port: int) -> int | None: - """Extract the host-mapped port from container port bindings. +def _get_host_binding(container: Any, container_port: int) -> tuple[str, int] | None: + """Extract the host address and port from container port bindings. Args: container: Docker container object container_port: The container's internal port Returns: - Host port number or None if not mapped. + Host address and port, or None if not mapped. """ ports = container.attrs.get("NetworkSettings", {}).get("Ports") or {} @@ -138,29 +139,50 @@ def _get_host_port(container: Any, container_port: int) -> int | None: bindings = ports.get(port_key) if bindings and len(bindings) > 0: - host_port = bindings[0].get("HostPort") + binding = bindings[0] + host_port = binding.get("HostPort") if host_port: - return int(host_port) + return str(binding.get("HostIp") or ""), int(host_port) return None -def _get_single_mapped_host_port(container: Any) -> int | None: - """Return a host port when only one TCP port mapping exists.""" +def _get_host_port(container: Any, container_port: int) -> int | None: + """Extract the host-mapped port from container port bindings.""" + binding = _get_host_binding(container, container_port) + return binding[1] if binding else None + + +def _get_single_mapped_host_binding(container: Any) -> tuple[str, int] | None: + """Return a host binding when only one TCP host port is mapped.""" ports = container.attrs.get("NetworkSettings", {}).get("Ports") or {} - mapped_ports: set[int] = set() + mapped_ports: dict[int, str] = {} for port_key, bindings in ports.items(): if not port_key.endswith("/tcp") or not bindings: continue for binding in bindings: host_port = binding.get("HostPort") if host_port: - mapped_ports.add(int(host_port)) + mapped_ports.setdefault(int(host_port), str(binding.get("HostIp") or "")) if len(mapped_ports) == 1: - return mapped_ports.pop() + host_port, host_ip = next(iter(mapped_ports.items())) + return host_ip, host_port return None +def _resolve_published_host(preferred_host: str, bound_host: str | None) -> str: + """Keep forced-TCP hosts on the address family published by Docker.""" + if preferred_host != "127.0.0.1" or not bound_host: + return preferred_host + try: + address = ip_address(bound_host) + except ValueError: + return bound_host + if address.is_unspecified: + return "::1" if address.version == 6 else preferred_host + return bound_host + + def _get_exposed_tcp_ports(container: Any) -> list[int]: """Return exposed TCP ports declared in the container config.""" exposed = container.attrs.get("Config", {}).get("ExposedPorts") or {} @@ -256,11 +278,16 @@ def _detect_containers_with_status( # Get host-mapped port (only available for running containers) host_port = None + bound_host = None if container_status == ContainerStatus.RUNNING: if default_port: - host_port = _get_host_port(container, default_port) + binding = _get_host_binding(container, default_port) + if binding: + bound_host, host_port = binding if host_port is None: - host_port = _get_single_mapped_host_port(container) + binding = _get_single_mapped_host_binding(container) + if binding: + bound_host, host_port = binding network_mode = container.attrs.get("HostConfig", {}).get("NetworkMode") if host_port is None and network_mode == "host" and default_port: @@ -279,9 +306,9 @@ def _detect_containers_with_status( if container_name.startswith("/"): container_name = container_name[1:] - # Use 127.0.0.1 for MySQL/MariaDB to force TCP connection - # (localhost causes them to try Unix socket which doesn't exist on host) - host = detector.preferred_host + # Providers can choose an explicit loopback to force a network protocol. + # Preserve Docker's concrete binding so its address family stays reachable. + host = _resolve_published_host(detector.preferred_host, bound_host) # For databases that don't require auth, use empty string instead of None # This prevents the UI from prompting for a password diff --git a/tests/test_docker_detector.py b/tests/test_docker_detector.py index 1b37f458..efe6f2aa 100644 --- a/tests/test_docker_detector.py +++ b/tests/test_docker_detector.py @@ -325,8 +325,20 @@ def test_detect_containers_success(self): assert containers[0].database == "testdb" assert containers[0].connectable is True - def test_detected_mssql_container_uses_ipv4_loopback(self): - """Test MSSQL Docker connections use IPv4 loopback.""" + @pytest.mark.parametrize( + ("host_ip", "expected_host"), + [ + ("127.0.0.1", "127.0.0.1"), + ("0.0.0.0", "127.0.0.1"), + ("::1", "::1"), + ("::", "::1"), + ("192.0.2.10", "192.0.2.10"), + ], + ) + def test_detected_mssql_container_uses_published_address_family( + self, host_ip: str, expected_host: str + ): + """Test MSSQL Docker connections remain reachable through their binding.""" mock_container = MagicMock() mock_container.name = "test-mssql" mock_container.short_id = "abc123" @@ -336,7 +348,7 @@ def test_detected_mssql_container_uses_ipv4_loopback(self): "HostConfig": {"NetworkMode": "bridge"}, "NetworkSettings": { "Ports": { - "1433/tcp": [{"HostIp": "127.0.0.1", "HostPort": "1433"}], + "1433/tcp": [{"HostIp": host_ip, "HostPort": "1433"}], } }, } @@ -354,8 +366,8 @@ def test_detected_mssql_container_uses_ipv4_loopback(self): _, containers = detect_database_containers() assert len(containers) == 1 - assert containers[0].host == "127.0.0.1" - assert container_to_connection_config(containers[0]).server == "127.0.0.1" + assert containers[0].host == expected_host + assert container_to_connection_config(containers[0]).server == expected_host def test_detect_container_tagless_image(self): """Test detection when image tags are missing."""