Coverage for icloudpy/base.py: 78%

1"""Library base file.""" 

2 

3import base64 

4import getpass 

5import hashlib 

6import http.cookiejar as cookielib 

7import inspect 

8import json 

9import logging 

10from os import mkdir, path 

11from re import match 

12from tempfile import gettempdir 

13from uuid import uuid1 

14 

15import srp 

16from requests import Session 

17 

18from icloudpy.exceptions import ( 

19 ICloudPy2SARequiredException, 

20 ICloudPyAPIResponseException, 

21 ICloudPyFailedLoginException, 

22 ICloudPyServiceNotActivatedException, 

23) 

24from icloudpy.services import ( 

25 AccountService, 

26 CalendarService, 

27 ContactsService, 

28 DriveService, 

29 FindMyiPhoneServiceManager, 

30 PhotosService, 

31 RemindersService, 

32) 

33from icloudpy.utils import get_password_from_keyring 

34 

35LOGGER = logging.getLogger(__name__) 

36 

37HEADER_DATA = { 

38 "X-Apple-ID-Account-Country": "account_country", 

39 "X-Apple-ID-Session-Id": "session_id", 

40 "X-Apple-Session-Token": "session_token", 

41 "X-Apple-TwoSV-Trust-Token": "trust_token", 

42 "scnt": "scnt", 

43} 

44 

45 

46class ICloudPyPasswordFilter(logging.Filter): 

47 """Password log hider.""" 

48 

49 def __init__(self, password): 

50 super().__init__(password) 

51 

52 def filter(self, record): 

53 message = record.getMessage() 

54 if self.name in message: 

55 record.msg = message.replace(self.name, "*" * 8) 

56 record.args = [] 

57 

58 return True 

59 

60 

61class ICloudPySession(Session): 

62 """iCloud session.""" 

63 

64 def __init__(self, service): 

65 self.service = service 

66 Session.__init__(self) 

67 

68 def request(self, method, url, **kwargs): # pylint: disable=arguments-differ 

69 # Charge logging to the right service endpoint 

70 callee = inspect.stack()[2] 

71 module = inspect.getmodule(callee[0]) 

72 request_logger = logging.getLogger(module.__name__).getChild("http") 

73 if self.service.password_filter not in request_logger.filters: 

74 request_logger.addFilter(self.service.password_filter) 

75 

76 request_logger.debug(f"{method} {url} {kwargs.get('data', '')}") 

77 

78 has_retried = kwargs.get("retried") 

79 kwargs.pop("retried", None) 

80 response = super().request(method, url, **kwargs) 

81 

82 content_type = response.headers.get("Content-Type", "").split(";")[0] 

83 json_mimetypes = ["application/json", "text/json"] 

84 

85 for header, value in HEADER_DATA.items(): 

86 if response.headers.get(header): 

87 session_arg = value 

88 self.service.session_data.update( 

89 {session_arg: response.headers.get(header)}, 

90 ) 

91 

92 # Save session_data to file 

93 with open(self.service.session_path, "w", encoding="utf-8") as outfile: 

94 json.dump(self.service.session_data, outfile) 

95 LOGGER.debug("Saved session data to file") 

96 

97 # Save cookies to file 

98 self.cookies.save(ignore_discard=True, ignore_expires=True) 

99 LOGGER.debug("Cookies saved to %s", self.service.cookiejar_path) 

100 

101 if not response.ok and (content_type not in json_mimetypes or response.status_code in [421, 450, 500]): 

102 try: 

103 # pylint: disable=W0212 

104 fmip_url = self.service._get_webservice_url("findme") 

105 if has_retried is None and response.status_code == 450 and fmip_url in url: 

106 # Handle re-authentication for Find My iPhone 

107 LOGGER.debug("Re-authenticating Find My iPhone service") 

108 try: 

109 self.service.authenticate(True, "find") 

110 except ICloudPyAPIResponseException: 

