1#!/usr/bin/env python3

2#

3# This program is free software: you can redistribute it and/or modify

4# it under the terms of the GNU General Public License version 3,

5# as published by the Free Software Foundation.

6#

7# This program is distributed in the hope that it will be useful,

8# but WITHOUT ANY WARRANTY; without even the implied warranty of

9# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

10# GNU General Public License for more details.

11#

12# You should have received a copy of the GNU General Public License

13# along with this program; if not, write to the Free Software Foundation,

14# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

15#

16# Parts of this are based on the example python code that ships with

17# NetworkManager

18# http://cgit.freedesktop.org/NetworkManager/NetworkManager/tree/examples/python

19#

20# Copyright (C) 2012-2019 Canonical, Ltd.

21

22 import argparse

23 import fcntl

24 from natsort import natsorted

25 import os

26 import socket

27 import struct

28 from subprocess import check_output, CalledProcessError, STDOUT

29 import sys

30

31 from checkbox_support.parsers.modinfo import ModinfoParser

32 from checkbox_support.parsers.udevadm import UdevadmParser

33

34

35 class Utils:

36

37 sys_path = "/sys/class/net"

38

39 @staticmethod

40 def get_ethtool_info(interface, key):

41 """Return ethtool information identified by key"""

42 cmd = ["ethtool", interface]

43 try:

44 output = check_output(cmd, stderr=STDOUT, universal_newlines=True)

45 except CalledProcessError:

46 return "ethtool returned error"

47 except FileNotFoundError:

48 return "ethtool not installed"

49 if not output:

50 return "ethtool returned no output"

51 data = []

52 output_line = False

53 for line in iter(output.splitlines()):

54 if key in line:

55 label_start = line.find(key)

56 data_start = label_start + len(key)

57 output_line = True

58 if (

59 output_line is True

60 and line[label_start] != " "

61 and key not in line

62 ):

63 output_line = False

64 if output_line:

65 data.append(line[data_start:].strip())

66 if len(data) == 0:

67 data.append("Not reported")

68 result = "\n" + "".join(

69 ["\t" + item + "\n" for item in natsorted(data)]

70 )

71 result = result.rstrip()

72 return result

73

74 @classmethod

75 def is_iface_connected(cls, iface):

76 try:

77 carrier_file = os.path.join(cls.sys_path, iface, "carrier")

78 return int(open(carrier_file, "r").read()) == 1

79 except Exception:

80 pass

81 return False

82

83 @staticmethod

84 def get_ipv4_address(interface):

85 s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

86 try:

87

88 buffer = struct.pack("256s", interface[:15].encode())

89 # retrieve the IP address associated with a network interface.

90 data = fcntl.ioctl(s.fileno(), 0x8915, buffer)

91 ipv4_addr = socket.inet_ntoa(data[20:24])

92 except OSError as e:

93 print(

94 "OSError: failed to get the IPv4 address for %s: %s"

95 % (interface, repr(e)),

96 file=sys.stderr,

97 )

98 ipv4_addr = "***NOT CONFIGURED***"

99 finally:

100 return ipv4_addr

101

102 @staticmethod

103 def get_ipv6_address(interface):

104 cmd = [

105 "/sbin/ip",

106 "-6",

107 "-o",

108 "addr",

109 "show",

110 "dev",

111 interface,

112 "scope",

113 "link",

114 ]

115 proc = check_output(cmd, universal_newlines=True)

116 try:

117 # If it failed at split(), it means there is no output from cmd.

118 ipv6_addr = proc.split()[3].strip()

119 except IndexError as e:

120 print(

121 "IndexError: failed to get the IPv6 address for %s: %s"

122 % (interface, repr(e)),

123 file=sys.stderr,

124 )

125 ipv6_addr = "***NOT CONFIGURED***"

126 return ipv6_addr

127

128 @classmethod

129 def get_mac_address(cls, interface):

130 address_file = os.path.join(cls.sys_path, interface, "address")

131 try:

132 return open(address_file, "r").read().strip()

133 except IOError:

134 return "UNKNOWN"

135

136 @classmethod

137 def get_speed(cls, interface):

138 speed_file = os.path.join(cls.sys_path, interface, "speed")

139 try:

140 return open(speed_file, "r").read().strip()

141 except IOError:

142 return "UNKNOWN"

143

144 @staticmethod

145 def get_driver_version(driver):

146 cmd = ["/sbin/modinfo", driver]

147 try:

148 output = check_output(cmd, stderr=STDOUT, universal_newlines=True)

149 except CalledProcessError:

150 return None

151 if not output:

152 return None

153 parser = ModinfoParser(output)

154 modinfo = parser.get_all()

155

156 # try the version field first, then vermagic second, some audio

157 # drivers don't report version if the driver is in-tree

158 version = modinfo.get("version")

159

