1#!/usr/bin/env python3

2# Copyright 2018-2019 Canonical Ltd.

3# All rights reserved.

4#

5# Written by:

6# Gavin Lin <gavin.lin@canonical.com>

7# Jonathan Cave <jonathan.cave@canonical.com>

8

9"""

10 This script will test Wi-Fi device with netplan automatically.

11

12 To see how to use, please run "./wifi_client_test_netplan.py --help"

13"""

14

15 import argparse

16 import datetime

17 import functools

18 import glob

19 import os

20 import subprocess as sp

21 import textwrap

22 import time

23 import shutil

24 import sys

25 import ipaddress

26 import yaml

27

28 from gateway_ping_test import ping

29 from checkbox_support.snap_utils.system import get_series

30

31 print = functools.partial(print, flush=True)

32

33

34 def print_head(txt):

35 print("##", txt)

36

37

38 def print_cmd(cmd):

39 print("+", cmd)

40

41

42# Configuration file path

43 if "SNAP_USER_DATA" in os.environ: 43 ↛ 44 line 43 didn't jump to line 44 because the condition on line 43 was never true

44 TMP_PATH = os.path.join(

45 os.environ["SNAP_USER_DATA"], "WIFI-TEST-NETPLAN-CREATED-BY-CHECKBOX"

46 )

47 else:

48 TMP_PATH = os.path.join("/tmp", "WIFI-TEST-NETPLAN-CREATED-BY-CHECKBOX")

49

50 NETPLAN_CFG_PATHS = ("/etc/netplan", "/lib/netplan", "/run/netplan")

51 NETPLAN_TEST_CFG = "/etc/netplan/99-CREATED-BY-CHECKBOX.yaml"

52

53

54 def netplan_renderer():

55 """

56 Check the renderer used by netplan on the system.

57 This function looks for the renderer used in the yaml files located in the

58 NETPLAN_CFG_PATHS directories, and returns the first renderer found. If the

59 renderer is not found, it defaults to "networkd".

60 """

61 for basedir in NETPLAN_CFG_PATHS:

62 if os.path.exists(basedir): 62 ↛ 61 line 62 didn't jump to line 61 because the condition on line 62 was always true

63 files = glob.glob(os.path.join(basedir, "*.yaml"))

64 for f in files:

65 with open(f, "r") as file:

66 data = yaml.safe_load(file)

67 if "renderer" in data["network"]: 67 ↛ 64 line 67 didn't jump to line 64

68 return data["network"]["renderer"]

69 return "networkd"

70

71

72 def check_and_get_renderer(renderer):

73 """

74 Check if the renderer provided matches the one used by netplan. If the

75 renderer is set to "AutoDetect", it will return the detected renderer.

76 """

77 machine_renderer = netplan_renderer()

78

79 if renderer == "AutoDetect":

80 return machine_renderer

81 elif renderer != machine_renderer:

82 raise SystemExit(

83 "ERROR: Renderer mismatch, expected: {}, got: {}".format(

84 machine_renderer, renderer

85 )

86 )

87 return renderer

88

89

90 def get_netplan_config_files():

91 config_files = []

92 for basedir in NETPLAN_CFG_PATHS:

93 if os.path.exists(basedir): 93 ↛ 92 line 93 didn't jump to line 92 because the condition on line 93 was always true

94 files = glob.glob(os.path.join(basedir, "*.yaml"))

95 config_files.extend(files)

96 return config_files

97

98

99 def netplan_config_backup():

100 print_head("Backup any existing netplan configuration files")

101 if os.path.exists(TMP_PATH):

102 print("Clear backup location")

103 shutil.rmtree(TMP_PATH)

104

105 config_files = get_netplan_config_files()

106

107 for f in config_files:

108 basedir = os.path.dirname(f)

109 print("Backing up from {}".format(basedir))

110 backup_loc = os.path.join(TMP_PATH, *basedir.split("/"))

111 os.makedirs(backup_loc, exist_ok=True)

112 print(" ", f)

113 shutil.copy(f, backup_loc)