111 LOGGER.debug("Re-authentication failed") 

112 kwargs["retried"] = True 

113 return self.request(method, url, **kwargs) 

114 except Exception: 

115 pass 

116 

117 if has_retried is None and response.status_code in [421, 450, 500]: 

118 api_error = ICloudPyAPIResponseException( 

119 response.reason, 

120 response.status_code, 

121 retry=True, 

122 ) 

123 request_logger.debug(api_error) 

124 kwargs["retried"] = True 

125 return self.request(method, url, **kwargs) 

126 

127 self._raise_error(response.status_code, response.reason) 

128 

129 if content_type not in json_mimetypes: 

130 return response 

131 

132 try: 

133 data = response.json() 

134 except: # noqa: E722 

135 request_logger.warning("Failed to parse response with JSON mimetype") 

136 return response 

137 

138 request_logger.debug(data) 

139 

140 if isinstance(data, dict): 

141 reason = data.get("errorMessage") 

142 reason = reason or data.get("reason") 

143 reason = reason or data.get("errorReason") 

144 if not reason and isinstance(data.get("error"), str): 

145 reason = data.get("error") 

146 if not reason and data.get("error"): 

147 reason = "Unknown reason" 

148 

149 code = data.get("errorCode") 

150 if not code and data.get("serverErrorCode"): 

151 code = data.get("serverErrorCode") 

152 

153 if reason: 

154 self._raise_error(code, reason) 

155 

156 return response 

157 

158 def _raise_error(self, code, reason): 

159 if self.service.requires_2sa and reason == "Missing X-APPLE-WEBAUTH-TOKEN cookie": 

160 raise ICloudPy2SARequiredException(self.service.user["apple_id"]) 

161 if code in ("ZONE_NOT_FOUND", "AUTHENTICATION_FAILED"): 

162 reason = reason + ". Please log into https://icloud.com/ to manually finish setting up your iCloud service" 

163 api_error = ICloudPyServiceNotActivatedException(reason, code) 

164 LOGGER.error(api_error) 

165 

166 raise api_error 

167 if code == "ACCESS_DENIED": 

168 reason = ( 

169 reason + ". Please wait a few minutes then try again." 

170 "The remote servers might be trying to throttle requests." 

171 ) 

172 if code in [421, 450, 500]: 

173 reason = "Authentication required for Account." 

174 

175 api_error = ICloudPyAPIResponseException(reason, code) 

176 LOGGER.error(api_error) 

177 raise api_error 

178 

179 # Public method to resolve linting error 

180 def raise_error(self, code, reason): 

181 return self._raise_error(code=code, reason=reason) 

182 

183 

184# Apple serves its sign-in widget from here, so this is the origin a browser 

185# reports when signing a challenge for it. 

186SECURITY_KEY_ORIGIN = "https://idmsa.apple.com" 

187 

188# Apple answers an accepted security key assertion with 409, alongside the 

189# ordinary success codes. 250 is its own "two-factor completed". 

190SECURITY_KEY_ACCEPTED_STATUSES = frozenset({200, 204, 250, 409}) 

191 

192 

193def _b64_decode(value): 

194 """Decode Apple's base64, which is unpadded and may use either alphabet.""" 

195 return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)) 

196 

197 

198def _b64_encode(value): 

199 """Encode bytes the way Apple's own client does.""" 

200 return base64.b64encode(value).decode("ascii") 

201 

202 

203def build_security_key_assertion(response, rp_id, request_id=None): 

204 """Build the payload Apple expects from a WebAuthn assertion response. 

205 

206 ``response`` is a ``fido2`` ``AuthenticatorAssertionResponse`` or any 

207 object exposing the same ``response`` and ``raw_id`` attributes. 

208 

209 The challenge is taken from ``clientData`` rather than passed in, which 

210 keeps the value signed and the value declared in step. 

211 """ 

