253 lines
8.5 KiB
Python
253 lines
8.5 KiB
Python
"""
|
|
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()
|