1#!/usr/bin/env python3

2# Copyright 2017-2024 Canonical Ltd.

3# All rights reserved.

4#

5# Written by:

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

7# Taihsiang Ho <taihsiang.ho@canonical.com>

8# Isaac Yang <isaac.yang@canonical.com>

9# Pierre Equoy <pierre.equoy@canonical.com>

10#

11# wireless connection tests using nmcli

12

13

14 import argparse

15 import datetime

16 import functools

17 import os

18 import subprocess as sp

19 import sys

20 import shlex

21

22 from packaging import version as version_parser

23

24 from checkbox_support.helpers.retry import retry

25 from gateway_ping_test import ping

26

27

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

29

30

31 def legacy_nmcli():

32 cmd = "nmcli -v"

33 output = sp.check_output(shlex.split(cmd))

34 version = version_parser.parse(output.strip().split()[-1].decode())

35 # check if using the 16.04 nmcli because of this bug

36 # https://bugs.launchpad.net/plano/+bug/1896806

37 return version < version_parser.parse("1.9.9")

38

39

40 def print_head(txt):

41 print("##", txt)

42

43

44 def print_cmd(cmd):

45 print("+", cmd)

46

47

48 def _get_nm_wireless_connections():

49 cmd = "nmcli -t -f TYPE,UUID,NAME,STATE connection"

50 print_cmd(cmd)

51 output = sp.check_output(shlex.split(cmd))

52 connections = {}

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

54 type, uuid, name, state = line.strip().split(":", 3)

55 if type == "802-11-wireless":

56 connections[name] = {"uuid": uuid, "state": state}

57 return connections

58

59

60 def get_nm_activate_connection():

61 print_head("Get NM activate connection name")

62 connections = _get_nm_wireless_connections()

63 for name, value in connections.items():

64 state = value["state"]

65 uuid = value["uuid"]

66 if state == "activated":

67 print("Activated Connection: {} {}".format(name, uuid))

68 return uuid

69 return ""

70

71

72 def turn_up_connection(uuid):

73 # uuid can also be connection name

74 print_head("Turn up NM connection")

75 cmd = "nmcli c up {}".format(uuid)

76 print("Turn up {}".format(uuid))

77 activate_uuid = get_nm_activate_connection()

78 if uuid == activate_uuid:

79 print("{} state is already activated".format(uuid))

80 return None

81 try:

82 print_cmd(cmd)

83 sp.call(shlex.split(cmd))

84 except Exception as e:

85 print("Can't turn on {}: {}".format(uuid, str(e)))

86

87

88 def turn_down_nm_connections():

89 print_head("Turn off NM all connections")

90 connections = _get_nm_wireless_connections()

91 for name, value in connections.items():

92 uuid = value["uuid"]

93 print("Turn down connection", name)

94 cmd = "nmcli c down {}".format(uuid)

95 print_cmd(cmd)

96 sp.check_call(shlex.split(cmd))

97 print("{} {} is down now".format(name, uuid))

98 print()

99

100

101 def delete_test_ap_ssid_connection():

102 print_head("Cleaning up TEST_CON connection")

103 connections = _get_nm_wireless_connections()

104 if "TEST_CON" not in connections:

105 print("No TEST_CON connection found, nothing to delete")

106 return

107 cmd = "nmcli c delete TEST_CON"

108 print_cmd(cmd)

109 sp.check_call(shlex.split(cmd))

110 print("TEST_CON is deleted")

111

112

113@retry(max_attempts=5, delay=60)

114 def device_rescan():

115 print_head("Calling a rescan")

116 cmd = "nmcli d wifi rescan"

117 print_cmd(cmd)

118 sp.check_call(shlex.split(cmd))

119 print()

120

121

122 def list_aps(ifname, essid=None):

123 if essid:

124 print_head("List APs with ESSID: {}".format(essid))

125 else:

126 print("List all APs")

127 aps_dict = {}

128 fields = "SSID,CHAN,FREQ,SIGNAL"

129 cmd = "nmcli -t -f {} d wifi list ifname {}".format(fields, ifname)

130 output = sp.check_output(shlex.split(cmd))

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

132 # lp bug #1723372 - extra line in output on zesty

133 if line.strip() == ifname: # Skip device name line

134 continue

135 ssid, channel, frequency, signal = line.strip().rsplit(":", 3)

136 if essid and ssid != essid:

137 continue

138 aps_dict[ssid] = {"Chan": channel, "Freq": frequency, "Signal": signal}

139 return aps_dict

140

141

