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
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
|
# Copyright (c) 2022, Dell Inc. or its subsidiaries. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
# See the LICENSE file for details.
#
# This file is part of NVMe STorage Appliance Services (nvme-stas).
#
# Authors: Martin Belanger <Martin.Belanger@dell.com>
'''A collection of IP address and network interface utilities'''
import struct
import socket
import ipaddress
RTM_BASE = 16
RTM_GETLINK = 18
RTM_NEWADDR = 20
RTM_GETADDR = 22
NLM_F_REQUEST = 0x01
NLM_F_ROOT = 0x100
NLMSG_DONE = 3
NLMSG_HDRLEN = 16
IFADDRMSG_SZ = 8
IFINFOMSG_SZ = 16
ARPHRD_ETHER = 1
ARPHRD_LOOPBACK = 772
NLMSG_LENGTH = lambda msg_len: msg_len + NLMSG_HDRLEN # pylint: disable=unnecessary-lambda-assignment
RTATTR_SZ = 4
RTA_ALIGN = lambda length: ((length + 3) & ~3) # pylint: disable=unnecessary-lambda-assignment
IFLA_ADDRESS = 1
IFLA_IFNAME = 3
def _nlmsghdr(nlmsg_type, nlmsg_flags, nlmsg_seq, nlmsg_pid, msg_len: int):
'''Implement this C struct:
struct nlmsghdr {
__u32 nlmsg_len; /* Length of message including header */
__u16 nlmsg_type; /* Message content */
__u16 nlmsg_flags; /* Additional flags */
__u32 nlmsg_seq; /* Sequence number */
__u32 nlmsg_pid; /* Sending process port ID */
};
'''
return struct.pack('<LHHLL', NLMSG_LENGTH(msg_len), nlmsg_type, nlmsg_flags, nlmsg_seq, nlmsg_pid)
def _ifaddrmsg(family=0, prefixlen=0, flags=0, scope=0, index=0):
'''Implement this C struct:
struct ifaddrmsg {
__u8 ifa_family;
__u8 ifa_prefixlen; /* The prefix length */
__u8 ifa_flags; /* Flags */
__u8 ifa_scope; /* Address scope */
__u32 ifa_index; /* Link index */
};
'''
return struct.pack('<BBBBL', family, prefixlen, flags, scope, index)
def _ifinfomsg(family=0, dev_type=0, index=0, flags=0, change=0):
'''Implement this C struct:
struct ifinfomsg {
unsigned char ifi_family; /* AF_UNSPEC */
unsigned char __ifi_pad;
unsigned short ifi_type; /* Device type: ARPHRD_* */
int ifi_index; /* Interface index */
unsigned int ifi_flags; /* Device flags: IFF_* */
unsigned int ifi_change; /* change mask: IFF_* */
};
'''
return struct.pack('<BBHiII', family, 0, dev_type, index, flags, change)
def _nlmsg(nlmsg_type, nlmsg_flags, msg: bytes):
'''Build a Netlink message'''
return _nlmsghdr(nlmsg_type, nlmsg_flags, 0, 0, len(msg)) + msg
# Netlink request (Get address command)
GETADDRCMD = _nlmsg(RTM_GETADDR, NLM_F_REQUEST | NLM_F_ROOT, _ifaddrmsg())
# Netlink request (Get address command)
GETLINKCMD = _nlmsg(RTM_GETLINK, NLM_F_REQUEST | NLM_F_ROOT, _ifinfomsg(family=socket.AF_UNSPEC, change=0xFFFFFFFF))
# ******************************************************************************
def _data_matches_mac(data, mac):
return mac.lower() == ':'.join([f'{x:02x}' for x in data[0:6]])
def mac2iface(mac: str): # pylint: disable=too-many-locals
'''@brief Find the interface that has @mac as its assigned MAC address.
@param mac: The MAC address to match
'''
with socket.socket(family=socket.AF_NETLINK, type=socket.SOCK_RAW, proto=socket.NETLINK_ROUTE) as sock:
sock.sendall(GETLINKCMD)
nlmsg = sock.recv(8192)
nlmsg_idx = 0
while True: # pylint: disable=too-many-nested-blocks
if nlmsg_idx >= len(nlmsg):
nlmsg += sock.recv(8192)
nlmsghdr = nlmsg[nlmsg_idx : nlmsg_idx + NLMSG_HDRLEN]
nlmsg_len, nlmsg_type, _, _, _ = struct.unpack('<LHHLL', nlmsghdr)
if nlmsg_type == NLMSG_DONE:
break
if nlmsg_type == RTM_BASE:
msg_indx = nlmsg_idx + NLMSG_HDRLEN
msg = nlmsg[msg_indx : msg_indx + IFINFOMSG_SZ] # ifinfomsg
_, _, ifi_type, ifi_index, _, _ = struct.unpack('<BBHiII', msg)
if ifi_type in (ARPHRD_LOOPBACK, ARPHRD_ETHER):
rtattr_indx = msg_indx + IFINFOMSG_SZ
while rtattr_indx < (nlmsg_idx + nlmsg_len):
rtattr = nlmsg[rtattr_indx : rtattr_indx + RTATTR_SZ]
rta_len, rta_type = struct.unpack('<HH', rtattr)
if rta_type == IFLA_ADDRESS:
data = nlmsg[rtattr_indx + RTATTR_SZ : rtattr_indx + rta_len]
if _data_matches_mac(data, mac):
return socket.if_indextoname(ifi_index)
rta_len = RTA_ALIGN(rta_len) # Round up to multiple of 4
rtattr_indx += rta_len # Move to next rtattr
nlmsg_idx += nlmsg_len # Move to next Netlink message
return ''
# ******************************************************************************
def ip_equal(ip1, ip2):
'''Check whther two IP addresses are equal.
@param ip1: IPv4Address or IPv6Address object
@param ip2: IPv4Address or IPv6Address object
'''
if not isinstance(ip1, ipaddress._BaseAddress): # pylint: disable=protected-access
return False
if not isinstance(ip2, ipaddress._BaseAddress): # pylint: disable=protected-access
return False
if ip1.version == 4 and ip2.version == 6:
ip2 = ip2.ipv4_mapped
elif ip1.version == 6 and ip2.version == 4:
ip1 = ip1.ipv4_mapped
return ip1 == ip2
# ******************************************************************************
def get_ipaddress_obj(ipaddr, ipv4_mapped_convert=False):
'''@brief Return a IPv4Address or IPv6Address depending on whether @ipaddr
is a valid IPv4 or IPv6 address. Return None otherwise.
If ipv4_mapped_resolve is set to True, IPv6 addresses that are IPv4-Mapped,
will be converted to their IPv4 equivalent.
'''
try:
ip = ipaddress.ip_address(ipaddr)
except ValueError:
return None
if ipv4_mapped_convert:
ipv4_mapped = getattr(ip, 'ipv4_mapped', None)
if ipv4_mapped is not None:
ip = ipv4_mapped
return ip
# ******************************************************************************
def net_if_addrs(): # pylint: disable=too-many-locals
'''@brief Return a dictionary listing every IP addresses for each interface.
The first IP address of a list is the primary address used as the default
source address.
@example: {
'wlp0s20f3': {
4: [IPv4Address('10.0.0.28')],
6: [
IPv6Address('fd5e:9a9e:c5bd:0:5509:890c:1848:3843'),
IPv6Address('fd5e:9a9e:c5bd:0:1fd5:e527:8df7:7912'),
IPv6Address('2605:59c8:6128:fb00:c083:1b8:c467:81d2'),
IPv6Address('2605:59c8:6128:fb00:e99d:1a02:38e0:ad52'),
IPv6Address('fe80::d71b:e807:d5ee:7614'),
],
},
'lo': {
4: [IPv4Address('127.0.0.1')],
6: [IPv6Address('::1')],
},
'docker0': {
4: [IPv4Address('172.17.0.1')],
6: []
},
}
'''
interfaces = {}
with socket.socket(socket.AF_NETLINK, socket.SOCK_RAW) as sock:
sock.sendall(GETADDRCMD)
nlmsg = sock.recv(8192)
nlmsg_idx = 0
while True: # pylint: disable=too-many-nested-blocks
if nlmsg_idx >= len(nlmsg):
nlmsg += sock.recv(8192)
nlmsghdr = nlmsg[nlmsg_idx : nlmsg_idx + NLMSG_HDRLEN]
nlmsg_len, nlmsg_type, _, _, _ = struct.unpack('<LHHLL', nlmsghdr)
if nlmsg_type == NLMSG_DONE:
break
if nlmsg_type == RTM_NEWADDR:
msg_indx = nlmsg_idx + NLMSG_HDRLEN
msg = nlmsg[msg_indx : msg_indx + IFADDRMSG_SZ] # ifaddrmsg
ifa_family, _, _, _, ifa_index = struct.unpack('<BBBBL', msg)
if ifa_family in (socket.AF_INET, socket.AF_INET6):
interfaces.setdefault(ifa_index, {4: [], 6: []})
rtattr_indx = msg_indx + IFADDRMSG_SZ
while rtattr_indx < (nlmsg_idx + nlmsg_len):
rtattr = nlmsg[rtattr_indx : rtattr_indx + RTATTR_SZ]
rta_len, rta_type = struct.unpack('<HH', rtattr)
if rta_type == IFLA_IFNAME:
data = nlmsg[rtattr_indx + RTATTR_SZ : rtattr_indx + rta_len]
ifname = data.rstrip(b'\0').decode()
interfaces[ifa_index]['name'] = ifname
elif rta_type == IFLA_ADDRESS:
data = nlmsg[rtattr_indx + RTATTR_SZ : rtattr_indx + rta_len]
ip = get_ipaddress_obj(data)
if ip:
family = 4 if ifa_family == socket.AF_INET else 6
interfaces[ifa_index][family].append(ip)
rta_len = RTA_ALIGN(rta_len) # Round up to multiple of 4
rtattr_indx += rta_len # Move to next rtattr
nlmsg_idx += nlmsg_len # Move to next Netlink message
if_addrs = {}
for value in interfaces.values():
name = value.pop('name', None)
if name is not None:
if_addrs[name] = value
return if_addrs
# ******************************************************************************
def get_interface(ifaces: dict, src_addr):
'''Get interface for given source address
@param ifaces: Interface info previously returned by @net_if_addrs()
@param src_addr: IPv4Address or IPv6Address object
'''
if not isinstance(src_addr, ipaddress._BaseAddress): # pylint: disable=protected-access
return ''
for iface, addr_map in ifaces.items():
for addrs in addr_map.values():
for addr in addrs:
if ip_equal(src_addr, addr):
return iface
return ''
|