114 print()

115

116

117 def netplan_config_wipe():

118 print_head("Delete any existing netplan configuration files")

119 config_files = get_netplan_config_files()

120 for f in config_files:

121 print(" ", f)

122 os.remove(f)

123

124 # If there's any file left in configuration folder then there's something

125 # not expected, stop the test

126 remaining_files = get_netplan_config_files()

127 if remaining_files:

128 print("ERROR: Failed to wipe netplan config files:")

129 for f in remaining_files:

130 print(" ", f)

131 netplan_config_restore()

132 raise SystemExit("Configuration file restored, exiting...")

133 print()

134

135

136 def netplan_config_restore():

137 print_head("Restore configuration files")

138 files = glob.glob("{}/**/*.yaml".format(TMP_PATH), recursive=True)

139 if files:

140 print("Restoring:")

141 for f in files:

142 restore_loc = f[len(TMP_PATH) :]

143 print(" ", restore_loc)

144 try:

145 shutil.move(f, restore_loc)

146 except shutil.Error:

147 raise SystemExit("Failed to restore {}".format(f))

148

149

150 def generate_test_config(interface, ssid, psk, address, dhcp, wpa3, renderer):

151 """

152 Produce valid netplan yaml from arguments provided

153

154 Typical open ap with dhcp:

155 # This is the network config written by checkbox

156 network:

157 version: 2

158 renderer: networkd

159 wifis:

160 eth0:

161 access-points:

162 my_ap:

163 auth:

164 password: s3cr3t

165 dhcp4: true

166 nameservers: {}

167 """

168 if not ssid:

169 raise SystemExit("A SSID is required for the test")

170 # Define the access-point with the ssid

171 access_point = {ssid: {}}

172 # If psk is provided, add it to the "auth" section

173 if psk:

174 ubuntu_version = int(get_series().split(".")[0])

175 if ubuntu_version == 16:

176 access_point[ssid] = {"password": psk}

177 else:

178 access_point[ssid] = {"auth": {"password": psk}}

179 # Set the key-management to "sae" when WPA3 is used

180 if wpa3:

181 access_point[ssid]["auth"]["key-management"] = "sae"

182 else:

183 access_point[ssid]["auth"]["key-management"] = "psk"

184

185 # Define the interface_info

186 interface_info = {

187 "access-points": access_point,

188 "dhcp4": dhcp,

189 "nameservers": {},

190 }

191

192 # If address is provided, add it to the interface_info

193 if address:

194 interface_info["addresses"] = [address]

195

196 network_config = {

197 "network": {

198 "version": 2,

199 "renderer": renderer,

200 "wifis": {interface: interface_info},

201 }

202 }

203

204 # Serialize the dictionary to a YAML string using pyyaml

205 yaml_output = yaml.safe_dump(network_config, default_flow_style=False)

206 output = textwrap.dedent(

207 "# This is the network config written by checkbox\n" + yaml_output

208 )

209 return output

210

211

212 def write_test_config(config):

213 print_head("Write the test netplan config file to disk")

214 with open(NETPLAN_TEST_CFG, "w", encoding="utf-8") as f:

215 f.write(config)

216 print()

217

218

219 def delete_test_config():

220 print_head("Delete the test file")

221 os.remove(NETPLAN_TEST_CFG)

222 print()

223

224

225 def netplan_apply_config():

226 cmd = "netplan --debug apply"

227 print_cmd(cmd)

228 # Make sure the python env used by netplan is from the base snap

229 env = os.environ

230 env.pop("PYTHONHOME", None)

231 env.pop("PYTHONPATH", None)

232 env.pop("PYTHONUSERBASE", None)

233 retcode = sp.call(cmd, shell=True, env=env)

234 if retcode != 0:

235 print("ERROR: failed netplan apply call")

236 print()

237 return False

238 print()

239 return True

240

241

242 def get_interface_info(interface, renderer):

243 if renderer == "networkd":

244 cmd = "networkctl status --no-pager --no-legend {}".format(interface)

