diff --git a/bittensor/utils/networking.py b/bittensor/utils/networking.py index 4558907a73..bd93629142 100644 --- a/bittensor/utils/networking.py +++ b/bittensor/utils/networking.py @@ -1,6 +1,7 @@ """Utils for handling local network with ip and ports.""" import os +import subprocess from typing import Optional from urllib import request as urllib_request @@ -66,55 +67,58 @@ def get_external_ip() -> str: # --- Try AWS try: external_ip = requests.get("https://checkip.amazonaws.com").text.strip() - assert isinstance(ip_to_int(external_ip), int) + ip_to_int(external_ip) return str(external_ip) - except ExternalIPNotFound: + except Exception: pass - # --- Try ipconfig. + # --- Try ifconfig.me try: - process = os.popen("curl -s ifconfig.me") - external_ip = process.readline() - process.close() - assert isinstance(ip_to_int(external_ip), int) + result = subprocess.run( + ["curl", "-s", "ifconfig.me"], capture_output=True, text=True, timeout=10 + ) + external_ip = result.stdout.strip() + ip_to_int(external_ip) return str(external_ip) - except ExternalIPNotFound: + except Exception: pass # --- Try ipinfo. try: - process = os.popen("curl -s https://ipinfo.io") - external_ip = json.loads(process.read())["ip"] - process.close() - assert isinstance(ip_to_int(external_ip), int) + result = subprocess.run( + ["curl", "-s", "https://ipinfo.io"], capture_output=True, text=True, timeout=10 + ) + external_ip = json.loads(result.stdout)["ip"] + ip_to_int(external_ip) return str(external_ip) - except ExternalIPNotFound: + except Exception: pass # --- Try myip.dnsomatic try: - process = os.popen("curl -s myip.dnsomatic.com") - external_ip = process.readline() - process.close() - assert isinstance(ip_to_int(external_ip), int) + result = subprocess.run( + ["curl", "-s", "myip.dnsomatic.com"], capture_output=True, text=True, timeout=10 + ) + external_ip = result.stdout.strip() + ip_to_int(external_ip) return str(external_ip) - except ExternalIPNotFound: + except Exception: pass # --- Try urllib ipv6 try: external_ip = urllib_request.urlopen("https://ident.me").read().decode("utf8") - assert isinstance(ip_to_int(external_ip), int) + ip_to_int(external_ip) return str(external_ip) - except ExternalIPNotFound: + except Exception: pass # --- Try Wikipedia try: external_ip = requests.get("https://www.wikipedia.org").headers["X-Client-IP"] - assert isinstance(ip_to_int(external_ip), int) + ip_to_int(external_ip) return str(external_ip) - except ExternalIPNotFound: + except Exception: pass raise ExternalIPNotFound diff --git a/tests/unit_tests/utils/test_networking.py b/tests/unit_tests/utils/test_networking.py index a3f2c54ac6..8d5c95a5a7 100644 --- a/tests/unit_tests/utils/test_networking.py +++ b/tests/unit_tests/utils/test_networking.py @@ -134,31 +134,22 @@ def mock_call(): def test_get_external_ip_os_request_urllib_broken(): - """Test getting the external IP address when os.popen and requests.get/urllib.request are broken.""" + """Test getting the external IP address when subprocess.run, requests.get, and urllib.request are broken.""" - class FakeReadline: - def readline(self): - return 1 - - def mock_call(): - return FakeReadline() + def mock_subprocess_run(*args, **kwargs): + raise OSError("subprocess.run mocked to fail") class FakeResponse: - def text(self): - return 1 + text = "not-an-ip" - def mock_call_two(): - return FakeResponse() + def mock_requests_get(*args, **kwargs): + raise requests.exceptions.ConnectionError("requests.get mocked to fail") - class FakeRequest: - def urlopen(self): - return 1 - - with mock.patch.object(os, "popen", new=mock_call): - with mock.patch.object(requests, "get", new=mock_call_two): - urllib.request = MagicMock(return_value=FakeRequest()) - with pytest.raises(Exception): - assert utils.networking.get_external_ip() + with mock.patch("subprocess.run", new=mock_subprocess_run): + with mock.patch.object(requests, "get", new=mock_requests_get): + with mock.patch("bittensor.utils.networking.urllib_request.urlopen", side_effect=OSError("urlopen mocked to fail")): + with pytest.raises(Exception): + assert utils.networking.get_external_ip() # Test formatting WebSocket endpoint URL