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
|
#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-2.0
import socket
import struct
import time
from lib.py import bkg, ip, ksft_exit, ksft_run, ksft_eq, ksft_ge, ksft_true, KsftSkipEx
from lib.py import CmdExitFailure, NetNS, NetNSEnter, RtnlAddrFamily
IPV4_ALL_HOSTS_MULTICAST = b'\xe0\x00\x00\x01'
IPV4_TEST_MULTICAST = b'\xef\x01\x01\x01'
IPV6_TEST_MULTICAST = bytes.fromhex('ff020000000000000000000000000123')
def _users_for(rtnl: RtnlAddrFamily, family: int, grp: bytes, ifindex: int):
"""Return mc-users for grp on ifindex, or 0 if absent."""
addrs = rtnl.getmulticast({"ifa-family": family}, dump=True)
matches = [addr for addr in addrs
if addr['multicast'] == grp and addr['ifa-index'] == ifindex]
if not matches:
return 0
if 'mc-users' not in matches[0]:
return None
return matches[0]['mc-users']
def dump_mcaddr_check() -> None:
"""
Verify IPv4 multicast addresses and their user counts in RTM_GETMULTICAST.
"""
with NetNS() as ns:
with NetNSEnter(str(ns)):
ip("link set lo up")
rtnl = RtnlAddrFamily()
lo_idx = socket.if_nametoindex('lo')
addresses = rtnl.getmulticast({"ifa-family": socket.AF_INET}, dump=True)
all_host_multicasts = [
addr for addr in addresses
if addr['multicast'] == IPV4_ALL_HOSTS_MULTICAST
]
ksft_ge(len(all_host_multicasts), 1,
"No interface found with the IPv4 all-hosts multicast address")
mreq = IPV4_TEST_MULTICAST + socket.inet_aton('127.0.0.1')
before = _users_for(rtnl, socket.AF_INET, IPV4_TEST_MULTICAST, lo_idx)
if before is None:
raise KsftSkipEx("kernel does not expose IFA_MC_USERS")
s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s2 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
s1.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
s2.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
after_join = _users_for(rtnl, socket.AF_INET,
IPV4_TEST_MULTICAST, lo_idx)
if after_join is None:
raise KsftSkipEx("kernel does not expose IFA_MC_USERS")
ksft_eq(after_join - before, 2,
f"users delta != 2 after two joins "
f"(before={before}, after={after_join})")
finally:
s1.close()
s2.close()
def dump_mcaddr6_check() -> None:
"""
Verify IPv6 multicast addresses and their user counts in RTM_GETMULTICAST.
"""
with NetNS() as ns:
with NetNSEnter(str(ns)):
ip("link set lo up")
rtnl = RtnlAddrFamily()
lo_idx = socket.if_nametoindex('lo')
before = _users_for(rtnl, socket.AF_INET6,
IPV6_TEST_MULTICAST, lo_idx)
if before is None:
raise KsftSkipEx("kernel does not expose IFA_MC_USERS for IPv6")
mreq = IPV6_TEST_MULTICAST + struct.pack('=I', lo_idx)
s1 = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
s2 = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
try:
s1.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_JOIN_GROUP, mreq)
s2.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_JOIN_GROUP, mreq)
after_join = _users_for(rtnl, socket.AF_INET6,
IPV6_TEST_MULTICAST, lo_idx)
if after_join is None:
raise KsftSkipEx("kernel does not expose IFA_MC_USERS for IPv6")
ksft_eq(after_join - before, 2,
f"IPv6 users delta != 2 after two joins "
f"(before={before}, after={after_join})")
finally:
s1.close()
s2.close()
def ipv4_devconf_notify() -> None:
"""
Configure an interface and set ipv4-devconf values through netlink
to verify that the appropriate netlink notifications are being sent.
"""
with NetNS() as ns:
with NetNSEnter(str(ns)):
ifname = "dummy1"
ip(f"link add name {ifname} type dummy", ns=str(ns))
with bkg("ip monitor", ns=str(ns)) as cmd_obj:
time.sleep(1)
try:
ip(f"link set dev {ifname} inet forwarding on")
ip(f"link set dev {ifname} inet proxy_arp on")
ip(f"link set dev {ifname} inet rp_filter 1")
ip(f"link set dev {ifname} inet ignore_routes_with_linkdown on")
except CmdExitFailure:
raise KsftSkipEx("iproute2 does not support IPv4 devconf attributes")
time.sleep(1)
ksft_true(f"inet {ifname} ignore_routes_with_linkdown on" in cmd_obj.stdout,
f"No 'ignore_routes_with_linkdown on' notificiation found for interface {ifname}")
ksft_true(f"inet {ifname} rp_filter strict" in cmd_obj.stdout,
f"No 'rp_filter strict' notificiation found for interface {ifname}")
ksft_true(f"inet {ifname} proxy_neigh on" in cmd_obj.stdout,
f"No 'proxy_neigh on' notificiation found for interface {ifname}")
ksft_true(f"inet {ifname} forwarding on" in cmd_obj.stdout,
f"No 'forwarding on' notificiation found for interface {ifname}")
def main() -> None:
ksft_run([dump_mcaddr_check, dump_mcaddr6_check, ipv4_devconf_notify])
ksft_exit()
if __name__ == "__main__":
main()
|