212 client_data = bytes(response.response.client_data) 

213 challenge = json.loads(client_data).get("challenge", "") 

214 user_handle = response.response.user_handle 

215 

216 return { 

217 "challenge": challenge.replace("-", "+").replace("_", "/"), 

218 "clientData": _b64_encode(client_data), 

219 "signatureData": _b64_encode(bytes(response.response.signature)), 

220 "authenticatorData": _b64_encode(bytes(response.response.authenticator_data)), 

221 "userHandle": _b64_encode(bytes(user_handle)).rstrip("=") if user_handle else "", 

222 "credentialID": _b64_encode(bytes(response.raw_id)), 

223 "rpId": rp_id, 

224 "requestId": request_id or "", 

225 } 

226 

227 

228class ICloudPyService: 

229 """ 

230 A base authentication class for the iCloud service. Handles the 

231 authentication required to access iCloud services. 

232 

233 Usage: 

234 from src import ICloudPyService 

235 icloudpy = ICloudPyService('username@apple.com', 'password') 

236 icloudpy.iphone.location() 

237 """ 

238 

239 def __init__( 

240 self, 

241 apple_id, 

242 password=None, 

243 cookie_directory=None, 

244 verify=True, 

245 client_id=None, 

246 with_family=True, 

247 auth_endpoint="https://idmsa.apple.com/appleauth/auth", 

248 # For China, use "https://www.icloud.com.cn" 

249 home_endpoint="https://www.icloud.com", 

250 # For China, use "https://setup.icloud.com.cn/setup/ws/1" 

251 setup_endpoint="https://setup.icloud.com/setup/ws/1", 

252 ): 

253 if password is None: 

254 password = get_password_from_keyring(apple_id) 

255 

256 self.user = {"accountName": apple_id, "password": password} 

257 self.data = {} 

258 self.params = {} 

259 self.client_id = client_id or (f"auth-{str(uuid1()).lower()}") 

260 self.with_family = with_family 

261 self.auth_endpoint = auth_endpoint 

262 self.home_endpoint = home_endpoint 

263 self.setup_endpoint = setup_endpoint 

264 

265 self.password_filter = ICloudPyPasswordFilter(password) 

266 LOGGER.addFilter(self.password_filter) 

267 

268 if cookie_directory: 

269 self._cookie_directory = path.expanduser(path.normpath(cookie_directory)) 

270 if not path.exists(self._cookie_directory): 

271 mkdir(self._cookie_directory, 0o700) 

272 else: 

273 topdir = path.join(gettempdir(), "icloudpy") 

274 self._cookie_directory = path.join(topdir, getpass.getuser()) 

275 if not path.exists(topdir): 

276 mkdir(topdir, 0o777) 

277 if not path.exists(self._cookie_directory): 

278 mkdir(self._cookie_directory, 0o700) 

279 

280 LOGGER.debug("Using session file %s", self.session_path) 

281 

282 self.session_data = {} 

283 try: 

284 with open(self.session_path, encoding="utf-8") as session_f: 

285 self.session_data = json.load(session_f) 

286 except: # noqa: E722 

287 LOGGER.info("Session file does not exist") 

288 if self.session_data.get("client_id"): 

289 self.client_id = self.session_data.get("client_id") 

290 self.params["clientId"] = self.client_id 

291 else: 

292 self.session_data.update({"client_id": self.client_id}) 

293 self.params["clientId"] = self.client_id 

294 

295 self.session = ICloudPySession(self) 

296 self.session.verify = verify 

297 self.session.headers.update( 

298 {"Origin": self.home_endpoint, "Referer": f"{self.home_endpoint}/"}, 

299 ) 

300 

301 cookiejar_path = self.cookiejar_path 

302 self.session.cookies = cookielib.LWPCookieJar(filename=cookiejar_path) 

303 if path.exists(cookiejar_path): 

304 try: 

305 self.session.cookies.load(ignore_discard=True, ignore_expires=True) 