160 if version is None or version == "in-tree":

161 # vermagic will look like this (below) and we only care about the

162 # first part:

163 # "3.2.0-29-generic SMP mod_unload modversions"

164 version = modinfo.get("vermagic").split()[0]

165

166 return version

167

168

169 class NetworkDeviceInfo:

170

171 def __init__(self):

172 self._category = None

173 self._interface = None

174 self._product = None

175 self._vendor = None

176 self._driver = None

177 self._driver_version = None

178 self._firmware_missing = None

179 self._path = None

180 self._id = None

181 self._subsystem_id = None

182 self._mac = None

183 self._carrier_status = None

184 self._ipv4 = None

185 self._ipv6 = None

186 self._speed = None

187 self._supported_modes = None

188 self._advertised_modes = None

189 self._partner_modes = None

190

191 def __str__(self):

192 ret = ""

193 for key, val in vars(self).items():

194 if val is not None:

195 # leading _ removed, remaining ones spaces

196 pretty_key = key.lstrip("_").replace("_", " ").title()

197 ret += "{}: {}\n".format(pretty_key, val)

198 return ret

199

200 @property

201 def interface(self):

202 return self._interface

203

204 @interface.setter

205 def interface(self, value):

206 self._interface = value

207 self._interface_populate()

208

209 @property

210 def carrier_status(self):

211 return self._carrier_status

212

213 @property

214 def driver(self):

215 return self._driver

216

217 @driver.setter

218 def driver(self, value):

219 self._driver = value

220 self._driver_populate()

221

222 @property

223 def category(self):

224 return self._category

225

226 @category.setter

227 def category(self, value):

228 self._category = value

229

230 @property

231 def product(self):

232 return self._product

233

234 @product.setter

235 def product(self, value):

236 self._product = value

237

238 @property

239 def vendor(self):

240 return self._vendor

241

242 @vendor.setter

243 def vendor(self, value):

244 self._vendor = value

245

246 @property

247 def firmware_missing(self):

248 return self._firmware_missing

249

250 @firmware_missing.setter

251 def firmware_missing(self, value):

252 self._firmware_missing = value

253

254 @property

255 def path(self):

256 return self._path

257

258 @path.setter

259 def path(self, value):

260 self._path = value

261

262 @property

263 def id(self):

264 return self._id

265

266 @id.setter

267 def id(self, value):

268 self._id = value

269

270 @property

271 def subsystem_id(self):

272 return self._subsystem_id

273

274 @subsystem_id.setter

275 def subsystem_id(self, value):

276 self._subsystem_id = value

277

278 def _interface_populate(self):

279 """Get extra attributes based on interface"""

280 if self.interface is None:

281 return

282 self._mac = Utils.get_mac_address(self.interface)

283 self._supported_modes = Utils.get_ethtool_info(

284 self.interface, "Supported link modes:"

285 )

286 self._advertised_modes = Utils.get_ethtool_info(

287 self.interface, "Advertised link modes:"

288 )

289 if Utils.is_iface_connected(self.interface):

290 self._carrier_status = "Connected"

291 self._ipv4 = Utils.get_ipv4_address(self.interface)

292 self._ipv6 = Utils.get_ipv6_address(self.interface)

293 self._speed = Utils.get_speed(self.interface)

294 self._partner_modes = Utils.get_ethtool_info(

295 self.interface, "Link partner advertised link modes:"

296 )

297 else:

298 self._carrier_status = "Disconnected"

299

300 def _driver_populate(self):

301 """Get extra attributes based on driver"""

302 if self.driver is None:

303 return

304 self._driver_version = Utils.get_driver_version(self.driver)

305

306

307 class NMDevices:

308 # This example lists basic information about network interfaces known to NM

309 devtypes = {

310 1: "Ethernet",

311 2: "WiFi",

312 5: "Bluetooth",

313 6: "OLPC",

314 7: "WiMAX",

315 8: "Modem",

316 }

317

318 def __init__(self, category="NETWORK"):

319 self.category = category

320 self._devices = []

321 self._collect_devices()

322

323 def __len__(self):

324 return len(self._devices)

325

326 def _collect_devices(self):

327 import dbus

328

329 bus = dbus.SystemBus()

330 proxy = bus.get_object(

331 "org.freedesktop.NetworkManager", "/org/freedesktop/NetworkManager"

332 )

333 manager = dbus.Interface(proxy, "org.freedesktop.NetworkManager")

334 self._devices = manager.GetDevices()

335

336 def devices(self):

337 """Convert to list of NetworkDevice with NM derived attrs set"""

338 import dbus

339

340 for d in self._devices:

341 bus = dbus.SystemBus()

342 dev_proxy = bus.get_object("org.freedesktop.NetworkManager", d)

343 prop_iface = dbus.Interface(

344 dev_proxy, "org.freedesktop.DBus.Properties"

345 )

