1#!/usr/bin/env python3

2# This file is part of Checkbox.

3#

4# Copyright 2007-2024 Canonical Ltd.

5# Written by:

6# Brendan Donegan <brendan.donegan@canonical.com>

7# Daniel Manrique <daniel.manrique@canonical.com>

8# David Murphy <david.murphy@canonical.com>

9# Javier Collado <javier.collado@canonical.com>

10# Jeff Lane <jeffrey.lane@canonical.com>

11# Marc Tardif <marc.tardif@canonical.com>

12# Mathieu Trudel-Lapierre <mathieu.trudel-lapierre@canonical.com>

13# Zygmunt Krynicki <zygmunt.krynicki@canonical.com>

14# Massimiliano Girardi <massimiliano.giarardi@canonical.com>

15#

16# Checkbox is free software: you can redistribute it and/or modify

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

18# as published by the Free Software Foundation.

19#

20# Checkbox is distributed in the hope that it will be useful,

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

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

23# GNU General Public License for more details.

24#

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

26# along with Checkbox. If not, see <http://www.gnu.org/licenses/>.

27

28 from gettext import gettext as _

29 import argparse

30 import gettext

31 import logging

32 import os

33 import re

34 import socket

35 import struct

36 import subprocess

37 import sys

38 import time

39

40 from contextlib import suppress

41 from typing import Dict, List

42

43

44 class Route:

45 """

46 Gets routing information from the system.

47 """

48

49 def __init__(self, interface):

50 self.interface = interface

51

52 def _num_to_dotted_quad(self, number):

53 """

54 Convert long int to dotted quad string

55 """

56 return socket.inet_ntoa(struct.pack("<L", number))

57

58 def _get_default_gateway_from_ip(self):

59 try:

60 # Note: this uses -o instead of -j for xenial/bionic compatibility

61 routes = subprocess.check_output(

62 [

63 "ip",

64 "-o",

65 "route",

66 "show",

67 "default",

68 "0.0.0.0/0",

69 "dev",

70 self.interface,

71 ],

72 universal_newlines=True,

73 )

74 except subprocess.CalledProcessError:

75 return None

76 for route in routes.splitlines():

77 if route.startswith("default via"):

78 return route.split()[2]

79 return None

80

81 def _get_default_gateway_from_proc(self):

82 """

83 Returns the current default gateway, reading that from /proc

84 """

85 logging.debug(_("Reading default gateway information from /proc"))

86 try:

87 with open("/proc/net/route", "rt") as stream:

88 route = stream.read()

89 except Exception:

90 logging.error(_("Failed to read def gateway from /proc"))

91 return None

92

93 proc_table_re = re.compile(

94 r"\n(?P<interface>\w+)\s+00000000\s+" r"(?P<def_gateway>[\w]+)\s+"

95 )

96 proc_table_lines = proc_table_re.finditer(route)

97 for proc_table_line in proc_table_lines:

98 def_gateway = proc_table_line.group("def_gateway")

99 interface = proc_table_line.group("interface")

100 if interface == self.interface and def_gateway:

101 return self._num_to_dotted_quad(int(def_gateway, 16))

102 logging.error(_("Could not find def gateway info in /proc"))

103 return None

104

105 def _get_default_gateway_from_bin_route(self):

106 """

107 Return the gateway for the interface associated with this Route object.

108 """

109 default_gws = get_default_gateways()

110 return default_gws.get(self.interface)

111

112 def _get_ip_addr_info(self):

113 return subprocess.check_output(

114 ["ip", "-o", "addr", "show"], universal_newlines=True

115 )

116

117 def get_broadcast(self):

118 # Get list of all IPs from all my interfaces,

119 ip_addr_infos = self._get_ip_addr_info()

120 for addr_info_line in ip_addr_infos.splitlines():

121 # id: if_name inet addr/mask brd broadcast

122 addr_info_fields = addr_info_line.split()

123 with suppress(IndexError):