306 LOGGER.debug("Read cookies from %s", cookiejar_path) 

307 except: # noqa: E722 

308 # Most likely a pickled cookiejar from earlier versions. 

309 # The cookiejar will get replaced with a valid one after 

310 # successful authentication. 

311 LOGGER.warning("Failed to read cookiejar %s", cookiejar_path) 

312 

313 self.authenticate() 

314 

315 self._drive = None 

316 self._photos = None 

317 

318 def authenticate(self, force_refresh=False, service=None): 

319 """ 

320 Handles authentication, and persists cookies so that 

321 subsequent logins will not cause additional e-mails from Apple. 

322 """ 

323 

324 login_successful = False 

325 if self.session_data.get("session_token") and not force_refresh: 

326 LOGGER.debug("Checking session token validity") 

327 try: 

328 self.data = self._validate_token() 

329 login_successful = True 

330 except ICloudPyAPIResponseException: 

331 LOGGER.debug("Invalid authentication token, will log in from scratch.") 

332 

333 if not login_successful and service is not None: 

334 app = self.data["apps"][service] 

335 if "canLaunchWithOneFactor" in app and app["canLaunchWithOneFactor"] is True: 

336 LOGGER.debug( 

337 "Authenticating as %s for %s", 

338 self.user["accountName"], 

339 service, 

340 ) 

341 

342 try: 

343 self._authenticate_with_credentials_service(service) 

344 login_successful = True 

345 except Exception as error: 

346 LOGGER.debug( 

347 "Could not log into service. Attempting brand new login. %s", 

348 str(error), 

349 ) 

350 

351 if not login_successful: 

352 LOGGER.debug("Authenticating as %s", self.user["accountName"]) 

353 

354 headers = self._get_auth_headers() 

355 

356 if self.session_data.get("scnt"): 

357 headers["scnt"] = self.session_data.get("scnt") 

358 

359 if self.session_data.get("session_id"): 

360 headers["X-Apple-ID-Session-Id"] = self.session_data.get("session_id") 

361 

362 class SrpPassword: 

363 def __init__(self, password: str): 

364 self.password = password 

365 

366 def set_encrypt_info( 

367 self, 

368 salt: bytes, 

369 iterations: int, 

370 key_length: int, 

371 ): 

372 self.salt = salt 

373 self.iterations = iterations 

374 self.key_length = key_length 

375 

376 def encode(self): 

377 password_hash = hashlib.sha256( 

378 self.password.encode("utf-8"), 

379 ).digest() 

380 return hashlib.pbkdf2_hmac( 

381 "sha256", 

382 password_hash, 

383 salt, 

384 iterations, 

385 key_length, 

386 ) 

387 

388 srp_password = SrpPassword(self.user["password"]) 

389 srp.rfc5054_enable() 

390 srp.no_username_in_x() 

391 usr = srp.User( 

392 self.user["accountName"], 

393 srp_password, 

394 hash_alg=srp.SHA256, 

395 ng_type=srp.NG_2048, 

396 ) 

397 

398 uname, a_bytes = usr.start_authentication() 

399 

400 data = { 

401 "a": base64.b64encode(a_bytes).decode(), 

402 "accountName": uname, 

403 "protocols": ["s2k", "s2k_fo"], 

404 } 

405 

406 try: 

407 response = self.session.post( 

408 f"{self.auth_endpoint}/signin/init", 

409 data=json.dumps(data), 

410 headers=headers, 

411 ) 

412 response.raise_for_status() 

413 except ICloudPyAPIResponseException as error: 

414 msg = "Failed to initiate srp authentication." 

415 raise ICloudPyFailedLoginException(msg, error) from error 

416 

417 body = response.json() 

418 

419 salt = base64.b64decode(body["salt"]) 

420 b = base64.b64decode(body["b"]) 

421 c = body["c"] 

422 iterations = body["iteration"] 