346 props = prop_iface.GetAll("org.freedesktop.NetworkManager.Device")

347 devtype = self.devtypes.get(props.get("DeviceType"))

348 if devtype is None:

349 continue

350 nd = NetworkDeviceInfo()

351 if self.category == "NETWORK":

352 if devtype == "Ethernet":

353 nd.category = self.category

354 else:

355 continue

356 if self.category == "WIRELESS":

357 if devtype == "WiFi":

358 nd.category = self.category

359 else:

360 continue

361 nd.interface = props.get("Interface")

362 nd.driver = props.get("Driver")

363 nd.firmware_missing = props.get("FirmwareMissing")

364 yield nd

365

366

367 class UdevDevices:

368

369 def __init__(self, category="NETWORK"):

370 self.category = category

371 self._devices = []

372 self._collect_devices()

373

374 def __len__(self):

375 return len(self._devices)

376

377 def _collect_devices(self):

378 cmd = ["udevadm", "info", "--export-db"]

379 try:

380 output = check_output(cmd).decode(

381 sys.stdout.encoding, errors="ignore"

382 )

383 except CalledProcessError as err:

384 sys.stderr.write(err)

385 return

386 udev = UdevadmParser(output)

387 for device in udev.run():

388 if (

389 device.category == self.category

390 and device.interface != "UNKNOWN"

391 ):

392 self._devices.append(device)

393

394 def devices(self):

395 """Convert to list of NetworkDevice with UDev derived attrs set"""

396 for device in self._devices:

397 nd = NetworkDeviceInfo()

398 nd.category = getattr(device, "category", None)

399 nd.interface = getattr(device, "interface", None)

400 nd.product = getattr(device, "product", None)

401 nd.vendor = getattr(device, "vendor", None)

402 nd.driver = getattr(device, "driver", None)

403 nd.path = getattr(device, "path", None)

404 vid = getattr(device, "vendor_id", None)

405 pid = getattr(device, "product_id", None)

406 if vid and pid:

407 nd.id = "[{0:04x}:{1:04x}]".format(vid, pid)

408 svid = getattr(device, "subvendor_id", None)

409 spid = getattr(device, "subproduct_id", None)

410 if svid and spid:

411 nd.subsystem_id = "[{0:04x}:{1:04x}]".format(svid, spid)

412 yield nd

413

414

415 if __name__ == "__main__":

416 parser = argparse.ArgumentParser(

417 description="Gather information about network devices"

418 )

419 parser.add_argument(

420 "action",

421 choices=("detect", "info"),

422 help="Detect mode or just report info",

423 )

424 parser.add_argument(

425 "category",

426 choices=("NETWORK", "WIRELESS"),

427 help="Either ethernet or WLAN devices",

428 )

429 parser.add_argument(

430 "--no-nm",

431 action="store_true",

432 help="Don't attempt to get info from network manager",

433 )

434 parser.add_argument(

435 "--interface", help="Restrict info action to specified interface"

436 )

437 parser.add_argument(

438 "--fail-on-disconnected",

439 action="store_true",

440 help=(

441 "Script will exit with a non-zero return code if"

442 " any interface is not connected"

443 ),

444 )

445 args = parser.parse_args()

446

447 udev = UdevDevices(args.category)

448 disconnected_ifaces = []

449

450 # The detect action should indicate presence of a device belonging to the

451 # category and cause the job to fail if none present

452 if args.action == "detect":

453 if len(udev) == 0:

454 raise SystemExit("No devices detected by udev")

455 else:

456 print("[ Devices found by udev ]".center(80, "-"))

457 for device in udev.devices():

458 print(device)

459

460 # The info action should just gather infomation about any ethernet devices

461 # found and report for inclusion in e.g. an attachment job and include

462 # NetworkManager as a source if desired

463 if args.action == "info":

464 # If interface has been specified

465 if args.interface:

466 for device in udev.devices():

467 if device.interface == args.interface:

468 print(device)

469 sys.exit(0)

470

471 # Report udev detected devices first

472 print("[ Devices found by udev ]".center(80, "-"))

473 for device in udev.devices():

474 print(device)

475 if device.carrier_status == "Disconnected":

476 disconnected_ifaces.append(device.interface)

477

478 # Attempt to report devices found by NetworkManager. This can be

479 # skipped as doesn't make sense for server installs

480 if not args.no_nm:

481 nm = NMDevices(args.category)

482 print("[ Devices found by Network Manager ]".center(80, "-"))

483 for device in nm.devices():

484 print(device)

485

486 if disconnected_ifaces and args.fail_on_disconnected:

487 print("WARNING: The following interfaces are not connected:")

488 for iface in disconnected_ifaces:

489 print(iface)

490 sys.exit(1)

491

492 sys.exit(0)