1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
|
"""
transport/ice_filter — the interface filter must name the same adapter on
every platform, and must never leave the gather with no address at all.
"""
import aioice.ice
import ifaddr
import pytest
from meshbay_node.transport import ice_filter
class _IP:
def __init__(self, ip):
self.ip = ip
class _Adapter:
def __init__(self, name, nice_name, ips):
self.name, self.nice_name = name, nice_name
self.ips = [_IP(ip) for ip in ips]
def _linux(name, *ips):
"""On Linux ifaddr repeats the kernel name in nice_name."""
return _Adapter(name, name, list(ips))
def _windows(guid, description, *ips):
"""On Windows `name` is a GUID and `nice_name` the device description."""
return _Adapter(guid, description, list(ips))
@pytest.fixture(autouse=True)
def _restore():
saved = aioice.ice.get_host_addresses
yield
aioice.ice.get_host_addresses = saved
@pytest.fixture
def adapters(monkeypatch):
"""Install a fake adapter set; returns a setter the test calls."""
def _set(*adapters):
monkeypatch.setattr(ifaddr, "get_adapters", lambda: list(adapters))
return _set
def _gather(include=None):
ice_filter.install(include)
return aioice.ice.get_host_addresses()
def test_auto_excludes_virtual_adapters(adapters):
adapters(
_linux("lo", "127.0.0.1"),
_linux("wlp3s0f0", "192.168.1.22"),
_linux("virbr0", "192.168.200.254"),
_linux("docker0", "172.17.0.1"),
)
assert _gather() == ["192.168.1.22"]
def test_auto_drops_cgnat_range(adapters):
adapters(
_linux("eth0", "192.168.1.22"),
_linux("tun0", "100.101.102.103"),
)
assert _gather() == ["192.168.1.22"]
def test_include_list_by_kernel_name(adapters):
adapters(
_linux("eth0", "192.168.1.22"),
_linux("wlp3s0f0", "192.168.1.23"),
)
assert _gather(["wlp3s0f0"]) == ["192.168.1.23"]
def test_include_list_keeps_explicitly_named_cgnat_adapter(adapters):
"""An explicit name beats the auto-exclusion rules."""
adapters(_linux("eth0", "192.168.1.22"), _linux("tun0", "100.101.102.103"))
assert _gather(["tun0"]) == ["100.101.102.103"]
def test_include_list_matches_windows_description(adapters):
"""
The Windows adapter GUID is not something an operator ever types, so the
device description has to match too.
"""
adapters(
_windows("{846EE342-7039-11DE-9D20-806E6F6E6963}",
"Gigabit Network Connection", "192.168.200.173"),
_windows("{0C6A2C4B-1111-2222-3333-444455556666}",
"Loopback Pseudo-Interface", "127.0.0.1"),
)
assert _gather(["Gigabit Network Connection"]) == ["192.168.200.173"]
def test_include_list_matching_is_case_insensitive(adapters):
adapters(_windows("{GUID}", "Gigabit Network Connection", "192.168.200.173"))
assert _gather(["gigabit network connection"]) == ["192.168.200.173"]
def test_include_list_matches_an_ip_literal(adapters):
"""The address is the one identifier visible on every platform."""
adapters(
_windows("{GUID-A}", "Gigabit Network Connection", "192.168.200.173"),
_windows("{GUID-B}", "Virtual Adapter", "10.0.0.5"),
)
assert _gather(["192.168.200.173"]) == ["192.168.200.173"]
def test_unmatched_include_list_falls_back_to_every_interface(adapters):
"""
The regression: a Linux-style ice_interfaces carried to a Windows guest
matched no GUID and no description, leaving the node with zero host
candidates — silently unconnectable rather than merely slow.
"""
adapters(
_windows("{846EE342-7039-11DE-9D20-806E6F6E6963}",
"Gigabit Network Connection", "192.168.200.173"),
)
assert _gather(["wlp0s20f3"]) == ["192.168.200.173"]
def test_fallback_when_every_adapter_is_virtual(adapters):
"""Auto-exclude must not empty the list either."""
adapters(_linux("virbr0", "192.168.200.254"), _linux("docker0", "172.17.0.1"))
assert _gather() == ["192.168.200.254", "172.17.0.1"]
def test_fallback_warns(adapters, caplog):
adapters(_windows("{GUID}", "Gigabit Network Connection", "192.168.200.173"))
with caplog.at_level("WARNING", logger=ice_filter.log.name):
_gather(["wlp0s20f3"])
assert "matched no address" in caplog.text
def test_loopback_only_host_yields_nothing_and_does_not_warn(adapters, caplog):
"""No address to offer is not a filter misconfiguration."""
adapters(_linux("lo", "127.0.0.1"))
with caplog.at_level("WARNING", logger=ice_filter.log.name):
assert _gather() == []
assert caplog.text == ""
def test_ipv6_global_kept_link_local_dropped(adapters):
adapters(_Adapter("eth0", "eth0", [
("2a01:cb15:80e8:de00::1", 0, 0),
("fe80::af4c:4912:c1ec:b6dc", 0, 3),
("::1", 0, 0),
]))
assert _gather() == ["2a01:cb15:80e8:de00::1"]
def test_ipv4_is_not_index_probed_as_ipv6(adapters):
"""
An IPv4 str must be discriminated by type: the old code fell through to the
IPv6 branch and indexed characters out of it.
"""
ice_filter.install(None)
adapters(_linux("eth0", "192.168.1.22"))
assert aioice.ice.get_host_addresses(use_ipv4=False, use_ipv6=True) == []
|