423 key_length = 32 

424 srp_password.set_encrypt_info(salt, iterations, key_length) 

425 

426 m1 = usr.process_challenge(salt, b) 

427 m2 = usr.H_AMK 

428 

429 data = { 

430 "accountName": uname, 

431 "c": c, 

432 "m1": base64.b64encode(m1).decode(), 

433 "m2": base64.b64encode(m2).decode(), 

434 "rememberMe": True, 

435 "trustTokens": [], 

436 } 

437 

438 if self.session_data.get("trust_token"): 

439 data["trustTokens"] = [self.session_data.get("trust_token")] 

440 

441 try: 

442 self.session.post( 

443 f"{self.auth_endpoint}/signin/complete", 

444 params={"isRememberMeEnabled": "true"}, 

445 data=json.dumps(data), 

446 headers=headers, 

447 ) 

448 except ICloudPyAPIResponseException as error: 

449 msg = "Invalid email/password combination." 

450 raise ICloudPyFailedLoginException(msg, error) from error 

451 

452 self._authenticate_with_token() 

453 

454 self._webservices = self.data["webservices"] 

455 

456 LOGGER.debug("Authentication completed successfully") 

457 

458 def _authenticate_with_token(self): 

459 """Authenticate using session token.""" 

460 data = { 

461 "accountCountryCode": self.session_data.get("account_country"), 

462 "dsWebAuthToken": self.session_data.get("session_token"), 

463 "extended_login": True, 

464 "trustToken": self.session_data.get("trust_token", ""), 

465 } 

466 

467 try: 

468 req = self.session.post( 

469 f"{self.setup_endpoint}/accountLogin", 

470 data=json.dumps(data), 

471 ) 

472 self.data = req.json() 

473 except ICloudPyAPIResponseException as error: 

474 msg = "Invalid authentication token." 

475 raise ICloudPyFailedLoginException(msg, error) from error 

476 

477 def _authenticate_with_credentials_service(self, service): 

478 """Authenticate to a specific service using credentials.""" 

479 data = { 

480 "appName": service, 

481 "apple_id": self.user["accountName"], 

482 "password": self.user["password"], 

483 } 

484 

485 try: 

486 self.session.post( 

487 f"{self.setup_endpoint}/accountLogin", 

488 data=json.dumps(data), 

489 ) 

490 

491 self.data = self._validate_token() 

492 except ICloudPyAPIResponseException as error: 

493 msg = "Invalid email/password combination." 

494 raise ICloudPyFailedLoginException(msg, error) from error 

495 

496 def _validate_token(self): 

497 """Checks if the current access token is still valid.""" 

498 LOGGER.debug("Checking session token validity") 

499 try: 

500 req = self.session.post(f"{self.setup_endpoint}/validate", data="null") 

501 LOGGER.debug("Session token is still valid") 

502 return req.json() 

503 except ICloudPyAPIResponseException as err: 

504 LOGGER.debug("Invalid authentication token") 

505 raise err 

506 

507 def _get_auth_headers(self, overrides=None): 

508 headers = { 

509 "Accept": "application/json, text/javascript", 

510 "Content-Type": "application/json", 

511 "X-Apple-OAuth-Client-Id": "d39ba9916b7251055b22c7f910e2ea796ee65e98b2ddecea8f5dde8d9d1a815d", 

512 "X-Apple-OAuth-Client-Type": "firstPartyAuth", 

513 "X-Apple-OAuth-Redirect-URI": "https://www.icloud.com", 

514 "X-Apple-OAuth-Require-Grant-Code": "true", 

515 "X-Apple-OAuth-Response-Mode": "web_message", 

516 "X-Apple-OAuth-Response-Type": "code", 

517 "X-Apple-OAuth-State": self.client_id, 

518 "X-Apple-Widget-Key": "d39ba9916b7251055b22c7f910e2ea796ee65e98b2ddecea8f5dde8d9d1a815d", 

519 } 