142 def show_aps(aps_dict):

143 for ssid, values in aps_dict.items():

144 print(

145 "SSID: {} Chan: {} Freq: {} Signal: {}".format(

146 ssid, values["Chan"], values["Freq"], values["Signal"]

147 )

148 )

149 print()

150

151

152 def print_address_info(interface):

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

154 print_cmd(cmd)

155 sp.call(shlex.split(cmd))

156 print()

157

158

159 def print_route_info():

160 cmd = "ip route"

161 print_cmd(cmd)

162 sp.call(shlex.split(cmd))

163 print()

164

165

166 def perform_ping_test(interface):

167 target = None

168 cmd = "nmcli --mode tabular --terse --fields IP4.GATEWAY c show TEST_CON"

169 print_cmd(cmd)

170 output = sp.check_output(shlex.split(cmd))

171 target = output.decode(sys.stdout.encoding).strip()

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

173

174 if target: 174 ↛ exit line 174 didn't return from function 'perform_ping_test' because the condition on line 174 was always true

175 count = 5

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

177 if result["received"] != count:

178 raise ValueError(

179 "{} packets expected but only {} received".format(

180 count, result["received"]

181 )

182 )

183

184

185@retry(max_attempts=5, delay=1)

186 def wait_for_connected(interface, essid):

187 cmd = (

188 "nmcli -m tabular -t -f GENERAL.STATE,GENERAL.CONNECTION "

189 "d show {}".format(interface)

190 )

191 print_cmd(cmd)

192 output = sp.check_output(shlex.split(cmd), universal_newlines=True)

193 print(output)

194 state, ssid = output.strip().splitlines()

195

196 if state.startswith("100") and ssid == essid:

197 print("Reached connected state with ESSID: {}".format(essid))

198 elif ssid != essid:

199 error_msg = (

200 "ERROR: did not reach connected state with ESSID: {}\n"

201 "ESSID mismatch:\n Excepted:{}\n Actually:{}"

202 ).format(essid, ssid, essid)

203 raise SystemExit(error_msg)

204 elif not state.startswith("100"): 204 ↛ 207 line 204 didn't jump to line 207 because the condition on line 204 was always true

205 error_msg = "State is not connected: {}".format(state)

206 raise SystemExit(error_msg)

207 print()

208

209

210 def connection(cmd, device):

211 print_head("Connection attempt")

212 print_cmd(cmd)

213 sp.check_call(shlex.split(cmd))

214

215 # Make sure the connection is brought up

216 turn_up_connection("TEST_CON")

217

218 print_head("Ensure interface is connected")

219 wait_for_connected(device, "TEST_CON")

220

221 print_head("Display address")

222 print_address_info(device)

223

224 print_head("Display route table")

225 print_route_info()

226

227 print_head("Perform a ping test")

228 perform_ping_test(device)

229 print("Connection test passed\n")

230

231

232 def open_connection(args):

233 # Configure the connection

234 # ipv4.dhcp-timeout 30 : ensure plenty of time to get address

235 # ipv6.method ignore : I believe that NM can report the device as Connected

236 # if an IPv6 address is setup. This should ensure in

237 # this test we are using IPv4

238 cmd = (

239 "nmcli c add con-name TEST_CON "

240 "ifname {} "

241 "type wifi "

242 "ssid '{}' "

243 "-- "

244 "ipv4.method auto "

245 "ipv4.dhcp-timeout 30 "

246 "ipv6.method ignore".format(args.device, args.essid)

247 )

248 connection(cmd, args.device)

249

250

251 def secured_connection(args):

252 # Configure the connection

253 # ipv4.dhcp-timeout 30 : ensure plenty of time to get address

254 # ipv6.method ignore : I believe that NM can report the device as Connected

255 # if an IPv6 address is setup. This should ensure in

256 # this test we are using IPv4

257 cmd = (

258 "nmcli c add con-name TEST_CON "

259 "ifname {} "

260 "type wifi "

261 "ssid '{}' "

262 "-- "

263 "wifi-sec.key-mgmt {} "

264 "wifi-sec.psk {} "

265 "ipv4.method auto "

266 "ipv4.dhcp-timeout 30 "

267 "ipv6.method ignore".format(

268 args.device, args.essid, args.exchange, args.psk

269 )

270 )

271 connection(cmd, args.device)

272

273

274 def hotspot(args):

275 print_head("Create Wi-Fi hotspot")

276 cmd = (

277 "nmcli c add type wifi ifname {} con-name TEST_CON autoconnect no"

278 " ssid CHECKBOX_AP".format(args.device)

279 )