245 key_map = {"State": "state", "Gateway": "gateway"}

246 elif renderer == "NetworkManager":

247 cmd = "nmcli device show {}".format(interface)

248 key_map = {"GENERAL.STATE": "state", "IP4.GATEWAY": "gateway"}

249 else:

250 raise ValueError("Unknown renderer: {}".format(renderer))

251

252 return _get_cmd_info(cmd, key_map, renderer)

253

254

255 def _get_cmd_info(cmd, key_map, renderer):

256 info = {}

257 try:

258 output = sp.check_output(cmd, shell=True)

259 for line in output.decode(sys.stdout.encoding).splitlines():

260 # Skip lines that don't have a "key: value" format

261 if ":" not in line:

262 continue

263 key, val = line.strip().split(":", maxsplit=1)

264 key = key.strip()

265 val = val.strip()

266 if key in key_map:

267 info[key_map[key]] = val

268 except sp.CalledProcessError as e:

269 print("Error running {} command: {}".format(renderer, e))

270 return info

271

272

273 def _check_routable_state(interface, renderer):

274 """

275 Check if the interface is in a routable state depending on the renderer

276 """

277 routable = False

278 state = ""

279 info = get_interface_info(interface, renderer)

280 state = info.get("state", "")

281 if renderer == "networkd":

282 routable = "routable" in state

283 elif renderer == "NetworkManager": 283 ↛ 286 line 283 didn't jump to line 286 because the condition on line 283 was always true

284 routable = "connected" in state and "disconnected" not in state

285 else:

286 raise ValueError("Unknown renderer: {}".format(renderer))

287 return (routable, state)

288

289

290 def wait_for_routable(interface, renderer, max_wait=30):

291 attempts = 0

292 routable = False

293 state = ""

294 while not routable and attempts < max_wait:

295 (routable, state) = _check_routable_state(interface, renderer)

296 time.sleep(1)

297 attempts += 1

298

299 if routable:

300 print("Reached routable state")

301 else:

302 if "degraded" in state:

303 print("ERROR: degraded state, no IP address assigned")

304 else:

305 print("ERROR: did not reach routable state")

306 print()

307 return routable

308

309

310 def print_address_info(interface):

311 cmd = "ip address show dev {}".format(interface)

312 print_cmd(cmd)

313 sp.call(cmd, shell=True)

314 print()

315

316

317 def print_route_info():

318 cmd = "ip route"

319 print_cmd(cmd)

320 sp.call(cmd, shell=True)

321 print()

322

323

324 def _validate_gateway_ip(gateway):

325 if not gateway:

326 return ""

327 # Check if the gateway is a valid IP address

328 # Examples:

329 # 192.168.144.1 (TP-Link 123)

330 # 192.168.144.1

331 ip_address = gateway.split()[0] # Get the first part before any space

332 try:

333 ipaddress.ip_address(ip_address)

334 return ip_address

335 except ValueError:

336 return ""

337

338

339 def get_gateway(interface, renderer):

340 info = get_interface_info(interface, renderer)

341 gateway = info.get("gateway") or ""

342 validated_gateway = _validate_gateway_ip(gateway)

343 print("Got gateway address: {}".format(gateway))

344 print("Validated gateway address: {}".format(validated_gateway))

345 return validated_gateway

346

347

348 def perform_ping_test(interface, renderer):

349 target = get_gateway(interface, renderer)

350

351 if target:

352 count = 5

353 result = ping(target, interface, count, 10)

354 print("Ping result: {}".format(result))

355 if result["received"] == count:

356 return True

357

358 return False

359

360

361 def print_journal_entries(start, renderer):

362 if renderer == "networkd":

363 render_service = "systemd-networkd.service"

364 elif renderer == "NetworkManager":

365 render_service = "NetworkManager.service"

366 else:

367 raise ValueError("Unknown renderer: {}".format(renderer))

368 print_head("Journal Entries")

369 cmd = (

370 "journalctl -q --no-pager "

371 "-u {} "

372 "-u wpa_supplicant.service "

373 "-u netplan-* "

374 '--since "{}" '.format(

375 render_service, start.strftime("%Y-%m-%d %H:%M:%S")

376 )

377 )

