""" Host IP Detection Utilities Provides robust methods to detect the Docker host IP from within a container, supporting multiple Docker environments and network configurations. """ import os import socket import asyncio import logging from typing import Optional, List from functools import lru_cache import time logger = logging.getLogger(__name__) class HostIPDetector: """Detects the Docker host IP address from container perspective.""" # Common Docker gateway IPs to try as fallbacks COMMON_GATEWAYS = [ "172.17.0.1", # Default Docker bridge "172.18.0.1", # Docker networks "192.168.65.1", # Docker Desktop "192.168.66.1", # Alternative Docker Desktop ] def __init__(self): self._detected_ip: Optional[str] = None self._last_detection: float = 0 self._cache_timeout: float = 300 # 5 minutes cache @lru_cache(maxsize=1) def detect_host_ip(self) -> str: """ Detect the Docker host IP using multiple methods with fallbacks. Returns: str: The detected host IP address Raises: RuntimeError: If no host IP can be detected """ current_time = time.time() # Use cached result if recent if ( self._detected_ip and (current_time - self._last_detection) < self._cache_timeout ): logger.debug(f"Using cached host IP: {self._detected_ip}") return self._detected_ip logger.info("Detecting Docker host IP...") detection_methods = [ self._detect_via_docker_internal, self._detect_via_gateway_env, self._detect_via_route_table, self._detect_via_network_connect, self._detect_via_common_gateways, ] for method in detection_methods: try: ip = method() if ip and self._validate_ip(ip): logger.info( f"Successfully detected host IP using {method.__name__}: {ip}" ) self._detected_ip = ip self._last_detection = current_time return ip else: logger.debug(f"Method {method.__name__} returned invalid IP: {ip}") except Exception as e: logger.debug(f"Method {method.__name__} failed: {e}") # If all methods fail, raise an error raise RuntimeError( "Could not detect Docker host IP. Tried all detection methods. " "Please check your Docker network configuration or set HOST_IP environment variable." ) def _detect_via_docker_internal(self) -> Optional[str]: """Detect via host.docker.internal (Docker Desktop, Docker for Mac/Windows).""" try: # Try to resolve host.docker.internal ip = socket.gethostbyname("host.docker.internal") if ip != "127.0.0.1": # Make sure it's not localhost return ip except socket.gaierror: pass return None def _detect_via_gateway_env(self) -> Optional[str]: """Detect via Docker gateway environment variables.""" # Check common Docker gateway environment variables gateway_vars = [ "DOCKER_HOST_GATEWAY", "GATEWAY", "HOST_IP", ] for var in gateway_vars: ip = os.getenv(var) if ip: logger.debug(f"Found host IP in environment variable {var}: {ip}") return ip return None def _detect_via_route_table(self) -> Optional[str]: """Detect via Linux route table (/proc/net/route).""" try: with open("/proc/net/route", "r") as f: for line in f: fields = line.strip().split() if ( len(fields) >= 8 and fields[0] != "Iface" and fields[7] == "00000000" ): # Found default route, convert hex gateway to IP gateway_hex = fields[2] if len(gateway_hex) == 8: # Convert from hex to IP (little endian) ip_parts = [] for i in range(0, 8, 2): ip_parts.append(str(int(gateway_hex[i : i + 2], 16))) ip = ".".join(reversed(ip_parts)) if ip != "0.0.0.0": return ip except (IOError, ValueError, IndexError) as e: logger.debug(f"Failed to read route table: {e}") return None def _detect_via_network_connect(self) -> Optional[str]: """Detect by attempting to connect to a known external service.""" try: # Try to connect to a reliable external service to determine local IP # We'll use the Docker daemon itself as a reference docker_host = os.getenv("DOCKER_HOST", "tcp://host.docker.internal:2376") if docker_host.startswith("tcp://"): host_part = docker_host[6:].split(":")[0] if host_part not in ["localhost", "127.0.0.1"]: # Try to resolve the host try: ip = socket.gethostbyname(host_part) if ip != "127.0.0.1": return ip except socket.gaierror: pass except Exception as e: logger.debug(f"Network connect detection failed: {e}") return None def _detect_via_common_gateways(self) -> Optional[str]: """Try common Docker gateway IPs.""" for gateway in self.COMMON_GATEWAYS: if self._test_ip_connectivity(gateway): logger.debug(f"Found working gateway: {gateway}") return gateway return None def _test_ip_connectivity(self, ip: str) -> bool: """Test if an IP address is reachable.""" try: # Try to connect to a common port (Docker API or SSH) sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(1.0) result = sock.connect_ex((ip, 22)) # SSH port, commonly available sock.close() return result == 0 except Exception: return False def _validate_ip(self, ip: str) -> bool: """Validate that the IP address is reasonable.""" try: socket.inet_aton(ip) # Basic validation - should not be localhost or invalid ranges if ip.startswith("127."): return False if ip == "0.0.0.0": return False # Should be a private IP range parts = ip.split(".") if len(parts) != 4: return False first_octet = int(parts[0]) # Common Docker gateway ranges return first_octet in [10, 172, 192] except socket.error: return False async def async_detect_host_ip(self) -> str: """Async version of detect_host_ip for testing.""" import asyncio import concurrent.futures loop = asyncio.get_event_loop() with concurrent.futures.ThreadPoolExecutor() as executor: return await loop.run_in_executor(executor, self.detect_host_ip) # Global instance for caching _host_detector = HostIPDetector() def get_host_ip() -> str: """ Get the Docker host IP address from container perspective. This function caches the result for performance and tries multiple detection methods with fallbacks for different Docker environments. Returns: str: The detected host IP address Raises: RuntimeError: If host IP detection fails """ return _host_detector.detect_host_ip() async def async_get_host_ip() -> str: """ Async version of get_host_ip for use in async contexts. Since the actual detection is not async, this just wraps the sync version. """ # Run in thread pool to avoid blocking async context import concurrent.futures import asyncio loop = asyncio.get_event_loop() with concurrent.futures.ThreadPoolExecutor() as executor: return await loop.run_in_executor(executor, get_host_ip) def reset_host_ip_cache(): """Reset the cached host IP detection result.""" global _host_detector _host_detector = HostIPDetector()