fix(2fa): request a push notification when re-auth is required

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,14 +653,16 @@ 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.
 
     Args:
         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
 

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))