webui handles auth errors #1195 (#1210)

Author: AndreyNikiforov

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- fix: failed auth terminate webui [#1195](https://github.com/icloud-photos-downloader/icloud_photos_downloader/issues/1195)
 - fix: disable raw password auth fallback [#1176](https://github.com/icloud-photos-downloader/icloud_photos_downloader/issues/1176)
 
 ## 1.29.3 (2025-08-09)

src/icloudpd/authentication.py

@@ -11,6 +11,7 @@
 from icloudpd.mfa_provider import MFAProvider
 from icloudpd.status import Status, StatusExchange
 from pyicloud_ipd.base import PyiCloudService
+from pyicloud_ipd.exceptions import PyiCloudFailedMFAException
 from pyicloud_ipd.file_match import FileMatchPolicy
 from pyicloud_ipd.raw_policy import RawTreatmentPolicy
 
@@ -123,8 +124,7 @@ def request_2fa(icloud: PyiCloudService, logger: logging.Logger) -> None:
     device_index_alphabet = "abcdefghijklmnopqrstuvwxyz"
     if devices_count > 0:
         if devices_count > len(device_index_alphabet):
-            logger.error("Too many trusted devices for authentication")
-            sys.exit(1)
+            raise PyiCloudFailedMFAException("Too many trusted devices for authentication")
 
         for i, device in enumerate(devices):
             click.echo(f"  {device_index_alphabet[i]}: {device.obfuscated_number}")
@@ -175,8 +175,7 @@ def request_2fa(icloud: PyiCloudService, logger: logging.Logger) -> None:
             device_index = device_index_alphabet.index(index_or_code)
             device = devices[device_index]
             if not icloud.send_2fa_code_sms(device.id):
-                logger.error("Failed to send two-factor authentication code")
-                sys.exit(1)
+                raise PyiCloudFailedMFAException("Failed to send two-factor authentication code")
             while True:
                 code: str = click.prompt(
                     "Please enter two-factor authentication code that you received over SMS",
@@ -186,12 +185,10 @@ def request_2fa(icloud: PyiCloudService, logger: logging.Logger) -> None:
                 click.echo("Invalid code, should be six digits. Try again")
 
             if not icloud.validate_2fa_code_sms(device.id, code):
-                logger.error("Failed to verify two-factor authentication code")
-                sys.exit(1)
+                raise PyiCloudFailedMFAException("Failed to verify two-factor authentication code")
         else:
             if not icloud.validate_2fa_code(index_or_code):
-                logger.error("Failed to verify two-factor authentication code")
-                sys.exit(1)
+                raise PyiCloudFailedMFAException("Failed to verify two-factor authentication code")
     else:
         while True:
             code = click.prompt(
@@ -201,8 +198,7 @@ def request_2fa(icloud: PyiCloudService, logger: logging.Logger) -> None:
                 break
             click.echo("Invalid code, should be six digits. Try again")
         if not icloud.validate_2fa_code(code):
-            logger.error("Failed to verify two-factor authentication code")
-            sys.exit(1)
+            raise PyiCloudFailedMFAException("Failed to verify two-factor authentication code")
     logger.info(
         "Great, you're all set up. The script can now be run without "
         "user interaction until 2FA expires.\n"
@@ -217,40 +213,42 @@ def request_2fa_web(
 ) -> None:
     """Request two-factor authentication through Webui."""
     if not status_exchange.replace_status(Status.NO_INPUT_NEEDED, Status.NEED_MFA):
-        logger.error("Expected NO_INPUT_NEEDED, but got something else")
-        return
+        raise PyiCloudFailedMFAException(
+            f"Expected NO_INPUT_NEEDED, but got {status_exchange.get_status()}"
+        )
 
     # wait for input
     while True:
         status = status_exchange.get_status()
         if status == Status.NEED_MFA:
             time.sleep(1)
+            continue
         else:
-            break
+            pass
 
-    if status_exchange.replace_status(Status.SUPPLIED_MFA, Status.CHECKING_MFA):
-        code = status_exchange.get_payload()
-        if not code:
-            logger.error("Internal error: did not get code for SUPPLIED_MFA status")
-            status_exchange.replace_status(
-                Status.CHECKING_MFA, Status.NO_INPUT_NEEDED
-            )  # TODO Error
-            return
+        if status_exchange.replace_status(Status.SUPPLIED_MFA, Status.CHECKING_MFA):
+            code = status_exchange.get_payload()
+            if not code:
+                raise PyiCloudFailedMFAException(
+                    "Internal error: did not get code for SUPPLIED_MFA status"
+                )
 
-        if not icloud.validate_2fa_code(code):
-            logger.error("Failed to verify two-factor authentication code")
-            status_exchange.replace_status(
-                Status.CHECKING_MFA, Status.NO_INPUT_NEEDED
-            )  # TODO Error
-            return
-        status_exchange.replace_status(Status.CHECKING_MFA, Status.NO_INPUT_NEEDED)  # done
-
-        logger.info(
-            "Great, you're all set up. The script can now be run without "
-            "user interaction until 2FA expires.\n"
-            "You can set up email notifications for when "
-            "the two-factor authentication expires.\n"
-            "(Use --help to view information about SMTP options.)"
-        )
-    else:
-        logger.error("Failed to change status")
+            if not icloud.validate_2fa_code(code):
+                if status_exchange.set_error("Failed to verify two-factor authentication code"):
+                    # that will loop forever
+                    # TODO give user an option to restart auth in case they missed code
+                    continue
+                else:
+                    raise PyiCloudFailedMFAException("Failed to chage status of invalid code")
+            else:
+                status_exchange.replace_status(Status.CHECKING_MFA, Status.NO_INPUT_NEEDED)  # done
+
+                logger.info(
+                    "Great, you're all set up. The script can now be run without "
+                    "user interaction until 2FA expires.\n"
+                    "You can set up email notifications for when "
+                    "the two-factor authentication expires.\n"
+                    "(Use --help to view information about SMTP options.)"
+                )
+        else:
+            raise PyiCloudFailedMFAException("Failed to change status")

src/icloudpd/base.py

@@ -66,6 +66,7 @@
 from pyicloud_ipd.base import PyiCloudService
 from pyicloud_ipd.exceptions import (
     PyiCloudFailedLoginException,
+    PyiCloudFailedMFAException,
     PyiCloudServiceNotActivatedException,
     PyiCloudServiceUnavailableException,
 )
@@ -169,43 +170,39 @@ def ask_password_in_console(_user: str) -> str | None:
 
 
 def get_password_from_webui(
-    logger: Logger, status_exchange: StatusExchange
-) -> Callable[[str], str | None]:
-    def _intern(_user: str) -> str | None:
-        """Request two-factor authentication through Webui."""
-        if not status_exchange.replace_status(Status.NO_INPUT_NEEDED, Status.NEED_PASSWORD):
-            logger.error("Expected NO_INPUT_NEEDED, but got something else")
+    logger: Logger, status_exchange: StatusExchange, _user: str
+) -> str | None:
+    """Request two-factor authentication through Webui."""
+    if not status_exchange.replace_status(Status.NO_INPUT_NEEDED, Status.NEED_PASSWORD):
+        logger.error("Expected NO_INPUT_NEEDED, but got something else")
+        return None
+
+    # wait for input
+    while True:
+        status = status_exchange.get_status()
+        if status == Status.NEED_PASSWORD:
+            time.sleep(1)
+        else:
+            break
+    if status_exchange.replace_status(Status.SUPPLIED_PASSWORD, Status.CHECKING_PASSWORD):
+        password = status_exchange.get_payload()
+        if not password:
+            logger.error("Internal error: did not get password for SUPPLIED_PASSWORD status")
+            status_exchange.replace_status(
+                Status.CHECKING_PASSWORD, Status.NO_INPUT_NEEDED
+            )  # TODO Error
             return None
+        return password
 
-        # wait for input
-        while True:
-            status = status_exchange.get_status()
-            if status == Status.NEED_PASSWORD:
-                time.sleep(1)
-            else:
-                break
-        if status_exchange.replace_status(Status.SUPPLIED_PASSWORD, Status.CHECKING_PASSWORD):
-            password = status_exchange.get_payload()
-            if not password:
-                logger.error("Internal error: did not get password for SUPPLIED_PASSWORD status")
-                status_exchange.replace_status(
-                    Status.CHECKING_PASSWORD, Status.NO_INPUT_NEEDED
-                )  # TODO Error
-                return None
-            return password
-
-        return None  # TODO
+    return None  # TODO
 
-    return _intern
 
+def update_password_status_in_webui(status_exchange: StatusExchange, _u: str, _p: str) -> None:
+    status_exchange.replace_status(Status.CHECKING_PASSWORD, Status.NO_INPUT_NEEDED)
 
-def update_password_status_in_webui(status_exchange: StatusExchange) -> Callable[[str, str], None]:
-    def _intern(_u: str, _p: str) -> None:
-        # TODO we are not handling wrong passwords...
-        status_exchange.replace_status(Status.CHECKING_PASSWORD, Status.NO_INPUT_NEEDED)
-        return None
 
-    return _intern
+def update_auth_error_in_webui(status_exchange: StatusExchange, error: str) -> bool:
+    return status_exchange.set_error(error)
 
 
 # def get_click_param_by_name(_name: str, _params: List[Parameter]) -> Optional[Parameter]:
@@ -234,6 +231,7 @@ def password_provider_generator(
 ) -> Dict[str, Tuple[Callable[[str], str | None], Callable[[str, str], None]]]:
     def _map(provider: str) -> Tuple[Callable[[str], str | None], Callable[[str, str], None]]:
         if provider == "webui":
+            # ask_password_in_console will be replaced once we setup web
             return (ask_password_in_console, dummy_password_writter)
         if provider == "console":
             return (ask_password_in_console, dummy_password_writter)
@@ -805,8 +803,8 @@ def main(
         if "webui" in password_providers:
             # replace
             password_providers["webui"] = (
-                get_password_from_webui(logger, status_exchange),
-                update_password_status_in_webui(status_exchange),
+                partial(get_password_from_webui, logger, status_exchange),
+                partial(update_password_status_in_webui, status_exchange),
             )
 
         # hacky way to inject logger
@@ -1559,13 +1557,31 @@ def should_break(counter: Counter) -> bool:
         except PyiCloudFailedLoginException as _error:
             logger.info("Invalid email/password combination.")
             dump_responses(logger.debug, captured_responses)
-            if "webui" in password_providers and mfa_provider == MFAProvider.WEBUI:
-                pass
+            if "webui" in password_providers:
+                update_auth_error_in_webui(status_exchange, "Invalid email/password combination.")
+                continue
+            else:
+                return 1
+        except PyiCloudFailedMFAException as error:
+            logger.info(str(error))
+            dump_responses(logger.debug, captured_responses)
+            if mfa_provider == MFAProvider.WEBUI:
+                update_auth_error_in_webui(status_exchange, str(error))
+                continue
             else:
                 return 1
         except (PyiCloudServiceNotActivatedException, PyiCloudServiceUnavailableException) as error:
             logger.info(error)
             dump_responses(logger.debug, captured_responses)
+            # webui will display error and wait for password again
+            if "webui" in password_providers or mfa_provider == MFAProvider.WEBUI:
+                if update_auth_error_in_webui(status_exchange, str(error)):
+                    # retry if it was during auth
+                    continue
+                else:
+                    pass
+            else:
+                pass
             # it not watching then return error
             if not watch_interval:
                 return 1
@@ -1575,6 +1591,17 @@ def should_break(counter: Counter) -> bool:
             logger.info("Cannot connect to Apple iCloud service")
             dump_responses(logger.debug, captured_responses)
             # logger.debug(error)
+            # webui will display error and wait for password again
+            if "webui" in password_providers or mfa_provider == MFAProvider.WEBUI:
+                if update_auth_error_in_webui(
+                    status_exchange, "Cannot connect to Apple iCloud service"
+                ):
+                    # retry if it was during auth
+                    continue
+                else:
+                    pass
+            else:
+                pass
             # it not watching then return error
             if not watch_interval:
                 return 1

src/icloudpd/server/__init__.py

@@ -26,14 +26,19 @@ def get_status() -> Response | str:
         _status = _status_exchange.get_status()
         _config = _status_exchange.get_config()
         _progress = _status_exchange.get_progress()
+        _error = _status_exchange.get_error()
         if _status == Status.NO_INPUT_NEEDED:
             return render_template(
-                "no_input.html", status=_status, progress=_progress, config=vars(_config)
+                "no_input.html",
+                status=_status,
+                error=_error,
+                progress=_progress,
+                config=vars(_config),
             )
         if _status == Status.NEED_MFA:
-            return render_template("code.html")
+            return render_template("code.html", error=_error)
         if _status == Status.NEED_PASSWORD:
-            return render_template("password.html", config=_config)
+            return render_template("password.html", error=_error, config=_config)
         return render_template("status.html", status=_status)
 
     @app.route("/code", methods=["POST"])

src/icloudpd/server/templates/code.html

@@ -1,6 +1,13 @@
 <form hx-post="/code" hx-swap="outerHTML" class="row align-items-center" hx-target-error="#toast-content">
     <fieldset>
         <legend>Authentication</legend>
+        {% if error %}
+        <ul class="list-group list-group-flush">
+            <li class="list-group-item d-flex justify-content-between align-items-center">
+                <div class="fw-bold">{{ error }}</div>
+            </li>
+        </ul>
+        {% endif %}
         <div class="col-12 mb-3">
           <label for="code" class="form-label">Two-Factor code of the user {{ config.username }}</label>
           <input type="text" class="form-control" id="code" name="code" placeholder="Enter Two-Factor Code">

src/icloudpd/server/templates/no_input.html

@@ -7,6 +7,13 @@
                     <div class="card-header text-bg-primary">
                         Status
                     </div>
+                    {% if error %}
+                    <ul class="list-group list-group-flush">
+                        <li class="list-group-item d-flex justify-content-between align-items-center">
+                            <div class="fw-bold">{{ error }}</div>
+                        </li>
+                    </ul>
+                    {% endif %}
                     <ul class="list-group list-group-flush">
                         <li class="list-group-item d-flex justify-content-between align-items-center">
                             <div class="fw-bold">No input is needed</div>

src/icloudpd/server/templates/password.html

@@ -1,6 +1,13 @@
 <form hx-post="/password" hx-swap="outerHTML" class="row align-items-center" hx-target-error="#toast-content">
     <fieldset>
         <legend>Authentication</legend>
+        {% if error %}
+        <ul class="list-group list-group-flush">
+            <li class="list-group-item d-flex justify-content-between align-items-center">
+                <div class="fw-bold">{{ error }}</div>
+            </li>
+        </ul>
+        {% endif %}
         <div class="col-12 mb-3">
           <label for="password" class="form-label">Password of the user {{ config.username }}</label>
           <input type="password" class="form-control" id="password" name="password" placeholder="Enter password">

src/icloudpd/status.py

@@ -23,6 +23,7 @@ def __init__(self) -> None:
         self.lock = Lock()
         self._status = Status.NO_INPUT_NEEDED
         self._payload: str | None = None
+        self._error: str | None = None
         self._config: Config | None = None
         self._progress = Progress()
 
@@ -47,6 +48,7 @@ def set_payload(self, payload: str) -> bool:
             self._status = (
                 Status.SUPPLIED_MFA if self._status == Status.NEED_MFA else Status.SUPPLIED_PASSWORD
             )
+            self._error = None
             return True
 
     def get_payload(self) -> str | None:
@@ -61,6 +63,30 @@ def get_payload(self) -> str | None:
 
             return self._payload
 
+    def set_error(self, error: str) -> bool:
+        with self.lock:
+            if self._status != Status.CHECKING_MFA and self._status != Status.CHECKING_PASSWORD:
+                return False
+
+            self._error = error
+            self._status = (
+                Status.NO_INPUT_NEEDED
+                if self._status == Status.CHECKING_PASSWORD
+                else Status.NEED_MFA
+            )
+            return True
+
+    def get_error(self) -> str | None:
+        with self.lock:
+            if self._status not in [
+                Status.NO_INPUT_NEEDED,
+                Status.NEED_PASSWORD,
+                Status.NEED_MFA,
+            ]:
+                return None
+
+            return self._error
+
     def set_config(self, config: Config) -> None:
         with self.lock:
             self._config = config