124 if (

125 addr_info_fields[1] == self.interface

126 and addr_info_fields[4] == "brd"

127 ):

128 return addr_info_fields[5]

129 raise ValueError(

130 "Unable to determine broadcast for iface {}".format(self.interface)

131 )

132

133 def _get_default_gateway_from_networkctl(self):

134 try:

135 network_info = subprocess.check_output(

136 [

137 "networkctl",

138 "status",

139 "--no-pager",

140 "--no-legend",

141 self.interface,

142 ],

143 universal_newlines=True,

144 )

145 except subprocess.CalledProcessError:

146 return None

147 for line in network_info.splitlines():

148 line = line.strip()

149 if line.startswith("Gateway:"):

150 return line.split()[-1]

151 return None

152

153 def get_default_gateways(self) -> set:

154 """

155 Use multiple sources to get the default gateway to be robust to

156 possible platform bugs

157 """

158 def_gateways = {

159 self._get_default_gateway_from_ip(),

160 self._get_default_gateway_from_proc(),

161 self._get_default_gateway_from_bin_route(),

162 self._get_default_gateway_from_networkctl(),

163 }

164 def_gateways -= {None}

165 if len(def_gateways) > 1:

166 logging.warning(

167 "Found more than one default gateway for interface {}".format(

168 self.interface

169 )

170 )

171 return def_gateways

172

173 @staticmethod

174 def get_interface_from_ip(ip):

175 # Note: this uses -o instead of -j for xenial/bionic compatibility

176 route_info = subprocess.check_output(

177 ["ip", "-o", "route", "get", ip], universal_newlines=True

178 )

179 for line in route_info.splitlines():

180 # ip dev device_name src ...

181 fields = line.split()

182 if len(fields) > 3: 182 ↛ 179 line 182 didn't jump to line 179 because the condition on line 182 was always true

183 return fields[2]

184 raise ValueError(

185 "Unable to determine any device used for {}".format(ip)

186 )

187

188 @staticmethod

189 def get_any_interface():

190 # Note: this uses -o instead of -j for xenial/bionic compatibility

191 route_infos = subprocess.check_output(

192 ["ip", "-o", "route", "show", "default", "0.0.0.0/0"],

193 universal_newlines=True,

194 )

195 for route_info in route_infos.splitlines():

196 route_info_fields = route_info.split()

197 if len(route_info_fields) > 5:

198 return route_info_fields[4]

199 raise ValueError("Unable to determine any valid interface")

200

201 @staticmethod

202 def from_ip(ip: str):

203 """

204 Build an instance of Route given an ip, if no ip is provided the best

205 interface that can route to 0.0.0.0/0 is selected (as described by

206 metric)

207 """

208 if ip:

209 interface = Route.get_interface_from_ip(ip)

210 return Route(interface)

211 return Route(Route.get_any_interface())

212

213

214 def is_reachable(ip, interface):

215 """

216 Ping an ip to see if it is reachable

217 """

218 result = ping(ip, interface)

219 return result["transmitted"] >= result["received"] > 0

220

221

222 def get_default_gateway_reachable_on(interface: str) -> str:

223 """

224 Returns the default gateway of an interface if it is reachable

225 """

226 if not interface:

227 raise ValueError("Unable to ping on interface None")

228 route = Route(interface=interface)

229 desired_targets = route.get_default_gateways()

230 for desired_target in desired_targets:

231 if is_reachable(desired_target, interface):

232 return desired_target

233 raise ValueError(

234 "Unable to reach any estimated gateway of interface {}".format(

235 interface

236 ),

237 )

238

239

240 def get_any_host_reachable_on(interface: str) -> str:

241 """

242 Returns any host that it can reach from a given interface

243 """

244 if not interface:

245 raise ValueError("Unable to ping on interface None")

246 route = Route(interface=interface)

247 broadcast = route.get_broadcast()

248 arp_parser_re = re.compile(

249 r"\? \((?P<ip>[\d.]+)\) at (?P<mac>[a-f0-9\:]+) "

250 r"\[ether\] on (?P<iface>[\w\d]+)"

251 )

