Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 26 additions & 22 deletions bittensor/utils/networking.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Utils for handling local network with ip and ports."""

import os
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

dead import

import subprocess
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why use subprocess when requests is available?

from typing import Optional
from urllib import request as urllib_request

Expand Down Expand Up @@ -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
)
Comment on lines +77 to +79
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

subprocess + curl -> requests.get

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
Expand Down
31 changes: 11 additions & 20 deletions tests/unit_tests/utils/test_networking.py
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Two things missing on the test side:

First: test_get_external_ip_os_broken still patches os.popen, but this PR removes all os.popen usage from production code. That mock now does nothing; the test passes solely because requests.get is mocked to succeed on the first try. The test name and docstring say "when os.popen is broken," but it no longer tests that scenario.
Pls update it to mock subprocess.run failing (or whichever replacement we land on) while keeping requests.get succeeding, so it actually validates "first method works, broken fallback methods don't interfere."

Second: there's no test for the cascade itself. For example: mock requests.get to raise, mock the first subprocess.run call to return a valid IP -> assert the function returns that IP. This is the core contract of get_external_ip() - "try multiple sources in order" - and it has zero coverage today.

Original file line number Diff line number Diff line change
Expand Up @@ -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:
Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

isn't FakeResponse dead now?

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
Expand Down