378 print_cmd(cmd)

379 sp.call(cmd, shell=True)

380

381

382 def parse_args():

383 # Read arguments

384 parser = argparse.ArgumentParser(

385 description=(

386 "This script will test wireless network with netplan"

387 " in client mode."

388 )

389 )

390 parser.add_argument(

391 "-i",

392 "--interface",

393 type=str,

394 help=("The interface which will be tested, default is wlan0"),

395 default="wlan0",

396 )

397 parser.add_argument(

398 "-s",

399 "--ssid",

400 type=str,

401 help=("SSID of target network, this is required argument"),

402 required=True,

403 )

404 parser.add_argument(

405 "-k",

406 "--psk",

407 type=str,

408 help=(

409 "Pre-shared key of target network, this is optional argument,"

410 " only for PSK protected network"

411 ),

412 )

413 ip_method = parser.add_mutually_exclusive_group(required=True)

414 ip_method.add_argument(

415 "-d",

416 "--dhcp",

417 action="store_true",

418 help=("Enable DHCP for IPv4"),

419 default=False,

420 )

421 ip_method.add_argument(

422 "-a",

423 "--address",

424 type=str,

425 help=(

426 "Set ip address and netmask for the test interface,"

427 " example: 192.168.1.1/24"

428 ),

429 default="",

430 )

431 parser.add_argument(

432 "--wpa3",

433 action="store_true",

434 help=("Configure WPA3 key management for the network"),

435 default=False,

436 )

437

438 parser.add_argument(

439 "--renderer",

440 choices=["networkd", "NetworkManager", "AutoDetect"],

441 help=(

442 "Set the backend daemon to use for netplan configuration. "

443 "If 'AutoDetect' is set, the script will try to determine the "

444 "renderer based on the system configuration. "

445 "(default: %(default)s)"

446 ),

447 default="AutoDetect",

448 )

449

450 return parser.parse_args()

451

452

453 def main():

454 args = parse_args()

455

456 start_time = datetime.datetime.now()

457

458 renderer = check_and_get_renderer(args.renderer)

459 args.renderer = renderer

460 netplan_config_backup()

461 netplan_config_wipe()

462

463 # Create wireless network test configuration file

464 print_head("Generate a test netplan configuration")

465 config_data = generate_test_config(**vars(args))

466 print(config_data)

467 print()

468

469 write_test_config(config_data)

470

471 # Bring up the interface

472 print_head("Apply the test configuration")

473 if not netplan_apply_config():

474 delete_test_config()

475 netplan_config_restore()

476 print_journal_entries(start_time, renderer)

477 raise SystemExit(1)

478 time.sleep(20)

479

480 print_head("Wait for interface to be routable")

481 reached_routable = wait_for_routable(args.interface, renderer)

482

483 test_result = False

484 if reached_routable: 484 ↛ 499 line 484 didn't jump to line 499 because the condition on line 484 was always true

485 print_head("Display address")

486 print_address_info(args.interface)

487

488 print_head("Display route table")

489 print_route_info()

490

491 # Check connection by ping or link status

492 print_head("Perform a ping test")

493 test_result = perform_ping_test(args.interface, renderer)

494 if test_result:

495 print("Connection test passed\n")

496 else:

497 print("Connection test failed\n")

498

499 delete_test_config()

500 netplan_config_restore()

501

502 if not netplan_apply_config(): 502 ↛ 503 line 502 didn't jump to line 503 because the condition on line 502 was never true

503 print_journal_entries(start_time, renderer)

504 raise SystemExit("ERROR: failed to apply restored config")

505

506 if not test_result:

507 print_journal_entries(start_time, renderer)

508 raise SystemExit(1)

509

510 print_journal_entries(start_time, renderer)

511

512

513 if __name__ == "__main__":

514 # Check if executed with root

515 if os.geteuid() != 0:

516 raise SystemExit("Error: please run this command as root")

517 main()