252 # retry a few times to get something in the arp table

253 for i in range(10):

254 ping(broadcast, interface, broadcast=True)

255 # Get output from arp -a -n to get known IPs

256 arp_table = subprocess.check_output(

257 ["arp", "-a", "-n"], universal_newlines=True

258 )

259 hosts_in_arp_table = [

260 arp_entry.group("ip") # ip

261 for arp_entry in arp_parser_re.finditer(arp_table)

262 if arp_entry.group("iface") == interface

263 ]

264 # we don't know how an ip got in the arp table, lets try to reach them

265 # and return the first that we can acutally reach

266 for host in hosts_in_arp_table:

267 if is_reachable(host, interface):

268 return host

269 # we were unable to get any reachable host in the arp table, this may

270 # be due to a slow network, lets retry in a few seconds

271 time.sleep(5)

272 raise ValueError(

273 "Unable to reach any host on interface {}".format(interface)

274 )

275

276

277 def get_host_to_ping(interface: str, target: str = None) -> "str|None":

278 """

279 Attempts to determine a reachable host to ping on the specified network

280 interface. First it tries to ping the provided target. If no target is

281 specified or the target is not reachable, it then attempts to find a

282 reachable host by trying the default gateway and finally falling back on

283 any host on the network interface.

284

285 @returns: The reachable host if any, else None

286 """

287 # Try to use the provided target if it is reachable

288 if target and is_reachable(target, interface):

289 return target

290 # From here onward, lets try to estimate a reachable target

291 if not interface:

292 route = Route.from_ip(target)

293 interface = route.interface

294

295 # Try first with any default gateway that we can gather on the interface

296 with suppress(ValueError):

297 return get_default_gateway_reachable_on(interface)

298

299 # Try with any host we can find reachable on the interface

300 with suppress(ValueError):

301 return get_any_host_reachable_on(interface)

302

303 # Unable to estimate any host to reach

304 return None

305

306

307 def ping(

308 host: str,

309 interface: "str|None",

310 count: int = 2,

311 deadline: int = 4,

312 broadcast=False,

313):

314 """

315 pings an host via an interface count times within the given deadline.

316 If the interface is None, it will not be used.

317 If the host is a broadcast host, use the broadcast kwarg

318

319 @returns: on success the stats of the ping "transmitted", "received" and

320 "pct_loss"

321 @returns: on failure a dict with a "cause" key, with the failure reason

322 """

323 command = ["ping", str(host), "-c", str(count), "-w", str(deadline)]

324 if interface:

325 command.append("-I{}".format(interface))

326 if broadcast:

327 command.append("-b")

328 reg = re.compile(

329 r"(\d+) packets transmitted, (\d+) received,"

330 r".*([0-9]*\.?[0-9]*.)% packet loss"

331 )

332 ping_summary = {"transmitted": 0, "received": 0, "pct_loss": 0}

333 try:

334 output = subprocess.check_output(

335 command, universal_newlines=True, stderr=subprocess.PIPE

336 )

337 except (OSError, FileNotFoundError) as e:

338 ping_summary["cause"] = str(e)

339 return ping_summary

340 except subprocess.CalledProcessError as e:

341 # Ping returned fail exit code

342 # broadcast will always do so

343 if broadcast:

344 return

345 ping_summary["cause"] = (

346 "Failed with exception: {}\nstdout: {}\nstderr: {}".format(

347 str(e), e.stdout, e.stderr

348 )

349 )

350 return ping_summary

351

352 print(output)

353 try:

354 received = next(re.finditer(reg, output))

355 ping_summary = {

356 "transmitted": int(received.group(1)),

357 "received": int(received.group(2)),

358 "pct_loss": int(received.group(3)),

359 }

360 except StopIteration:

361 ping_summary["cause"] = (

362 "Failed to parse the stats from the ping output. Log: {}".format(

363 output

364 )

365 )

366 return ping_summary

367

368

