diff --git a/bittensor/utils/networking.py b/bittensor/utils/networking.py index 4558907a73..ede3e0ff5c 100644 --- a/bittensor/utils/networking.py +++ b/bittensor/utils/networking.py @@ -21,6 +21,9 @@ def int_to_ip(int_val: int) -> str: Returns: str_val: The string representation of an ip. Of form *.*.*.* for ipv4 or *::*:*:*:* for ipv6 + + Raises: + netaddr.AddrFormatError: If the integer value is out of valid IP range. """ return str(netaddr.IPAddress(int_val)) @@ -33,6 +36,9 @@ def ip_to_int(str_val: str) -> int: Returns: int_val: The integer representation of an ip. Must be in the range (0, 3.4028237e+38). + + Raises: + netaddr.AddrFormatError: If the string is not a valid IP address. """ return int(netaddr.IPAddress(str_val)) @@ -45,30 +51,63 @@ def ip_version(str_val: str) -> int: Returns: int_val: The ip version (Either 4 or 6 for IPv4/IPv6) + + Raises: + netaddr.AddrFormatError: If the string is not a valid IP address. """ return int(netaddr.IPAddress(str_val).version) -def ip__str__(ip_type: int, ip_str: str, port: int): - """Return a formatted ip string""" +def ip__str__(ip_type: int, ip_str: str, port: int) -> str: + """Return a formatted ip string + + Parameters: + ip_type: The IP version (4 or 6). + ip_str: The IP address as a string. + port: The port number. + + Returns: + A formatted string of form /ipv/: + """ return "/ipv%i/%s:%i" % (ip_type, ip_str, port) +def _validate_ip_response(ip_str: str) -> str: + """Validate and normalize an IP address string obtained from an external service. + + Parameters: + ip_str: The IP address string to validate. + + Returns: + The validated IP address as a string. + + Raises: + ValueError: If the string is not a valid IP address. + """ + ip_str = ip_str.strip() + ip_to_int(ip_str) # raises netaddr.AddrFormatError if invalid + return ip_str + + def get_external_ip() -> str: """Checks CURL/URLLIB/IPIFY/AWS for your external ip. + Tries multiple external services in sequence. If one fails (due to network + errors, invalid responses, timeouts, etc.), it falls through to the next. + Returns: external_ip (str): Your routers external facing ip as a string. Raises: - ExternalIPNotFound(Exception): Raised if all external ip attempts fail. + ExternalIPNotFound: Raised if all external ip attempts fail. """ # --- Try AWS try: - external_ip = requests.get("https://checkip.amazonaws.com").text.strip() - assert isinstance(ip_to_int(external_ip), int) - return str(external_ip) - except ExternalIPNotFound: + external_ip = requests.get( + "https://checkip.amazonaws.com", timeout=5 + ).text.strip() + return _validate_ip_response(external_ip) + except Exception: pass # --- Try ipconfig. @@ -76,9 +115,8 @@ def get_external_ip() -> str: process = os.popen("curl -s ifconfig.me") external_ip = process.readline() process.close() - assert isinstance(ip_to_int(external_ip), int) - return str(external_ip) - except ExternalIPNotFound: + return _validate_ip_response(external_ip) + except Exception: pass # --- Try ipinfo. @@ -86,9 +124,8 @@ def get_external_ip() -> str: 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) - return str(external_ip) - except ExternalIPNotFound: + return _validate_ip_response(external_ip) + except Exception: pass # --- Try myip.dnsomatic @@ -96,25 +133,26 @@ def get_external_ip() -> str: process = os.popen("curl -s myip.dnsomatic.com") external_ip = process.readline() process.close() - assert isinstance(ip_to_int(external_ip), int) - return str(external_ip) - except ExternalIPNotFound: + return _validate_ip_response(external_ip) + 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) - return str(external_ip) - except ExternalIPNotFound: + external_ip = urllib_request.urlopen( + "https://ident.me", timeout=5 + ).read().decode("utf8") + return _validate_ip_response(external_ip) + 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) - return str(external_ip) - except ExternalIPNotFound: + external_ip = requests.get( + "https://www.wikipedia.org", timeout=5 + ).headers["X-Client-IP"] + return _validate_ip_response(external_ip) + 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..554f46250b 100644 --- a/tests/unit_tests/utils/test_networking.py +++ b/tests/unit_tests/utils/test_networking.py @@ -4,7 +4,18 @@ import requests import unittest.mock as mock from bittensor import utils +from bittensor.utils.networking import ( + _validate_ip_response, + ExternalIPNotFound, + get_external_ip, + get_formatted_ws_endpoint_url, + int_to_ip, + ip_to_int, + ip_version, + ip__str__, +) from unittest.mock import MagicMock +import netaddr # Test conversion functions for IPv4 @@ -82,6 +93,112 @@ def test_int_to_ip6_underflow(): utils.networking.int_to_ip(underflow) +# Tests for ip_version +class TestIpVersion: + """Tests for the ip_version function.""" + + def test_ipv4_version(self): + """Test that IPv4 addresses return version 4.""" + assert ip_version("192.168.1.1") == 4 + assert ip_version("10.0.0.1") == 4 + assert ip_version("0.0.0.0") == 4 + assert ip_version("255.255.255.255") == 4 + + def test_ipv6_version(self): + """Test that IPv6 addresses return version 6.""" + assert ip_version("::1") == 6 + assert ip_version("fe80::1") == 6 + assert ip_version("ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff") == 6 + + def test_invalid_ip_raises(self): + """Test that invalid IP strings raise an error.""" + with pytest.raises(netaddr.AddrFormatError): + ip_version("not_an_ip") + with pytest.raises(netaddr.AddrFormatError): + ip_version("") + with pytest.raises(netaddr.AddrFormatError): + ip_version("999.999.999.999") + + +# Tests for ip_to_int with invalid input +class TestIpToIntValidation: + """Tests for ip_to_int input validation.""" + + def test_invalid_ip_string_raises(self): + """Test that invalid IP strings raise netaddr.AddrFormatError.""" + with pytest.raises(netaddr.AddrFormatError): + ip_to_int("not_an_ip") + + def test_empty_string_raises(self): + """Test that empty string raises netaddr.AddrFormatError.""" + with pytest.raises(netaddr.AddrFormatError): + ip_to_int("") + + def test_partial_ip_raises(self): + """Test that partial IP addresses raise an error.""" + with pytest.raises(netaddr.AddrFormatError): + ip_to_int("192.168.1") + + +# Tests for ip__str__ return type +class TestIpStrFormat: + """Tests for the ip__str__ function.""" + + def test_return_type_is_str(self): + """Test that ip__str__ always returns a string.""" + result = ip__str__(4, "127.0.0.1", 8080) + assert isinstance(result, str) + + def test_ipv6_format(self): + """Test IPv6 formatted string.""" + result = ip__str__(6, "::1", 9944) + assert result == "/ipv6/::1:9944" + + def test_port_zero(self): + """Test with port 0.""" + result = ip__str__(4, "10.0.0.1", 0) + assert result == "/ipv4/10.0.0.1:0" + + def test_high_port(self): + """Test with high port number.""" + result = ip__str__(4, "10.0.0.1", 65535) + assert result == "/ipv4/10.0.0.1:65535" + + +# Tests for _validate_ip_response +class TestValidateIpResponse: + """Tests for the _validate_ip_response helper function.""" + + def test_valid_ipv4(self): + """Test validation with a valid IPv4 address.""" + assert _validate_ip_response("192.168.1.1") == "192.168.1.1" + + def test_valid_ipv4_with_whitespace(self): + """Test that leading/trailing whitespace is stripped.""" + assert _validate_ip_response(" 10.0.0.1 ") == "10.0.0.1" + assert _validate_ip_response("\n203.0.113.5\n") == "203.0.113.5" + + def test_valid_ipv6(self): + """Test validation with a valid IPv6 address.""" + assert _validate_ip_response("::1") == "::1" + assert _validate_ip_response("2001:db8::1") == "2001:db8::1" + + def test_invalid_ip_raises(self): + """Test that invalid IP strings raise an error.""" + with pytest.raises(Exception): + _validate_ip_response("not_an_ip") + + def test_empty_string_raises(self): + """Test that empty string raises an error.""" + with pytest.raises(Exception): + _validate_ip_response("") + + def test_html_response_raises(self): + """Test that HTML responses (common failure mode) raise an error.""" + with pytest.raises(Exception): + _validate_ip_response("Error") + + # Test getting external IP address def test_get_external_ip(mocker): """Test getting the external IP address.""" @@ -101,7 +218,9 @@ def test_get_external_ip(mocker): assert utils.networking.get_external_ip() == "192.168.1.1" - mocked_requests_get.assert_called_once_with("https://checkip.amazonaws.com") + mocked_requests_get.assert_called_once_with( + "https://checkip.amazonaws.com", timeout=5 + ) def test_get_external_ip_os_broken(mocker): @@ -130,35 +249,115 @@ def mock_call(): with mock.patch.object(os, "popen", new=mock_call): assert utils.networking.get_external_ip() == "192.168.1.1" - mocked_requests_get.assert_called_once_with("https://checkip.amazonaws.com") - - -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.""" - - class FakeReadline: - def readline(self): - return 1 - - def mock_call(): - return FakeReadline() - - class FakeResponse: - def text(self): - return 1 + mocked_requests_get.assert_called_once_with( + "https://checkip.amazonaws.com", timeout=5 + ) - def mock_call_two(): - return FakeResponse() - 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() +def test_get_external_ip_all_broken_raises(mocker): + """Test that ExternalIPNotFound is raised when all services return invalid data.""" + mocker.patch.object( + requests, + "get", + side_effect=requests.exceptions.ConnectionError("no network"), + ) + mocker.patch.object( + os, "popen", side_effect=OSError("no curl") + ) + mocker.patch( + "bittensor.utils.networking.urllib_request.urlopen", + side_effect=OSError("no urllib"), + ) + with pytest.raises(ExternalIPNotFound): + get_external_ip() + + +class TestGetExternalIpFallthrough: + """Tests for get_external_ip fallthrough behavior when services fail.""" + + def test_aws_connection_error_falls_through(self, mocker): + """Test that a ConnectionError from AWS falls through to the next service. + + This was a bug where ExternalIPNotFound was caught instead of Exception, + meaning real network errors would crash instead of falling through. + """ + # requests.get is called for AWS (1st) and Wikipedia (2nd) + mocker.patch.object( + requests, + "get", + side_effect=[ + requests.exceptions.ConnectionError("AWS down"), + mock.Mock(headers={"X-Client-IP": "203.0.113.5"}), + ], + ) + # Mock os.popen to fail (3 curl calls: ifconfig.me, ipinfo.io, dnsomatic) + mocker.patch.object( + os, "popen", side_effect=OSError("curl not found") + ) + # Mock urllib to fail + mocker.patch( + "bittensor.utils.networking.urllib_request.urlopen", + side_effect=OSError("network error"), + ) + result = get_external_ip() + assert result == "203.0.113.5" + + def test_all_services_fail_raises_external_ip_not_found(self, mocker): + """Test that ExternalIPNotFound is raised when all services fail.""" + mocker.patch.object( + requests, + "get", + side_effect=requests.exceptions.ConnectionError("no network"), + ) + mocker.patch.object( + os, "popen", side_effect=OSError("curl not found") + ) + mocker.patch( + "bittensor.utils.networking.urllib_request.urlopen", + side_effect=OSError("network error"), + ) + with pytest.raises(ExternalIPNotFound): + get_external_ip() + + def test_aws_returns_invalid_ip_falls_through(self, mocker): + """Test that an invalid IP response from AWS falls through.""" + mocker.patch.object( + requests, + "get", + side_effect=[ + mock.Mock(text="Error Page"), # AWS returns HTML error + mock.Mock(headers={"X-Client-IP": "198.51.100.1"}), # Wikipedia works + ], + ) + mocker.patch.object( + os, "popen", side_effect=OSError("curl not found") + ) + mocker.patch( + "bittensor.utils.networking.urllib_request.urlopen", + side_effect=OSError("network error"), + ) + result = get_external_ip() + assert result == "198.51.100.1" + + def test_timeout_error_falls_through(self, mocker): + """Test that a timeout from the first service falls through.""" + mocker.patch.object( + requests, + "get", + side_effect=[ + requests.exceptions.Timeout("timed out"), + mock.Mock(headers={"X-Client-IP": "198.51.100.2"}), + ], + ) + mocker.patch.object( + os, "popen", side_effect=OSError("curl not found") + ) + mocker.patch( + "bittensor.utils.networking.urllib_request.urlopen", + side_effect=OSError("network error"), + ) + result = get_external_ip() + assert result == "198.51.100.2" # Test formatting WebSocket endpoint URL @@ -196,3 +395,25 @@ def urlopen(self): def test_format(url: str, expected: str): """Test formatting WebSocket endpoint URL.""" assert utils.networking.get_formatted_ws_endpoint_url(url) == expected + + +class TestGetFormattedWsEndpointUrlEdgeCases: + """Edge case tests for get_formatted_ws_endpoint_url.""" + + def test_none_returns_none(self): + """Test that None input returns None.""" + assert get_formatted_ws_endpoint_url(None) is None + + def test_empty_string_prepends_ws(self): + """Test that empty string gets ws:// prepended.""" + assert get_formatted_ws_endpoint_url("") == "ws://" + + def test_preserves_path(self): + """Test that paths are preserved.""" + result = get_formatted_ws_endpoint_url("ws://host:9944/ws") + assert result == "ws://host:9944/ws" + + def test_wss_not_modified(self): + """Test that wss:// URLs are not modified.""" + result = get_formatted_ws_endpoint_url("wss://secure.host:443") + assert result == "wss://secure.host:443"