520 if overrides: 

521 headers.update(overrides) 

522 return headers 

523 

524 @property 

525 def cookiejar_path(self): 

526 """Get path for cookiejar file.""" 

527 return path.join( 

528 self._cookie_directory, 

529 "".join([c for c in self.user.get("accountName") if match(r"\w", c)]), 

530 ) 

531 

532 @property 

533 def session_path(self): 

534 """Get path for session data file.""" 

535 return path.join( 

536 self._cookie_directory, 

537 "".join([c for c in self.user.get("accountName") if match(r"\w", c)]) + ".session", 

538 ) 

539 

540 @property 

541 def requires_2sa(self): 

542 """Returns True if two-step authentication is required.""" 

543 return self.data.get("dsInfo", {}).get("hsaVersion", 0) >= 1 and ( 

544 self.data.get("hsaChallengeRequired", False) or not self.is_trusted_session 

545 ) 

546 

547 @property 

548 def requires_2fa(self): 

549 """Returns True if two-factor authentication is required.""" 

550 return self.data["dsInfo"].get("hsaVersion", 0) == 2 and ( 

551 self.data.get("hsaChallengeRequired", False) or not self.is_trusted_session 

552 ) 

553 

554 @property 

555 def is_trusted_session(self): 

556 """Returns True if the session is trusted.""" 

557 return self.data.get("hsaTrustedBrowser", False) 

558 

559 @property 

560 def trusted_devices(self): 

561 """Returns devices trusted for two-step authentication.""" 

562 request = self.session.get( 

563 f"{self.setup_endpoint}/listDevices", 

564 params=self.params, 

565 ) 

566 return request.json().get("devices") 

567 

568 def send_verification_code(self, device): 

569 """Requests that a verification code is sent to the given device.""" 

570 data = json.dumps(device) 

571 request = self.session.post( 

572 f"{self.setup_endpoint}/sendVerificationCode", 

573 params=self.params, 

574 data=data, 

575 ) 

576 return request.json().get("success", False) 

577 

578 def validate_verification_code(self, device, code): 

579 """Verifies a verification code received on a trusted device.""" 

580 device.update({"verificationCode": code, "trustBrowser": True}) 

581 data = json.dumps(device) 

582 

583 try: 

584 self.session.post( 

585 f"{self.setup_endpoint}/validateVerificationCode", 

586 params=self.params, 

587 data=data, 

588 ) 

589 except ICloudPyAPIResponseException as error: 

590 if error.code == -21669: 

591 # Wrong verification code 

592 return False 

593 raise 

594 

595 self.trust_session() 

596 

597 return not self.requires_2sa 

598 

599 def trigger_2fa_push_notification(self): 

600 """Asks Apple to push a 2FA verification code to trusted devices. 

601 

602 Apple's auth flow (changed around iOS 26 / early 2026) requires a PUT 

603 to ``/verify/trusteddevice/securitycode`` (no body) to initiate code 

604 delivery to trusted devices. Without this step, the user is prompted 

605 for a code but no code is ever pushed to any device — auth stalls. 

606 

607 Callers should invoke this method before prompting the user for the 

608 2FA code (i.e., right after ``requires_2fa`` returns True). Failure 

609 is non-fatal — a code may still arrive via SMS or another path. 

610 

611 See: https://github.com/mandarons/icloud-docker/issues/426 

612 Ported from icloud_photos_downloader PR #1335. 

613 

614 :returns: True if the push request succeeded, False otherwise. 

615 """ 

616 headers = self._get_auth_headers({"Accept": "application/json"}) 

617 

618 if self.session_data.get("scnt"): 

619 headers["scnt"] = self.session_data.get("scnt") 

620 

621 if self.session_data.get("session_id"): 

622 headers["X-Apple-ID-Session-Id"] = self.session_data.get("session_id") 

623 

624 try: 

