clearml-agent/clearml_agent/helper/sdk_client/utilities/networking.py

import requests
import socket
import subprocess
from typing import Optional


def get_private_ip(config_obj):
    # type: (Config) -> str
    """
    Get the private IP of this machine

    :return: A string representing the IP of this machine
    """
    approaches = (
        _get_private_ip_from_socket,
        _get_private_ip_from_subprocess,
    )

    for approach in approaches:
        # noinspection PyBroadException
        try:
            return approach(config_obj)
        except Exception:
            continue

    raise Exception("error getting private IP")


def get_public_ip(config_obj):
    # type: (Config) -> Optional[str]
    """
    Get the public IP of this machine. External services such as `https://api.ipify.org` or `https://ident.me`
    are used to get the IP

    :return: A string representing the IP of this machine or `None` if getting the IP failed
    """

    # todo: add documentation in api section in conf file
    public_ip_service_urls = (
            (config_obj.get("api.public_ip_service_urls", None) if config_obj else None)
            or ["https://api.ipify.org", "https://ident.me"]
    )
    for external_service in public_ip_service_urls:
        ip = get_public_ip_from_external_service(external_service)
        if ip:
            return ip
    return None


def get_public_ip_from_external_service(external_service, timeout=5):
    # type: (str, Optional[int]) -> Optional[str]
    """
    Get the public IP of this machine from an external service.
    Fetching the IP is done via a GET request. The whole content of the request
    should be the IP address

    :param external_service: The address of the external service
    :param timeout: The GET request timeout

    :return: A string representing the IP of this machine or `None` if getting the IP failed
    """
    # noinspection PyBroadException
    try:
        response = requests.get(external_service, timeout=timeout)
        if not response.ok:
            return None
        ip = response.content.decode("utf8")
        # check that we actually received an IP address
        # noinspection PyBroadException
        try:
            socket.inet_pton(socket.AF_INET, ip)
            return ip
        except Exception:
            socket.inet_pton(socket.AF_INET6, ip)
            return ip
    except Exception:
        return None


def _get_private_ip_from_socket(config_obj):

    # todo: add documentation in api section in conf file
    public_ip_ping = (config_obj.get("api.public_ip_ping", None) if config_obj else None) or "8.8.8.8"

    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.settimeout(0)
    try:
        s.connect((public_ip_ping, 1))
        ip = s.getsockname()[0]
    except Exception as e:
        raise e
    finally:
        s.close()
    return ip


def _get_private_ip_from_subprocess(_):
    return subprocess.check_output("hostname -I", shell=True).split()[0].decode("utf-8")