Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Feature: Extended Windows support #731

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
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
3 changes: 3 additions & 0 deletions .gitattributes
Original file line number Diff line number Diff line change
@@ -1,2 +1,5 @@
*.mp3 filter=lfs diff=lfs merge=lfs -text
*.png filter=lfs diff=lfs merge=lfs -text
*.psd filter=lfs diff=lfs merge=lfs -text
*.mp4 filter=lfs diff=lfs merge=lfs -text
*.zip filter=lfs diff=lfs merge=lfs -text
74 changes: 61 additions & 13 deletions exo/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,13 +8,16 @@
import psutil
import uuid
from scapy.all import get_if_addr, get_if_list
from scapy.arch.windows import get_windows_if_list # Windows-specific
import re
import subprocess
from pathlib import Path
import tempfile
import json
from concurrent.futures import ThreadPoolExecutor
import traceback
import netifaces
from typing import List, Tuple

DEBUG = int(os.getenv("DEBUG", default="0"))
DEBUG_DISCOVERY = int(os.getenv("DEBUG_DISCOVERY", default="0"))
Expand Down Expand Up @@ -230,22 +233,67 @@ def pretty_print_bytes_per_second(bytes_per_second: int) -> str:
return f"{bytes_per_second / (1024 ** 4):.2f} TB/s"


def get_all_ip_addresses_and_interfaces():
def get_all_ip_addresses_and_interfaces() -> List[Tuple[str, str]]:
"""
Get all active IP addresses and their corresponding interfaces.
Excludes loopback, non-routable, and inactive interfaces.
"""
ip_addresses = []
for interface in get_if_list():
try:
ip = get_if_addr(interface)
if ip.startswith("0.0."): continue
simplified_interface = re.sub(r'^\\Device\\NPF_', '', interface)
ip_addresses.append((ip, simplified_interface))
except:
if DEBUG >= 1: print(f"Failed to get IP address for interface {interface}")
if DEBUG >= 1: traceback.print_exc()

# Exclude these IP prefixes (non-routable or problematic)
excluded_prefixes = (
"127.", # Loopback
"169.254.", # Link-local
"192.168.137.", # Hyper-V, VMware, etc.
"0.0.0.0", # Invalid
)

# Get all network interfaces
interfaces = netifaces.interfaces()

for interface in interfaces:
try:
# Get IPv4 addresses for the interface
addrs = netifaces.ifaddresses(interface).get(netifaces.AF_INET, [])
for addr in addrs:
ip = addr.get("addr")
if not ip:
continue

# Skip excluded IPs
if any(ip.startswith(prefix) for prefix in excluded_prefixes):
if DEBUG >= 1:
print(f"Skipping excluded IP: {ip} on interface {interface}")
continue

# On Windows, check if the interface is active
if psutil.WINDOWS:
try:
# Use psutil to check interface status
if_stats = psutil.net_if_stats().get(interface)
if if_stats and not if_stats.isup:
if DEBUG >= 1:
print(f"Skipping inactive interface: {interface}")
continue
except Exception as e:
if DEBUG >= 1:
print(f"Error checking interface status for {interface}: {e}")

# Add the IP and interface to the list
ip_addresses.append((ip, interface))

except Exception as e:
if DEBUG >= 1:
print(f"Error processing interface {interface}: {e}")
traceback.print_exc()

# If no valid IPs are found, default to localhost
if not ip_addresses:
if DEBUG >= 1: print("Failed to get any IP addresses. Defaulting to localhost.")
return [("localhost", "lo")]
return list(set(ip_addresses))
if DEBUG >= 1:
print("No valid IP addresses found. Defaulting to localhost.")
return [("127.0.0.1", "lo")]

return list(set(ip_addresses)) # Remove duplicates


async def get_macos_interface_type(ifname: str) -> Optional[Tuple[int, str]]:
Expand Down
Loading