diff --git a/src/sync.py b/src/sync.py index 98da35304..7879e1472 100644 --- a/src/sync.py +++ b/src/sync.py @@ -92,6 +92,10 @@ def __init__(self): self.enable_sync_drive = True self.enable_sync_photos = True self.last_send = None + # Whether a 2FA push has already been requested for the current + # re-auth episode. Reset to False on each successful authentication so + # a fresh episode triggers exactly one push (see _handle_2fa_required). + self.two_fa_triggered = False def _load_configuration(): @@ -649,7 +653,7 @@ def _send_usage_statistics(config, summary: SyncSummary) -> None: alive(config=config, data=usage_data) -def _handle_2fa_required(config, username: str, sync_state: SyncState): +def _handle_2fa_required(config, username: str, sync_state: SyncState, api): """ Handle 2FA authentication requirement. @@ -657,6 +661,8 @@ def _handle_2fa_required(config, username: str, sync_state: SyncState): config: Configuration dictionary username: iCloud username sync_state: Current sync state + api: Live ``ICloudPyService`` still in its 2FA-required state, used to + request a push notification to the user's trusted devices. Returns: bool: True if should continue (retry), False if should exit @@ -668,6 +674,20 @@ def _handle_2fa_required(config, username: str, sync_state: SyncState): LOGGER.info("retry_login_interval is < 0, exiting ...") return False + # Ask Apple to actually push a 2FA code to the trusted devices. Without this + # call the loop notified the user that re-auth was needed but never requested + # a code, so nothing was ever sent. Fire once per episode (two_fa_triggered is + # reset on successful auth) to avoid re-pushing every retry cycle -- the + # default interval is 600s -- and tripping Apple's rate limits. Best-effort: + # a failure here must not stop the retry loop. + if not sync_state.two_fa_triggered: + try: + api.trigger_2fa_push_notification() + LOGGER.info("Requested a 2FA push notification to your trusted devices.") + except Exception as e: # noqa: BLE001 + LOGGER.warning(f"Failed to request 2FA push notification: {e!s}") + sync_state.two_fa_triggered = True + _log_retry_time(sleep_for) server_region = config_parser.get_region(config=config) sync_state.last_send = notify.send( @@ -889,6 +909,10 @@ def sync(dry_run: bool = False, check_files: int | None = None): return if not api.requires_2sa: + # Authenticated: clear the 2FA trigger latch so a future + # re-auth episode requests a fresh push exactly once. + sync_state.two_fa_triggered = False + # Create summary for this sync cycle summary = SyncSummary() @@ -949,7 +973,7 @@ def sync(dry_run: bool = False, check_files: int | None = None): "Nothing to sync. Please add drive: and/or photos: section in config.yaml file.", ) else: - if not _handle_2fa_required(config, username, sync_state): + if not _handle_2fa_required(config, username, sync_state, api): break continue diff --git a/tests/test_sync.py b/tests/test_sync.py index e7c77ac72..8def7a76d 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -8,7 +8,7 @@ from copy import deepcopy from io import StringIO from types import SimpleNamespace -from unittest.mock import patch +from unittest.mock import Mock, patch from icloudpy import exceptions @@ -180,6 +180,40 @@ def test_sync_2fa_required( self.assertTrue(len(captured.records) > 1) self.assertTrue(len([e for e in captured[1] if "2FA is required" in e]) > 0) + def _run_2fa_handler(self, sync_state, api): + """Invoke the private 2FA handler with this test's config/user.""" + return sync._handle_2fa_required(self.config, data.REQUIRES_2FA_USER, sync_state, api) # noqa: SLF001 + + @patch("src.sync.sleep") + @patch("src.sync.notify.send", return_value=None) + def test_handle_2fa_requests_push_once_per_episode(self, _mock_notify, _mock_sleep): + """A 2FA push is requested exactly once per re-auth episode.""" + sync_state = sync.SyncState() + api = Mock() + + with self.assertLogs() as captured: + self.assertTrue(self._run_2fa_handler(sync_state, api)) + api.trigger_2fa_push_notification.assert_called_once() + self.assertTrue(sync_state.two_fa_triggered) + self.assertTrue(any("Requested a 2FA push notification" in e for e in captured[1])) + + # Second retry within the same episode must NOT push again. + self.assertTrue(self._run_2fa_handler(sync_state, api)) + api.trigger_2fa_push_notification.assert_called_once() + + @patch("src.sync.sleep") + @patch("src.sync.notify.send", return_value=None) + def test_handle_2fa_push_failure_is_non_fatal(self, _mock_notify, _mock_sleep): + """A failing trigger is swallowed; the retry loop still continues.""" + sync_state = sync.SyncState() + api = Mock() + api.trigger_2fa_push_notification.side_effect = RuntimeError("no trusted device") + + with self.assertLogs() as captured: + self.assertTrue(self._run_2fa_handler(sync_state, api)) + self.assertTrue(sync_state.two_fa_triggered) + self.assertTrue(any("Failed to request 2FA push notification" in e for e in captured[1])) + @patch("src.sync.sleep") @patch(target="keyring.get_password", return_value=data.VALID_PASSWORD) @patch(target="src.config_parser.get_username", return_value=data.AUTHENTICATED_USER) @@ -606,7 +640,6 @@ def test_perform_photos_sync_no_errors_when_all_succeed(self, mock_sync_photos): self.assertFalse(stats.has_errors()) self.assertEqual(len(stats.errors), 0) - @patch("src.sync.notify.send_sync_summary", side_effect=RuntimeError("notify failure")) @patch("src.sync._perform_photos_sync") @patch("src.sync._perform_drive_sync", return_value=DriveStats(files_downloaded=1))