369 def perform_ping_test(interfaces: List[str], target=None) -> None:

370 """

371 Perform a ping test on the specified interfaces.

372 If any of the provided interfaces successfully pinged the target host,

373 the function returns 0. Otherwise, it returns 1.

374 """

375 for iface in interfaces:

376 host = get_host_to_ping(iface, target)

377 if not host:

378 print(

379 "Failed to find a host to ping on interface {}".format(iface)

380 )

381 continue

382 print(

383 "Pinging {} with {} interface".format(

384 host, iface or "*unspecified*"

385 )

386 )

387 ping_summary = ping(host, iface)

388 if ping_summary["received"] != ping_summary["transmitted"]:

389 print("FAIL: {0}% packet loss.".format(ping_summary["pct_loss"]))

390 continue

391 if ping_summary["transmitted"] > 0:

392 print(_("PASS: 0% packet loss").format(host))

393 return 0

394 else:

395 print("FAIL: Unable to ping any host with the above interfaces")

396 return 1

397

398

399 def parse_args(argv):

400 parser = argparse.ArgumentParser()

401 parser.add_argument(

402 "host",

403 nargs="?",

404 default=None,

405 help=_("host to ping"),

406 )

407 iface_mutex_group = parser.add_mutually_exclusive_group()

408 iface_mutex_group.add_argument(

409 "-I",

410 "--interface",

411 help=_("use specified interface to send packets"),

412 action="append",

413 dest="interfaces",

414 )

415 iface_mutex_group.add_argument(

416 "--any-cable-interface",

417 help=_("use any cable interface to send packets"),

418 action="store_true",

419 )

420 return parser.parse_args(argv)

421

422

423 def main(argv) -> int:

424 gettext.textdomain("com.canonical.certification.checkbox")

425 gettext.bindtextdomain(

426 "com.canonical.certification.checkbox",

427 os.getenv("CHECKBOX_PROVIDER_LOCALE_DIR", None),

428 )

429

430 args = parse_args(argv)

431

432 if args.any_cable_interface:

433 print(_("Looking for all cable interfaces..."))

434 all_ifaces = get_default_gateways().keys()

435 args.interfaces = list(filter(is_cable_interface, all_ifaces))

436 if not args.interfaces:

437 raise SystemExit(

438 "FAIL: Couldn't find any suitable cable interface."

439 )

440

441 # If no interfaces were specified, use None to let the function

442 # determine the interface to use (this is to make it compliant with

443 # the original implementation)

444 if not args.interfaces:

445 args.interfaces = [None]

446

447 return perform_ping_test(args.interfaces, args.host)

448

449

450 def get_default_gateways() -> Dict[str, str]:

451 """

452 Use `route` program to find default gateways for all interfaces.

453

454 returns a dictionary in a form of {interface_name: gateway}

455 """

456 try:

457 routes = subprocess.check_output(

458 ["route", "-n"], universal_newlines=True

459 )

460 except subprocess.CalledProcessError as exc:

461 logging.debug("Failed to run `route -n `", exc)

462 return {}

463 regex = r"^0\.0\.0\.0\s+(?P<gw>[\w.]+)\s.*\s(?P<interface>[\w.]+)$"

464 matches = re.finditer(regex, routes, re.MULTILINE)

465

466 return {m.group("interface"): m.group("gw") for m in matches}

467

468

469 def is_cable_interface(interface: str) -> bool:

470 """

471 Check if the interface is a cable interface.

472 This is a simple heuristic that checks if the interface is named

473 "enX" or "ethX" where X is a number.

474

475 :param interface: the interface name to check

476 :return: True if the interface is a cable interface, False otherwise

477

478 Looking at the `man 7 systemd.net-naming-scheme` we can see that

479 even the `eth` matching may be an overkill.

480 """

481 if not isinstance(interface, str) or not interface:

482 return False

483 return interface.startswith("en") or interface.startswith("eth")

484

485

486 if __name__ == "__main__":

487 sys.exit(main(sys.argv[1:]))