280 print_cmd(cmd)

281 retcode = sp.call(shlex.split(cmd))

282 if retcode != 0:

283 print("Connection creation failed\n")

284 return retcode

285 cmd = (

286 "nmcli c modify TEST_CON 802-11-wireless.mode ap ipv4.method shared"

287 " 802-11-wireless.band {}".format(args.band)

288 )

289 print_cmd(cmd)

290 retcode = sp.call(shlex.split(cmd))

291 if retcode != 0:

292 print("Set band failed\n")

293 return retcode

294 cmd = (

295 "nmcli c modify TEST_CON wifi-sec.key-mgmt wpa-psk "

296 'wifi-sec.psk "ubuntu1234"'

297 )

298 print_cmd(cmd)

299 retcode = sp.call(shlex.split(cmd))

300 if retcode != 0:

301 print("Setting up wifi security failed\n")

302 return retcode

303 turn_up_connection("TEST_CON")

304 if retcode != 0:

305 print("Failed to bring up connection\n")

306 print()

307 return retcode

308

309

310 def print_journal_entries(start):

311 print_head("Journal Entries")

312 cmd = (

313 "journalctl -q --no-pager "

314 "-u snap.network-manager.networkmanager.service "

315 "-u NetworkManager.service "

316 "-u wpa_supplicant.service "

317 '--since "{}" '.format(start.strftime("%Y-%m-%d %H:%M:%S"))

318 )

319 print_cmd(cmd)

320 sp.call(shlex.split(cmd))

321

322

323 def parser_args():

324 parser = argparse.ArgumentParser(

325 description="WiFi connection test using mmcli"

326 )

327

328 subparsers = parser.add_subparsers(dest="test_type")

329 subparsers.required = True

330

331 parser_scan = subparsers.add_parser(

332 "scan", help="Test can scan for networks only"

333 )

334 parser_scan.add_argument("device", type=str, help="Device name e.g. wlan0")

335

336 parser_open = subparsers.add_parser(

337 "open", help="Test connection to an open access point"

338 )

339 parser_open.add_argument("device", type=str, help="Device name e.g. wlan0")

340 parser_open.add_argument("essid", type=str, help="ESSID")

341 parser_open.set_defaults(func=open_connection)

342

343 parser_secured = subparsers.add_parser(

344 "secured", help="Test connection to a secured access point"

345 )

346 parser_secured.add_argument(

347 "device", type=str, help="Device name e.g. wlan0"

348 )

349 parser_secured.add_argument("essid", type=str, help="ESSID")

350 parser_secured.add_argument("psk", type=str, help="Pre-Shared Key")

351 parser_secured.add_argument(

352 "--exchange",

353 type=str,

354 default="wpa-psk",

355 help="exchange type (default: %(default)s)",

356 )

357 parser_secured.set_defaults(func=secured_connection)

358

359 parser_ap = subparsers.add_parser("ap", help="Test creation of a hotspot")

360 parser_ap.add_argument("device", type=str, help="Device name e.g. wlan0")

361 parser_ap.add_argument("band", type=str, help="Band")

362 parser_ap.set_defaults(func=hotspot)

363

364 args = parser.parse_args()

365

366 return args

367

368

369@retry(max_attempts=5, delay=60)

370 def main():

371 args = parser_args()

372 start_time = datetime.datetime.now()

373 device_rescan()

374 essid = getattr(args, "essid", None)

375 aps_dict = list_aps(args.device, essid)

376 show_aps(aps_dict)

377

378 if args.test_type == "scan":

379 if not aps_dict:

380 raise SystemExit("Failed to find any access point.")

381 else:

382 print("Found {} access points".format(len(aps_dict)))

383 return

384

385 if not aps_dict:

386 raise SystemExit(

387 "Targed access point: {} not found".format(args.essid)

388 )

389

390 if args.func: 390 ↛ exit line 390 didn't return from function 'main' because the condition on line 390 was always true

391 delete_test_ap_ssid_connection()

392 activated_uuid = get_nm_activate_connection()

393 turn_down_nm_connections()

394 try:

395 args.func(args)

396 except Exception:

397 # The test is not required to run as root, but root access is

398 # required for journal access so only attempt to print when e.g.

399 # running under Remote

400 if os.geteuid() == 0:

401 print_journal_entries(start_time)

402 raise

403 finally:

404 turn_up_connection(activated_uuid)

405 delete_test_ap_ssid_connection()

406

407

408 if __name__ == "__main__":

409 sys.exit(main())