625 response = self.session.put( 

626 f"{self.auth_endpoint}/verify/trusteddevice/securitycode", 

627 headers=headers, 

628 ) 

629 if not (200 <= response.status_code < 300): 

630 LOGGER.debug( 

631 "2FA push notification trigger returned status %s", 

632 response.status_code, 

633 ) 

634 return False 

635 LOGGER.debug("2FA push notification triggered.") 

636 return True 

637 except Exception as error: # network, SSL, or Apple API errors are all non-fatal here 

638 LOGGER.debug("Failed to trigger 2FA push notification: %s", error) 

639 return False 

640 

641 def validate_2fa_code(self, code): 

642 """Verifies a verification code received via Apple's 2FA system (HSA2).""" 

643 data = {"securityCode": {"code": code}} 

644 

645 headers = self._get_auth_headers({"Accept": "application/json"}) 

646 

647 if self.session_data.get("scnt"): 

648 headers["scnt"] = self.session_data.get("scnt") 

649 

650 if self.session_data.get("session_id"): 

651 headers["X-Apple-ID-Session-Id"] = self.session_data.get("session_id") 

652 

653 try: 

654 self.session.post( 

655 f"{self.auth_endpoint}/verify/trusteddevice/securitycode", 

656 data=json.dumps(data), 

657 headers=headers, 

658 ) 

659 except ICloudPyAPIResponseException as error: 

660 if error.code == -21669: 

661 # Wrong verification code 

662 LOGGER.error("Code verification failed.") 

663 return False 

664 raise 

665 

666 LOGGER.debug("Code verification successful.") 

667 

668 self.trust_session() 

669 return not self.requires_2sa 

670 

671 @property 

672 def security_key_challenge(self): 

673 """Apple's pending WebAuthn challenge, or None if it is not asking for one. 

674 

675 Returned when the account has security keys enrolled, in which case 

676 Apple issues this instead of delivering a 6-digit code. The dict 

677 carries ``challenge``, ``keyHandles`` and ``rpId``. 

678 """ 

679 headers = self._get_auth_headers({"Accept": "application/json"}) 

680 if self.session_data.get("scnt"): 

681 headers["scnt"] = self.session_data.get("scnt") 

682 if self.session_data.get("session_id"): 

683 headers["X-Apple-ID-Session-Id"] = self.session_data.get("session_id") 

684 

685 try: 

686 response = self.session.get(self.auth_endpoint, headers=headers) 

687 challenge = response.json().get("fsaChallenge") 

688 except (ICloudPyAPIResponseException, ValueError) as error: 

689 LOGGER.debug("Could not read security key challenge: %s", error) 

690 return None 

691 

692 if not (challenge and challenge.get("challenge") and challenge.get("keyHandles")): 

693 return None 

694 return challenge 

695 

696 @property 

697 def fido2_devices(self): 

698 """Attached FIDO2 devices. 

699 

700 Empty when the optional ``fido2`` package is not installed. 

701 """ 

702 try: 

703 from fido2.hid import CtapHidDevice 

704 except ImportError: 

705 LOGGER.debug("fido2 is not installed; no local devices available.") 

706 return [] 

707 return list(CtapHidDevice.list_devices()) 

708 

709 def confirm_security_key(self, assertion=None, device=None): 

710 """Complete two-factor authentication with a security key. 

711 

712 Signs the pending challenge with an attached FIDO2 device and 

713 submits the result. Pass ``assertion`` to submit one that has 

714 already been built -- by ``sign_security_key_challenge`` or by any 

715 other means -- in which case no device is used. 

716 

717 Returns True when the session no longer requires a second factor. 

718 """ 

719 if assertion is None: 

720 challenge = self.security_key_challenge 

721 if not challenge: 

722 msg = "Apple is not requesting a security key." 

723 raise ICloudPyFailedLoginException(msg) 

724 if device is None: 

725 devices = self.fido2_devices 

726 if not devices: 

727 msg = "No FIDO2 device found." 

728 raise ICloudPyFailedLoginException(msg) 

729 device = devices[0] 

730 assertion = self.sign_security_key_challenge(challenge, device) 

731 

732 headers = self._get_auth_headers({"Accept": "application/json"}) 

733 if self.session_data.get("scnt"): 

734 headers["scnt"] = self.session_data.get("scnt") 

735 if self.session_data.get("session_id"): 

736 headers["X-Apple-ID-Session-Id"] = self.session_data.get("session_id") 

737 

738 try: 

739 response = self.session.post( 

740 f"{self.auth_endpoint}/verify/security/key", 

741 data=json.dumps(assertion), 

742 headers=headers, 

743 ) 

744 status = response.status_code 

745 except ICloudPyAPIResponseException as error: 

746 status = error.code 

747 

748 if status not in SECURITY_KEY_ACCEPTED_STATUSES: 

749 LOGGER.error("Security key verification failed (%s).", status) 

750 return False 

751 

752 LOGGER.debug("Security key verification successful.") 

753 self.trust_session() 

754 return not self.requires_2sa 

755 

756 def sign_security_key_challenge(self, challenge, device): 

757 """Sign ``challenge`` with ``device`` and return the assertion payload. 

758 

759 Requires the ``fido2`` package and a physical touch on the key. The 

760 result is accepted by ``confirm_security_key``. 

761 """ 

762 from fido2.client import DefaultClientDataCollector, Fido2Client 

763 from fido2.webauthn import ( 

764 PublicKeyCredentialDescriptor, 

765 PublicKeyCredentialRequestOptions, 

766 PublicKeyCredentialType, 

767 UserVerificationRequirement, 

768 ) 

769 

770 rp_id = challenge.get("rpId") or "apple.com" 

771 client = Fido2Client( 

772 device, 

773 client_data_collector=DefaultClientDataCollector(SECURITY_KEY_ORIGIN), 

774 ) 

775 options = PublicKeyCredentialRequestOptions( 

776 challenge=_b64_decode(challenge["challenge"]), 

777 rp_id=rp_id, 

778 allow_credentials=[ 

779 PublicKeyCredentialDescriptor( 

780 id=_b64_decode(handle), 

781 type=PublicKeyCredentialType("public-key"), 

782 ) 

783 for handle in challenge["keyHandles"] 

784 ], 

785 user_verification=UserVerificationRequirement("discouraged"), 

786 ) 

787 result = client.get_assertion(options).get_response(0) 

788 return build_security_key_assertion(result, rp_id, challenge.get("requestId")) 

789 

790 def trust_session(self): 

791 """Request session trust to avoid user log in going forward.""" 

792 headers = self._get_auth_headers() 

793 

794 if self.session_data.get("scnt"): 

795 headers["scnt"] = self.session_data.get("scnt") 

796 

797 if self.session_data.get("session_id"): 

798 headers["X-Apple-ID-Session-Id"] = self.session_data.get("session_id") 

799 

800 try: 

801 self.session.get( 

802 f"{self.auth_endpoint}/2sv/trust", 

803 headers=headers, 

804 ) 

805 self._authenticate_with_token() 

806 return True 

807 except ICloudPyAPIResponseException: 

808 LOGGER.error("Session trust failed.") 

809 return False 

810 

811 def _get_webservice_url(self, ws_key): 

812 """Get webservice URL, raise an exception if not exists.""" 

813 if self._webservices.get(ws_key) is None: 

814 raise ICloudPyServiceNotActivatedException( 

815 "Webservice not available", 

816 ws_key, 

817 ) 

818 return self._webservices[ws_key]["url"] 

819 

820 @property 

821 def devices(self): 

822 """Returns all devices.""" 

823 service_root = self._get_webservice_url("findme") 

824 return FindMyiPhoneServiceManager(