Coverage for src/web.py: 100%
594 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-08 21:05 +0000
1"""Web UI for icloud-docker.
3Goal — give the user a single page they can hit from any device to:
4 1) (primary) authenticate / re-authenticate Apple ID + 2FA;
5 2) (secondary) confirm config paths, mount markers, and last-sync status;
6 3) (tertiary) tail the recent log lines.
8The web server runs in a daemon thread spawned from ``main.py`` alongside
9the existing ``sync.sync()`` loop. The two share state through the
10filesystem (keyring, session cookies, log file). No new persistence layer.
12Designed for **LAN- or proxy-trusted** exposure. There is no built-in
13login on this UI — put Cloudflare Access / Authelia / Tailscale in front
14when exposing publicly. Opt-out via ``app.web_ui.enabled: false`` in
15``config.yaml``.
16"""
18__author__ = "Mandar Patil (mandarons@pm.me)"
20import base64
21import hmac
22import json
23import os
24import re
25import secrets
26import shutil
27import struct
28import tempfile
29import threading
30import time
31from typing import Any
33from flask import Flask, jsonify, redirect, render_template, request, url_for
34from werkzeug.serving import make_server
36from src import (
37 DEFAULT_CONFIG_FILE_PATH,
38 DEFAULT_COOKIE_DIRECTORY,
39 ENV_CONFIG_FILE_PATH_KEY,
40 config_parser,
41 get_logger,
42 read_config,
43 web_signals,
44)
45from src.config_utils import get_config_value_or_none
47LOGGER = get_logger()
49# Module-level holder for the live icloudpy session created during
50# POST /auth/password, so POST /auth/code can call validate_2fa_code on
51# the SAME session. Cleared after a successful trust_session or via
52# POST /auth/reset.
53_PENDING_AUTH: dict[str, Any] = {}
54_AUTH_LOCK = threading.Lock()
56# Drop stale pending auth after this many seconds. The submitted Apple ID
57# password sits in process memory (in ``_PENDING_AUTH["password"]``) while
58# waiting for the user to enter their 2FA code; without an expiry it would
59# linger indefinitely if the user closed the browser tab mid-flow. 10 min
60# is generous for typing a code -- and short enough that a forgotten
61# session evaporates before the next sync cycle picks up the keyring.
62_PENDING_AUTH_TTL_SECONDS = 600
65def _wake_sync_loop() -> None:
66 """Cut short the auth-retry wait after a re-auth succeeds.
68 Without this the loop serves out the rest of an interval that began
69 before the problem was solved -- the user completes the sign-in and
70 the dashboard keeps reporting the sync as stopped for up to
71 ``retry_login_interval``.
73 Best-effort: a missed nudge costs a delay, never correctness.
74 """
75 try:
76 web_signals.record_reauth_completed()
77 except Exception as e: # noqa: BLE001 - never fail a successful sign-in
78 LOGGER.debug(f"could not signal the completed re-auth: {e!s}")
81def _pending_auth_is_stale() -> bool:
82 """True when the in-memory password is older than the TTL.
84 Caller must hold ``_AUTH_LOCK``. Returns False for an empty dict
85 (nothing to expire) and for entries that pre-date stashed_at
86 bookkeeping (defensive — we never penalise a fresh stash).
87 """
88 if not _PENDING_AUTH:
89 return False
90 stashed_at = _PENDING_AUTH.get("stashed_at")
91 if stashed_at is None:
92 return False
93 return (time.monotonic() - stashed_at) > _PENDING_AUTH_TTL_SECONDS
96def _clear_pending_auth_unlocked() -> None:
97 """Forget the pending auth, and a security-key ceremony's private cookie
98 jar with it -- it holds that ceremony's Apple session cookies, and an
99 abandoned ceremony would otherwise leave them in the temp dir until the
100 container restarts. Caller holds the lock."""
101 _discard_ceremony_dir(_PENDING_AUTH.get("cookie_dir"))
102 _PENDING_AUTH.clear()
105def _expire_stale_pending_auth_unlocked() -> None:
106 """If the pending auth is older than the TTL, wipe it. Caller holds the lock."""
107 if _pending_auth_is_stale():
108 LOGGER.info("Web UI: expiring stale _PENDING_AUTH past TTL.")
109 _clear_pending_auth_unlocked()
112# CSRF defence. Threat model: even with the default host pinned to
113# 127.0.0.1, a user who opts into LAN exposure (host: 0.0.0.0) AND lacks
114# a proper auth proxy in front would otherwise be vulnerable to a
115# same-network attacker who tricks them into loading a page that posts
116# to ``/auth/refresh-trust`` or ``/api/sync``. Double-submit cookie
117# pattern: per-process random token, set as a SameSite=Strict cookie,
118# required on every state-changing POST as either a form field
119# ``csrf_token`` or an ``X-CSRF-Token`` header. SameSite=Strict alone
120# already blocks the cross-site cookie send in modern browsers; the
121# server-side compare is belt-and-braces for older clients.
122_CSRF_TOKEN = secrets.token_urlsafe(32)
123_CSRF_COOKIE_NAME = "csrf_token"
126def _get_csrf_token() -> str:
127 """Expose the token to templates (so forms can embed it) and to
128 tests (so they can post it). Process-lifetime, regenerated on
129 restart -- enough for a single-user operator console."""
130 return _CSRF_TOKEN
133def _require_csrf() -> tuple[Any, int] | None:
134 """Validate CSRF token on the current request. Returns ``None``
135 when the request is allowed, or a ``(response, status)`` tuple
136 when it should be rejected. Use at the top of every state-
137 changing endpoint:
139 rejection = _require_csrf()
140 if rejection is not None:
141 return rejection
143 **Calling from a script / monitor.** Every state-changing endpoint
144 needs BOTH the CSRF cookie and a matching token, so a bare
145 ``curl -X POST /api/sync -d service=drive`` gets a 403. The cookie
146 is only set by a prior page load, so fetch it first and echo it
147 back in the ``X-CSRF-Token`` header::
149 curl -c jar -s http://127.0.0.1:8080/ >/dev/null
150 TOKEN=$(awk '/csrf_token/ {print $7}' jar)
151 curl -b jar -H "X-CSRF-Token: $TOKEN" \
152 -X POST http://127.0.0.1:8080/api/sync -d service=drive
154 The 403 bodies name which leg failed ("CSRF cookie missing or
155 stale" vs "CSRF token mismatch") so the fix is obvious.
156 """
157 cookie = request.cookies.get(_CSRF_COOKIE_NAME)
158 submitted = request.form.get("csrf_token") or request.headers.get("X-CSRF-Token")
159 # The cookie must match this process's token AND the submitted value
160 # must match the cookie. Browsers won't include a SameSite=Strict
161 # cookie on a cross-site POST, so the cookie absence alone is the
162 # primary signal; the form/header echo is the belt-and-braces leg.
163 if not cookie or not hmac.compare_digest(cookie, _CSRF_TOKEN):
164 return jsonify({"error": "CSRF cookie missing or stale"}), 403
165 if not submitted or not hmac.compare_digest(submitted, cookie):
166 return jsonify({"error": "CSRF token mismatch"}), 403
167 return None
170def _current_config_path() -> str:
171 """Resolve the active config path the same way sync.py does."""
172 return os.environ.get(ENV_CONFIG_FILE_PATH_KEY, DEFAULT_CONFIG_FILE_PATH)
175def _load_current_config() -> dict | None:
176 """Re-read config.yaml fresh on every request so edits show up live.
178 Defensive: mandarons' ``read_config`` reaches into
179 ``config["app"]["credentials"]["username"]`` unconditionally and
180 crashes if the credentials block is missing. Catch that so a partial
181 config (e.g. fresh install with only ``app.logger`` set) still lets
182 the web UI render the setup-needed state instead of 500-ing.
183 """
184 path = _current_config_path()
185 if not os.path.isfile(path):
186 return None
187 try:
188 return read_config(config_path=path)
189 except Exception as e:
190 # Broad on purpose: ``/api/health`` exists for external monitors
191 # and must be robust against any config-loading failure (YAML
192 # parse errors, permission denied, missing credentials block,
193 # ruamel internals raising). A 500 on /api/health blinds the
194 # monitor; rendering a "config error" state lets the user fix
195 # it via the UI.
196 LOGGER.warning(f"Web UI: read_config failed: {e!s}")
197 return None
200def _get_marker_filename(config: dict) -> str:
201 """Marker filename from ``app.mount_marker_filename`` (default ``.mounted``)."""
202 return config_parser.get_mount_marker_filename(config=config)
205def _get_require_mount_marker(config: dict, service: str) -> bool:
206 """``{drive,photos}.require_mount_marker`` for the given service."""
207 getter = getattr(config_parser, f"get_{service}_require_mount_marker")
208 return bool(getter(config=config))
211def _get_library_destinations(config: dict) -> dict[str, str]:
212 """``photos.library_destinations`` mapping (empty dict when unset)."""
213 return config_parser.get_photos_library_destinations(config=config) or {}
216def _get_configured_libraries(config: dict) -> list[str]:
217 """Libraries the operator named: ``photos.filters.libraries``, then any
218 ``photos.library_destinations`` keys, in config order.
220 Read directly rather than through ``get_photos_libraries_filter``, which
221 logs "downloading all libraries" whenever the filter is unset -- once per
222 dashboard load.
223 """
224 named = get_config_value_or_none(config=config, config_path=["photos", "filters", "libraries"]) or []
225 if isinstance(named, str):
226 # ``libraries: PrimarySync`` rather than a list -- iterating the
227 # string would list one library per character.
228 named = [named]
229 named = [n for n in named if isinstance(n, str)]
230 return list(dict.fromkeys(named + list(_get_library_destinations(config=config))))
233def _build_service(config: dict, service: str, marker_filename: str) -> dict[str, Any]:
234 """Compose a single service entry (Photos or Drive) for /api/status."""
235 if service == "photos":
236 # Read-only on purpose: ``prepare_*_destination`` calls
237 # ``join_and_ensure_path`` (mkdir), so a plain ``GET /api/status``
238 # would write to disk -- and 500 the whole dashboard on a
239 # read-only destination mount. Composing the non-mutating getters
240 # lets a read-only or absent destination degrade to
241 # ``destination_exists: false`` instead.
242 destination = os.path.join(
243 config_parser.get_root_destination_path(config=config),
244 config_parser.get_photos_destination_path(config=config),
245 )
246 interval = config_parser.get_photos_sync_interval(
247 config=config,
248 log_messages=False,
249 )
250 name = "Photos"
251 library_destinations = _get_library_destinations(config=config)
252 else:
253 destination = os.path.join(
254 config_parser.get_root_destination_path(config=config),
255 config_parser.get_drive_destination_path(config=config),
256 )
257 interval = config_parser.get_drive_sync_interval(
258 config=config,
259 log_messages=False,
260 )
261 name = "Drive"
262 library_destinations = {}
264 marker_path = os.path.join(destination, marker_filename)
265 state = web_signals.get_sync_state(service=service)
266 stats = None
267 if state:
268 completed_at = state.get("completed_at")
269 stats = {
270 "last_sync_relative": (web_signals.format_relative_time(completed_at) if completed_at else None),
271 "files_downloaded": state.get("files_downloaded"),
272 "files_skipped": state.get("files_skipped"),
273 "files_removed": state.get("files_removed"),
274 # Named for what it is: the last COMPLETED cycle's
275 # downloaded+skipped total. The state record overwrites per
276 # cycle rather than accumulating, so this is not a running
277 # count of files on disk.
278 "last_cycle_total": (
279 (state.get("files_downloaded") or 0) + (state.get("files_skipped") or 0)
280 if (state.get("files_downloaded") is not None or state.get("files_skipped") is not None)
281 else None
282 ),
283 "errors": state.get("errors", 0),
284 "duration_seconds": state.get("duration_seconds"),
285 }
286 return {
287 "name": name,
288 "destination": destination,
289 "destination_exists": os.path.isdir(destination),
290 "sync_interval_s": interval,
291 "require_mount_marker": _get_require_mount_marker(
292 config=config,
293 service=service,
294 ),
295 "marker_present": os.path.isfile(marker_path),
296 "marker_path": marker_path,
297 "library_destinations": library_destinations,
298 "libraries": (
299 _build_libraries(_get_configured_libraries(config=config), library_destinations)
300 if service == "photos"
301 else []
302 ),
303 "indexing": _photos_indexing() if service == "photos" else None,
304 "stats": stats,
305 "force_sync_pending": service in web_signals.pending_force_syncs(),
306 }
309def _logger_filename(config: dict | None) -> str:
310 """Resolve where ``sync.py`` is writing log lines. Best-effort.
312 Reads ``app.logger.filename`` directly off the config dict to avoid
313 introducing a new ``config_parser`` helper just for this — keeps the
314 upstream PR diff small.
315 """
316 if not config:
317 return ""
318 try:
319 return config.get("app", {}).get("logger", {}).get("filename", "") or ""
320 except AttributeError:
321 return ""
324def _tail_log_file(path: str, lines: int = 200) -> list[str]:
325 """Return the last ``lines`` lines of ``path``.
327 Best-effort: missing path, unreadable file, or decode failure all
328 return an empty list. Reads from the end in 8 KiB blocks so the cost
329 is bounded by ``lines * average_line_length`` rather than file size.
330 """
331 if not path or not os.path.isfile(path):
332 return []
333 try:
334 with open(path, "rb") as f:
335 f.seek(0, os.SEEK_END)
336 size = f.tell()
337 block = 8192
338 data = b""
339 while size > 0 and data.count(b"\n") <= lines:
340 read_size = min(block, size)
341 size -= read_size
342 f.seek(size)
343 data = f.read(read_size) + data
344 return data.decode("utf-8", errors="replace").splitlines()[-lines:]
345 except OSError as e:
346 LOGGER.warning(f"Web UI could not tail log {path}: {e!s}")
347 return []
350def _build_status(config: dict | None) -> dict[str, Any]:
351 """Compose the payload returned by /api/status (and consumed by the
352 dashboard template)."""
353 if not config:
354 return {
355 "config_loaded": False,
356 "config_path": _current_config_path(),
357 "username": None,
358 "services": [],
359 }
361 marker_filename = _get_marker_filename(config=config)
362 services = []
363 if "photos" in config:
364 services.append(
365 _build_service(
366 config=config,
367 service="photos",
368 marker_filename=marker_filename,
369 ),
370 )
371 if "drive" in config:
372 services.append(
373 _build_service(
374 config=config,
375 service="drive",
376 marker_filename=marker_filename,
377 ),
378 )
380 username = config_parser.get_username(config=config)
381 trust = web_signals.get_trust_state()
382 trust_expires_at = trust.get("expires_at")
383 trust_days_remaining: int | None = None
384 if trust_expires_at:
385 try:
386 import datetime
388 exp = datetime.datetime.fromisoformat(trust_expires_at)
389 now = datetime.datetime.now(tz=exp.tzinfo) if exp.tzinfo else datetime.datetime.now()
390 trust_days_remaining = (exp - now).days
391 except (
392 ValueError,
393 TypeError,
394 ): # pragma: no cover -- defensive against malformed iso
395 trust_days_remaining = None
396 return {
397 "config_loaded": True,
398 "config_path": _current_config_path(),
399 "username": username,
400 "region": config_parser.get_region(config=config),
401 "marker_filename": marker_filename,
402 "services": services,
403 "auth_state": _detect_auth_state(username=username),
404 # Which mode the container is in, and why the loop says it is stuck.
405 # Without these the pill tells a password-free install that its
406 # keyring is populated and blames every stoppage on a second factor.
407 "password_stored": _password_is_stored(username) if username else False,
408 "auth_blocked_reason": web_signals.get_auth_blocked().get("reason"),
409 "auth_method": web_signals.get_auth_method(username) if username else None,
410 "force_sync_pending": web_signals.pending_force_syncs(),
411 "trust_expires_at": trust_expires_at,
412 "trust_days_remaining": trust_days_remaining,
413 }
416def _photos_indexing() -> dict[str, Any] | None:
417 """The Apple-side indexing wait Photos is in, if it is in one.
419 None when it is not, so the dashboard shows nothing in the normal
420 case. Service-level rather than per-library: no library is readable
421 while this is true (see ``sync_photos.is_photos_indexing``).
422 """
423 entry = web_signals.get_photos_indexing()
424 if not entry.get("waiting"):
425 return None
426 since = entry.get("since")
427 return {"since_relative": web_signals.format_relative_time(since) if since else None}
430def _build_libraries(configured: list[str], library_destinations: dict[str, str]) -> list[dict[str, Any]]:
431 """One row per photo library: where it goes and how it last went.
433 Only libraries this container actually syncs are listed: the ones the
434 operator named in config, plus any that have completed a sync or are
435 syncing right now. An account can carry libraries that never serve
436 anything -- there is nothing to do with them, so they are left out
437 rather than shown as a permanent failure. The log still records every
438 attempt, and naming a library in config always lists it.
440 Configured libraries come first, in config order.
441 """
442 states = web_signals.get_library_states()
443 rows = []
444 active = sorted(
445 n
446 for n, entry in states.items()
447 if n not in configured and (entry.get("completed_at") or entry.get("state") == "syncing")
448 )
449 for name in configured + active:
450 entry = states.get(name, {})
451 completed_at = entry.get("completed_at")
452 rows.append(
453 {
454 "name": name,
455 "subdir": library_destinations.get(name),
456 "state": entry.get("state"),
457 "error": entry.get("error"),
458 "completed_relative": (
459 web_signals.format_relative_time(completed_at) if completed_at else None
460 ),
461 },
462 )
463 return rows
466def _password_is_stored(username: str) -> bool:
467 """True when a password is cached for this Apple ID."""
468 try:
469 from icloudpy import utils as icloudpy_utils
471 return bool(icloudpy_utils.password_exists_in_keyring(username))
472 except Exception as e: # pragma: no cover - advisory, never breaks a render
473 LOGGER.debug(f"Web UI keyring probe raised: {e!s}")
474 return False
477def _saved_session_exists(username: str) -> bool:
478 r"""True when this Apple ID has a session file carrying a session token.
480 icloudpy names the file after the apple id with every non-``\w``
481 character dropped, plus ``.session`` (``ICloudPyService.session_path``),
482 and resumes from its ``session_token``. A file for *some other* account,
483 or one without that key, is not a credential this container can use --
484 and in a password-free setup it is the only credential there is.
485 """
486 sanitised = "".join(c for c in username if re.match(r"\w", c))
487 try:
488 with open(
489 os.path.join(DEFAULT_COOKIE_DIRECTORY, f"{sanitised}.session"),
490 encoding="utf-8",
491 ) as handle:
492 return bool(json.load(handle).get("session_token"))
493 except (OSError, ValueError):
494 return False
497def _detect_auth_state(username: str | None) -> str:
498 """Best-effort check of whether the sync loop can actually authenticate.
500 Returns one of:
501 - ``not_configured`` — no ``app.credentials.username`` in config.
502 - ``setup_needed`` — username set, but neither a cached keyring
503 password nor a saved session. The container's first 2FA flow
504 hasn't been completed.
505 - ``ready`` — username set, plus either a keyring entry or a saved
506 session. Sync loop can resume the session on the next retry.
507 - ``reauth_needed`` — a credential is present but the sync loop has
508 reported that it cannot authenticate with it.
510 Distinct from a *live* iCloud session check (which would require
511 hitting Apple). This is the cheap on-disk signal behind the pill the
512 dashboard shows before the sync loop has reported anything.
513 """
514 if not username:
515 return "not_configured"
516 try:
517 from icloudpy import utils as icloudpy_utils
519 if icloudpy_utils.password_exists_in_keyring(username):
520 # Ask the sync loop before claiming health: every on-disk
521 # signal still looks correct while an account is stuck on a
522 # second factor, so "keyring populated" alone would render a
523 # green dashboard over a sync that has not run for weeks.
524 if web_signals.get_auth_blocked().get("blocked"):
525 return "reauth_needed"
526 return "ready"
527 # An empty keyring is a supported configuration, not an unfinished
528 # setup: the sync loop runs off the saved session alone when no
529 # password is stored. Judge such a container on its session, or the
530 # dashboard would tell a deliberately password-free operator to go
531 # and finish a setup that is already finished.
532 if _saved_session_exists(username):
533 if web_signals.get_auth_blocked().get("blocked"):
534 return "reauth_needed"
535 return "ready"
536 except Exception as e:
537 LOGGER.debug(f"Web UI auth-state check raised: {e!s}")
538 return "setup_needed"
541def create_app(testing: bool = False) -> Flask:
542 """Construct the Flask app.
544 Splitting this out keeps ``tests/`` able to build the app under
545 ``TESTING=True`` without spawning a thread.
546 """
547 from werkzeug.middleware.proxy_fix import ProxyFix
549 # No ``static_folder``: templates are single-file with inline CSS and
550 # nothing ships under src/static, so wiring it would only add a 404
551 # route for /static/*.
552 template_dir = os.path.join(os.path.dirname(__file__), "templates")
553 app = Flask(__name__, template_folder=template_dir)
554 app.config["TESTING"] = testing
556 # Trust X-Forwarded-* from a single reverse-proxy hop (Cloudflare Tunnel,
557 # Authelia / Traefik). Lets ``url_for`` produce ``https://`` URLs and
558 # prevents Flask from mis-detecting the scheme when behind a TLS-
559 # terminating proxy. One hop is correct here — Cloudflare → backend.
560 app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1)
562 @app.after_request
563 def _no_cache(response):
564 """Defense against intermediaries (browser back/forward cache,
565 Cloudflare's auto-minify, mobile carrier proxies) serving stale
566 dashboard or auth payloads. The dashboard is always live data —
567 a cached snapshot would hide a missing mount marker or an
568 expired session."""
569 response.headers["Cache-Control"] = "private, no-store, no-cache, must-revalidate, max-age=0"
570 response.headers["Pragma"] = "no-cache"
571 response.headers["Expires"] = "0"
572 # CSRF defence: set the SameSite=Strict token cookie on every
573 # response so forms rendered server-side can read it (via the
574 # template) and same-site fetches automatically include it.
575 # ``secure=False`` because the default deployment is loopback
576 # over plain HTTP; users behind a TLS proxy benefit from the
577 # proxy's transport security, and SameSite=Strict is the load-
578 # bearing protection here regardless of TLS.
579 response.set_cookie(
580 _CSRF_COOKIE_NAME,
581 _CSRF_TOKEN,
582 samesite="Strict",
583 httponly=False,
584 secure=False,
585 path="/",
586 )
587 return response
589 @app.route("/")
590 def dashboard():
591 """Render the HTML dashboard — Apple-leaning design."""
592 config = _load_current_config()
593 status_payload = _build_status(config=config)
594 log_path = _logger_filename(config=config)
595 log_lines = _tail_log_file(path=log_path, lines=200)
596 return render_template(
597 "dashboard.html",
598 status=status_payload,
599 log_lines=log_lines,
600 log_path=log_path,
601 active_nav="dashboard",
602 version=os.environ.get("APP_VERSION", ""),
603 csrf_token=_get_csrf_token(),
604 )
606 @app.route("/api/health")
607 def health():
608 """Tiny endpoint for external monitors.
610 - 200 ``{"state": "ok"}`` when the config file is readable.
611 - 503 ``{"state": "config_missing"}`` when it isn't.
613 ``2fa_required`` is *not* a 503 — Apple sessions expire all the time
614 and the dashboard must stay reachable so the user can re-auth.
615 """
616 if not os.path.isfile(_current_config_path()):
617 return jsonify({"state": "config_missing"}), 503
618 return jsonify({"state": "ok"})
620 @app.route("/api/status")
621 def status():
622 """Live status payload for the dashboard + external consumers."""
623 config = _load_current_config()
624 payload = _build_status(config=config)
625 if not payload["config_loaded"]:
626 return jsonify(payload), 503
627 return jsonify(payload)
629 @app.route("/api/logs")
630 def logs():
631 """Last 200 lines of the configured log file. Best-effort: missing
632 or unreadable returns an empty list (never 500 — the dashboard
633 relies on this being reachable to render the rest of the page)."""
634 config = _load_current_config()
635 return jsonify(
636 {"lines": _tail_log_file(path=_logger_filename(config=config), lines=200)},
637 )
639 @app.route("/auth", methods=["GET"])
640 def auth_form():
641 """Auth form. Renders the password field by default; renders the
642 6-digit code field instead when ``_PENDING_AUTH`` indicates that
643 the password step already succeeded and 2FA is pending."""
644 return _render_auth(message=None, message_kind=None)
646 @app.route("/auth/password", methods=["POST"])
647 def auth_password():
648 """Step 1: store password in keyring, instantiate ICloudPyService,
649 trigger 2FA push if needed.
651 On success of either path: redirects — to /auth (now showing the
652 code form) if 2FA is pending, or back to / if the cached session
653 was still trusted.
655 Exceptions are caught and rendered as an error pill on /auth so
656 the user sees what Apple said.
657 """
658 rejection = _require_csrf()
659 if rejection is not None:
660 return rejection
662 password = request.form.get("password", "")
663 if not password:
664 return (
665 _render_auth(message="Password is required.", message_kind="err"),
666 400,
667 )
669 config = _load_current_config()
670 username = None
671 if config:
672 try:
673 username = config_parser.get_username(config=config)
674 except (KeyError, AttributeError, TypeError): # pragma: no cover
675 # Defensive: get_username walks app.credentials.username;
676 # partial configs (no credentials block) raise. Treat as
677 # missing. Rare in practice — coverage-pragma'd because
678 # mocking get_username globally breaks _render_auth.
679 username = None
680 if not username:
681 return (
682 _render_auth(
683 message="No app.credentials.username in config.yaml — set it and reload.",
684 message_kind="err",
685 ),
686 400,
687 )
689 try:
690 # Late import so /api/health still works if icloudpy is mid-upgrade.
691 import icloudpy
692 from icloudpy import utils as icloudpy_utils
694 api = icloudpy.ICloudPyService(
695 apple_id=username,
696 password=password,
697 cookie_directory=DEFAULT_COOKIE_DIRECTORY,
698 )
699 except Exception as e:
700 LOGGER.exception("Web UI auth failed during ICloudPyService instantiation")
701 return (
702 _render_auth(
703 message=f"Authentication failed: {e!s}",
704 message_kind="err",
705 ),
706 400,
707 )
709 if api.requires_2fa:
710 # PR 1 / fix/ios-26.4-auth dependency — best-effort. Catches all
711 # exceptions so a missing-method or push-trigger failure doesn't
712 # block the user from typing in a code they got via SMS.
713 try:
714 trigger = getattr(api, "trigger_2fa_push_notification", None)
715 if callable(trigger):
716 trigger()
717 except Exception as e:
718 LOGGER.warning(f"Web UI 2FA push trigger failed (non-fatal): {e!s}")
719 with _AUTH_LOCK:
720 _expire_stale_pending_auth_unlocked()
721 _PENDING_AUTH["api"] = api
722 _PENDING_AUTH["username"] = username
723 _PENDING_AUTH["password"] = password
724 _PENDING_AUTH["stashed_at"] = time.monotonic()
725 return redirect(url_for("auth_form"))
727 # No 2FA needed — cached session still trusted. Persist the
728 # password to the keyring so the sync loop can use it on the
729 # next retry, then bounce back to the dashboard.
730 try:
731 icloudpy_utils.store_password_in_keyring(
732 username=username,
733 password=password,
734 )
735 except Exception as e:
736 LOGGER.warning(f"Web UI keyring persist failed (non-fatal): {e!s}")
737 _wake_sync_loop()
738 return redirect(url_for("dashboard"))
740 @app.route("/auth/code", methods=["POST"])
741 def auth_code():
742 """Step 2: validate the 6-digit code on the in-flight session,
743 trust the browser, persist the password, clear pending, redirect.
745 - 400 if the code field is empty or no pending auth exists.
746 - 400 + 'Code rejected' if Apple says no — pending kept so the
747 user can retry without re-entering the password.
748 - On success: validate_2fa_code -> trust_session (failures here
749 are logged but non-fatal — the code already worked) ->
750 store_password_in_keyring -> clear pending -> redirect to /.
751 """
752 rejection = _require_csrf()
753 if rejection is not None:
754 return rejection
756 code = request.form.get("code", "").strip()
757 if not code:
758 return (
759 _render_auth(message="Enter the 6-digit code.", message_kind="err"),
760 400,
761 )
763 with _AUTH_LOCK:
764 _expire_stale_pending_auth_unlocked()
765 api = _PENDING_AUTH.get("api")
766 username = _PENDING_AUTH.get("username")
767 password = _PENDING_AUTH.get("password")
768 if api is None:
769 return (
770 _render_auth(
771 message="No pending auth — submit your password first.",
772 message_kind="err",
773 ),
774 400,
775 )
777 # All exit paths from here clear ``_PENDING_AUTH`` -- including
778 # failed validate_2fa_code, rejected codes, and trust_session
779 # failures. Without the ``finally`` the previous code only cleared
780 # on the success path, leaving the password sitting in process
781 # memory if Apple raised. On rejection the user retries via
782 # /auth/refresh-trust or by re-entering the password; we'd rather
783 # they take that path than leave a stale credential in memory.
784 try:
785 try:
786 accepted = api.validate_2fa_code(code)
787 except Exception as e:
788 LOGGER.exception("Web UI: validate_2fa_code raised")
789 return (
790 _render_auth(
791 message=f"2FA validation error: {e!s}",
792 message_kind="err",
793 ),
794 400,
795 )
797 if not accepted:
798 return (
799 _render_auth(
800 message="Code rejected by Apple. Try again — make sure you copy the latest code.",
801 message_kind="err",
802 ),
803 400,
804 )
806 # Code worked. Best-effort trust so the next session resume skips
807 # 2FA; if that fails (e.g. cookie store write error) just log it
808 # — the user's auth still succeeded for this session.
809 try:
810 api.trust_session()
811 except Exception as e:
812 LOGGER.warning(f"Web UI trust_session failed (non-fatal): {e!s}")
814 # Persist password to keyring so the sync-loop's next retry
815 # picks up the trusted session without prompting. A session-only
816 # re-auth has no password to persist, and writing one there would
817 # silently move the container out of password-free mode.
818 if password is not None:
819 try:
820 from icloudpy import utils as icloudpy_utils
822 icloudpy_utils.store_password_in_keyring(
823 username=username,
824 password=password,
825 )
826 except Exception as e:
827 LOGGER.warning(f"Web UI keyring persist failed (non-fatal): {e!s}")
829 _wake_sync_loop()
830 return redirect(url_for("dashboard"))
831 finally:
832 with _AUTH_LOCK:
833 _clear_pending_auth_unlocked()
835 @app.route("/auth/security-key", methods=["GET"])
836 def auth_security_key():
837 """Start a security-key (FIDO2/WebAuthn) re-auth ceremony.
839 Apple disables every other second factor once security keys are
840 enrolled, so the 6-digit paths (``/auth/code``, the Telegram
841 listener) can never complete for such an account -- Apple answers
842 every factor endpoint with an ``fsaChallenge`` instead of pushing
843 a code.
845 The assertion cannot be produced by this page: WebAuthn requires
846 ``rpId`` to be a registrable suffix of the page origin, and
847 Apple's ``rpId`` is ``apple.com``. Browsers additionally blocklist
848 FIDO devices from WebHID/WebUSB precisely to stop a page doing raw
849 CTAP against another origin. So the signing step runs natively on
850 whatever machine holds the key, and only the signed assertion
851 comes back here.
853 This handler authenticates with the keyring password, reads
854 Apple's challenge, stashes the live session under ``_PENDING_AUTH``
855 (same slot ``/auth/code`` uses) and renders the blob the operator
856 feeds to the signer.
857 """
858 # A plain page load must never contact Apple. Each sign-in attempt
859 # counts against a rate limit that answers 409 on /signin/init once
860 # tripped, and a reload -- or a browser prefetch -- would otherwise
861 # spend one. The challenge is only fetched when the operator asks
862 # for it, and a pending one is re-rendered rather than replaced so
863 # a command already copied stays valid.
864 with _AUTH_LOCK:
865 _expire_stale_pending_auth_unlocked()
866 pending_challenge = _PENDING_AUTH.get("fsa_challenge")
867 if pending_challenge:
868 LOGGER.info("Web UI security-key: re-rendering the pending challenge.")
869 return _render_auth(security_key_blob=_pack_challenge(pending_challenge))
870 return redirect(url_for("auth_form"))
872 @app.route("/auth/security-key/start", methods=["POST"])
873 def auth_security_key_start():
874 """Ask Apple for a challenge. This is the only path that signs in."""
875 rejection = _require_csrf()
876 if rejection is not None:
877 return rejection
879 # Pressing the button twice must not spend a second sign-in.
880 with _AUTH_LOCK:
881 _expire_stale_pending_auth_unlocked()
882 pending_challenge = _PENDING_AUTH.get("fsa_challenge")
883 if pending_challenge:
884 LOGGER.info("Web UI security-key: reusing the pending challenge.")
885 return _render_auth(security_key_blob=_pack_challenge(pending_challenge))
888 config = _load_current_config()
889 username = None
890 if config:
891 try:
892 username = config_parser.get_username(config=config)
893 except (KeyError, AttributeError, TypeError):
894 username = None
895 if not username:
896 return (
897 _render_auth(
898 message="No app.credentials.username in config.yaml — set it first.",
899 message_kind="err",
900 ),
901 400,
902 )
904 try:
905 from icloudpy import utils as icloudpy_utils
907 password = icloudpy_utils.get_password_from_keyring(username)
908 except Exception as e:
909 LOGGER.exception("Web UI security-key: keyring lookup raised")
910 return (
911 _render_auth(message=f"Keyring lookup failed: {e!s}", message_kind="err"),
912 500,
913 )
914 if not password:
915 return (
916 _render_auth(
917 message="No password in keyring — submit one below first.",
918 message_kind="warn",
919 ),
920 400,
921 )
923 try:
924 import icloudpy
926 ceremony_dir = _ceremony_cookie_dir()
927 api = icloudpy.ICloudPyService(
928 apple_id=username,
929 password=password,
930 cookie_directory=ceremony_dir,
931 )
932 except Exception as e:
933 _discard_ceremony_dir(locals().get("ceremony_dir"))
934 LOGGER.exception("Web UI security-key: ICloudPyService raised")
935 if _is_signin_throttled(e):
936 return (
937 _render_auth(
938 message=(
939 "Apple is rate-limiting sign-ins for this account "
940 "and will not issue a challenge yet. Nothing is "
941 "wrong with your key — wait a while before trying "
942 "again, as each attempt extends the limit."
943 ),
944 message_kind="warn",
945 ),
946 429,
947 )
948 return (
949 _render_auth(
950 message=f"Sign-in failed: {e!s}",
951 message_kind="err",
952 ),
953 400,
954 )
956 if not api.requires_2fa:
957 _wake_sync_loop()
958 return redirect(url_for("dashboard"))
960 fsa = api.security_key_challenge
961 if not fsa:
962 return (
963 _render_auth(
964 message=(
965 "Apple did not offer a security-key challenge. This "
966 "account may still use 6-digit codes — use the form below."
967 ),
968 message_kind="warn",
969 ),
970 400,
971 )
973 with _AUTH_LOCK:
974 _expire_stale_pending_auth_unlocked()
975 _PENDING_AUTH["api"] = api
976 _PENDING_AUTH["username"] = username
977 _PENDING_AUTH["password"] = password
978 _PENDING_AUTH["stashed_at"] = time.monotonic()
979 # Keep Apple's challenge string verbatim. It is echoed back in
980 # the assertion and Apple compares it byte-for-byte, so a
981 # re-encode differing only in base64 padding is rejected with a
982 # 409 -- the signer's encoding must never be trusted here.
983 stashed = dict(fsa)
984 _PENDING_AUTH["fsa_challenge"] = stashed
985 _PENDING_AUTH["cookie_dir"] = ceremony_dir
987 _record_auth_method(username, "security_key")
988 # Render exactly what was stashed. Handing over a different
989 # challenge than the one being waited on costs a physical touch to
990 # discover, so the two are deliberately the same object.
991 return _render_auth(security_key_blob=_pack_challenge(stashed))
993 @app.route("/auth/security-key", methods=["POST"])
994 def auth_security_key_submit():
995 """Finish the ceremony: submit the signed assertion to Apple.
997 The assertion is bound to the ``scnt``/``X-Apple-ID-Session-Id``
998 of the session stashed by the GET handler, so it must be
999 submitted with that same session -- which is why the signer
1000 never needs the password and never talks to Apple itself.
1001 """
1002 rejection = _require_csrf()
1003 if rejection is not None:
1004 return rejection
1006 raw = request.form.get("assertion", "").strip()
1007 if not raw:
1008 return (
1009 _render_auth(message="Paste the signer output.", message_kind="err"),
1010 400,
1011 )
1012 try:
1013 decoded = json.loads(base64.b64decode(raw))
1014 except Exception:
1015 return (
1016 _render_auth(
1017 message="That does not look like signer output — copy the whole line.",
1018 message_kind="err",
1019 ),
1020 400,
1021 )
1023 # The signer emits one assertion per candidate origin, because the
1024 # origin is fixed inside clientData at signing time and Apple rejects
1025 # a wrong one with the same opaque 409 it uses for everything else.
1026 signatures = decoded if isinstance(decoded, list) else [decoded]
1027 # Valid base64 JSON that is not signer output (``[]``, ``["x"]``)
1028 # must get the same answer as garbage, not a traceback.
1029 if not signatures or not all(isinstance(sig, dict) for sig in signatures):
1030 return (
1031 _render_auth(
1032 message="That does not look like signer output — copy the whole line.",
1033 message_kind="err",
1034 ),
1035 400,
1036 )
1038 with _AUTH_LOCK:
1039 _expire_stale_pending_auth_unlocked()
1040 api = _PENDING_AUTH.get("api")
1041 username = _PENDING_AUTH.get("username")
1042 password = _PENDING_AUTH.get("password")
1043 pending_fsa = _PENDING_AUTH.get("fsa_challenge") or {}
1044 ceremony_dir = _PENDING_AUTH.get("cookie_dir")
1045 issued_challenge = pending_fsa.get("challenge")
1046 if api is None:
1047 return (
1048 _render_auth(
1049 message=("Challenge expired — start a new security-key re-auth and sign the fresh challenge."),
1050 message_kind="err",
1051 ),
1052 400,
1053 )
1055 # Apple compares the echoed challenge byte-for-byte, and the signer
1056 # round-trips it through raw bytes, so it can come back re-encoded
1057 # (base64 padding Apple did not send). Compare on the decoded value
1058 # and submit the stashed original.
1059 #
1060 # A genuine mismatch means the signed challenge is not the one this
1061 # session is waiting on -- almost always a stale page, since every
1062 # page load starts a fresh Apple session with a fresh challenge.
1063 # Overwriting it here would hide that and produce an assertion whose
1064 # clientData disagrees with its challenge field, which Apple rejects
1065 # with an opaque 409. Say so instead.
1066 if issued_challenge:
1067 signed = str(signatures[0].get("challenge") or "")
1068 if _same_challenge(signed, issued_challenge):
1069 for signature in signatures:
1070 signature["challenge"] = issued_challenge
1071 else:
1072 LOGGER.warning(
1073 "Web UI security-key: signed challenge does not match the "
1074 "pending one -- stale page.",
1075 )
1076 with _AUTH_LOCK:
1077 _clear_pending_auth_unlocked()
1078 return (
1079 _render_auth(
1080 message=(
1081 "That signature is for an older challenge. Each time "
1082 "this page loads it starts a new one — reload, copy "
1083 "the command again, and re-sign."
1084 ),
1085 message_kind="err",
1086 ),
1087 400,
1088 )
1090 try:
1091 if not api.confirm_security_key(assertion=signatures[0]):
1092 LOGGER.warning("Web UI security-key: Apple refused the assertion.")
1093 return (
1094 _render_auth(
1095 message=(
1096 "Apple refused that signature. Reload to start a "
1097 "fresh challenge and sign again."
1098 ),
1099 message_kind="err",
1100 ),
1101 400,
1102 )
1104 try:
1105 from icloudpy import utils as icloudpy_utils
1107 icloudpy_utils.store_password_in_keyring(
1108 username=username,
1109 password=password,
1110 )
1111 except Exception as e:
1112 LOGGER.warning(f"Web UI security-key keyring persist failed: {e!s}")
1114 if ceremony_dir:
1115 _publish_ceremony_session(ceremony_dir)
1116 # Apple accepting the assertion is not proof the sync loop can
1117 # sign in, and claiming success without checking is how a broken
1118 # session got reported as working before.
1119 if not _session_authenticates(username):
1120 LOGGER.warning(
1121 "Web UI security-key: assertion accepted but sign-in still "
1122 "wants a factor.",
1123 )
1124 return (
1125 _render_auth(
1126 message=(
1127 "Apple accepted your key, but sign-in still asks for "
1128 "a second factor. Start over and sign a fresh "
1129 "challenge."
1130 ),
1131 message_kind="err",
1132 ),
1133 400,
1134 )
1135 LOGGER.info("Web UI: security-key re-auth succeeded; session trusted.")
1136 _wake_sync_loop()
1137 # The signer clears the clipboard itself once the signature is
1138 # read, so there is nothing left for the dashboard to do here.
1139 return redirect(url_for("dashboard"))
1140 except Exception as e:
1141 LOGGER.exception("Web UI security-key: assertion submit raised")
1142 return (
1143 _render_auth(message=f"Assertion submit failed: {e!s}", message_kind="err"),
1144 400,
1145 )
1146 finally:
1147 _discard_ceremony_dir(ceremony_dir)
1148 with _AUTH_LOCK:
1149 _clear_pending_auth_unlocked()
1151 @app.route("/auth/reset", methods=["POST"])
1152 def auth_reset():
1153 """Escape hatch — clear any in-flight pending-auth state.
1155 Useful when the user closed the tab mid-2FA and wants to start
1156 over without waiting for the in-memory state to expire.
1157 """
1158 rejection = _require_csrf()
1159 if rejection is not None:
1160 return rejection
1162 with _AUTH_LOCK:
1163 _clear_pending_auth_unlocked()
1164 return redirect(url_for("auth_form"))
1166 @app.route("/auth/refresh-trust", methods=["POST"])
1167 def auth_refresh_trust():
1168 """One-tap re-auth that reuses whatever credential the container has.
1170 When Apple's trusted-session lifetime is winding down (or has
1171 already expired since the last sync attempt), this lets the user
1172 kick off a fresh 2FA push without having to retype their
1173 password. Useful for "reset the clock" workflows where the
1174 password didn't change — only the trust window did.
1176 Flow:
1177 1. Look up keyring password by username from config.
1178 2. If absent → carry on without one. ``trust_session`` and the 2FA
1179 push both work from the saved session, so a password-free
1180 container can still use this button; only a session Apple has
1181 already stopped accepting needs a human with a password, and
1182 that case is reported as such.
1183 3. Spin up a transient client, fire the 2FA push if needed, stash
1184 the live session under the same _PENDING_AUTH dict /auth/code
1185 already consumes.
1186 4. Redirect to /auth — UI is now in "enter 6-digit code" mode.
1187 """
1188 rejection = _require_csrf()
1189 if rejection is not None:
1190 return rejection
1192 config = _load_current_config()
1193 username = None
1194 if config:
1195 try:
1196 username = config_parser.get_username(config=config)
1197 except (
1198 KeyError,
1199 AttributeError,
1200 TypeError,
1201 ): # pragma: no cover — defensive for hand-malformed configs
1202 username = None
1203 if not username:
1204 return (
1205 _render_auth(
1206 message="No app.credentials.username in config.yaml — set it first.",
1207 message_kind="err",
1208 ),
1209 400,
1210 )
1212 from icloudpy import exceptions as icloudpy_exceptions
1214 try:
1215 from icloudpy import utils as icloudpy_utils
1217 password = icloudpy_utils.get_password_from_keyring(username)
1218 except icloudpy_exceptions.ICloudPyNoStoredPasswordAvailableException:
1219 # Password-free container. Not an error, and not a reason to
1220 # refuse: the saved session is credential enough for both
1221 # trust_session and the 2FA push.
1222 password = None
1223 except Exception as e:
1224 LOGGER.exception("Web UI: keyring lookup raised")
1225 return (
1226 _render_auth(
1227 message=f"Keyring lookup failed: {e!s}",
1228 message_kind="err",
1229 ),
1230 500,
1231 )
1233 try:
1234 if password is None:
1235 # Session-only: the client refuses to sign in with
1236 # credentials, so a dead session surfaces as the
1237 # no-password exception below rather than a sign-in attempt.
1238 from src import sync
1240 api = sync.get_api_instance(username=username, password=None)
1241 else:
1242 import icloudpy
1244 api = icloudpy.ICloudPyService(
1245 apple_id=username,
1246 password=password,
1247 cookie_directory=DEFAULT_COOKIE_DIRECTORY,
1248 )
1249 except icloudpy_exceptions.ICloudPyNoStoredPasswordAvailableException:
1250 LOGGER.info("Web UI refresh-trust: no password and the saved session is not usable")
1251 return (
1252 _render_auth(
1253 message=(
1254 "The saved session is no longer valid and no password is stored. "
1255 "Enter your Apple ID password below, or run the documented "
1256 "`icloud --username=… --session-directory=…` command to sign in "
1257 "without storing one."
1258 ),
1259 message_kind="err",
1260 ),
1261 400,
1262 )
1263 except Exception as e:
1264 LOGGER.exception("Web UI refresh-trust: ICloudPyService raised")
1265 return (
1266 _render_auth(
1267 message=(
1268 f"Refresh trust failed: {e!s}. Your stored password may be stale — submit a new one below."
1269 ),
1270 message_kind="err",
1271 ),
1272 400,
1273 )
1275 if not api.requires_2fa:
1276 # Trust window was still alive — nothing to do, sync loop is
1277 # already authenticated. Bounce back to the dashboard with
1278 # the success state.
1279 _wake_sync_loop()
1280 return redirect(url_for("dashboard"))
1282 try:
1283 trigger = getattr(api, "trigger_2fa_push_notification", None)
1284 if callable(trigger):
1285 trigger()
1286 except Exception as e:
1287 LOGGER.warning(f"Web UI refresh-trust 2FA push failed: {e!s}")
1289 with _AUTH_LOCK:
1290 _expire_stale_pending_auth_unlocked()
1291 _PENDING_AUTH["api"] = api
1292 _PENDING_AUTH["username"] = username
1293 _PENDING_AUTH["password"] = password
1294 _PENDING_AUTH["stashed_at"] = time.monotonic()
1295 return redirect(url_for("auth_form"))
1297 @app.route("/api/sync", methods=["POST"])
1298 def api_sync():
1299 """Queue an immediate sync run for one or both services.
1301 ``service=drive`` / ``service=photos`` / ``service=all``. The
1302 web thread can't run sync.sync() directly — it would race with
1303 the existing loop. Instead this touches a sentinel file in
1304 ICLOUD_DOCKER_CONFIG_DIR; ``src.sync`` checks for it at the top
1305 of each loop iteration and resets the countdown when present.
1307 Idempotent: tapping repeatedly while a request is still queued
1308 is a no-op (the sentinel just gets re-touched).
1309 """
1310 rejection = _require_csrf()
1311 if rejection is not None:
1312 return rejection
1314 service = (request.form.get("service") or request.args.get("service") or "").strip().lower()
1315 if service == "all":
1316 wanted = ("drive", "photos")
1317 elif service in ("drive", "photos"):
1318 wanted = (service,)
1319 else:
1320 return (
1321 jsonify({"error": "service must be one of: drive, photos, all"}),
1322 400,
1323 )
1325 # Honour the user's config — only queue services that are
1326 # actually configured. Avoids touching a photos sentinel on a
1327 # drive-only install.
1328 config = _load_current_config()
1329 configured = {svc for svc in ("drive", "photos") if config and svc in config}
1330 if not configured:
1331 return jsonify({"error": "no services configured"}), 400
1333 queued = [svc for svc in wanted if svc in configured and web_signals.request_force_sync(svc)]
1335 # Browser form submit gets a redirect; API consumers (curl,
1336 # monitors) get JSON. Distinguished by Accept header.
1337 if request.headers.get("Accept", "").startswith("application/json"):
1338 return jsonify({"queued": queued})
1339 return redirect(url_for("dashboard"))
1341 return app
1344_SIGNER_PATH = os.path.join(os.path.dirname(__file__), "icloud_sign.py")
1347def _lookup_auth_method(status_payload: dict[str, Any]) -> str | None:
1348 """Recorded second factor for the configured account, if we've seen one.
1350 Drives which route the auth page leads with: an account Apple has
1351 answered with an ``fsaChallenge`` can never complete the 6-digit
1352 form, so offering it first just wastes the operator's time.
1353 """
1354 username = status_payload.get("username") if status_payload else None
1355 if not username:
1356 return None
1357 try:
1358 from src import web_signals
1360 return web_signals.get_auth_method(username)
1361 except Exception: # pragma: no cover - never break rendering over state
1362 return None
1365def _record_auth_method(username: str, method: str) -> None:
1366 """Persist the account's second factor; never fatal to the auth flow."""
1367 try:
1368 from src import web_signals
1370 web_signals.record_auth_method(username=username, method=method)
1371 except Exception as e: # pragma: no cover - bookkeeping must not break auth
1372 LOGGER.warning(f"Web UI: recording auth method failed: {e!s}")
1375def _pack_challenge(fsa: dict[str, Any]) -> str:
1376 """Pack the challenge + credential handles into one short token.
1378 Length matters: the operator copies this inside a single shell
1379 command. Packing raw bytes (2-byte length prefix each) rather than
1380 base64-of-JSON roughly halves it, and ``rpId`` is dropped entirely
1381 because Apple's is always ``apple.com``.
1382 """
1384 def decode(value: str) -> bytes:
1385 return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
1387 parts = [decode(fsa["challenge"])]
1388 parts.extend(decode(handle) for handle in fsa["keyHandles"])
1389 packed = b"".join(struct.pack("!H", len(part)) + part for part in parts)
1390 return base64.urlsafe_b64encode(packed).decode("ascii").rstrip("=")
1393def _build_signer_command(blob: str) -> str:
1394 """The self-contained form: the signer source travels inside the command.
1396 Piped to ``uv run`` on stdin, so nothing is left on the operator's disk
1397 and the machine holding the key needs no network route back to this
1398 container. Always offered, because the short form below cannot reach a
1399 dashboard that sits behind proxy authentication.
1400 """
1401 try:
1402 with open(_SIGNER_PATH, encoding="utf-8") as handle:
1403 source = handle.read()
1404 except OSError:
1405 return "src/icloud_sign.py is missing from this image."
1406 return f"uv run --quiet - {blob} <<'ICLOUDSIGN'\n{source}ICLOUDSIGN"
1409_SIGNER_REPO_RAW = "https://raw.githubusercontent.com/mandarons/icloud-docker"
1412def _signer_source_ref() -> str | None:
1413 """The release tag this image was built from, e.g. ``v1.29.0``.
1415 Release images are built from ``v*`` tags and carry the version in
1416 ``APP_VERSION``. Development and PR builds (``dev``, ``pr-123``) have no
1417 tag to point at, so they get None.
1418 """
1419 version = os.environ.get("APP_VERSION", "").strip()
1420 if not re.fullmatch(r"\d+\.\d+\.\d+", version):
1421 return None
1422 return f"v{version}"
1425def _build_short_signer_command(blob: str) -> str | None:
1426 """``uv run <signer at this release's tag> <blob>`` -- one line.
1428 Fetched from the public repo at the tag this image was built from:
1429 reachable from any machine (a dashboard behind an auth proxy, as one
1430 accepting an Apple ID password should be, cannot be fetched by uv), and
1431 readable before running. The signer declares its fido2 dependency inline
1432 (PEP 723), so no ``--with`` is needed.
1434 None for a build that is not a release; the self-contained form is then
1435 the only one offered.
1436 """
1437 ref = _signer_source_ref()
1438 if ref is None:
1439 return None
1440 return f"uv run --quiet {_SIGNER_REPO_RAW}/{ref}/src/icloud_sign.py {blob}"
1443def _session_authenticates(username: str) -> bool:
1444 """Confirm the sync loop can now sign in.
1446 Apple accepting the assertion is not the same as the loop being able
1447 to authenticate, and reporting success on the submit alone once
1448 announced a session that still wanted a second factor.
1449 """
1450 try:
1451 import icloudpy
1452 from icloudpy import utils as icloudpy_utils
1454 api = icloudpy.ICloudPyService(
1455 apple_id=username,
1456 password=icloudpy_utils.get_password_from_keyring(username),
1457 cookie_directory=DEFAULT_COOKIE_DIRECTORY,
1458 )
1459 except Exception as e:
1460 LOGGER.warning(f"Web UI: session failed to authenticate: {e!s}")
1461 return False
1462 LOGGER.info(f"Web UI: session check -> requires_2fa={api.requires_2fa}")
1463 return not api.requires_2fa
1466def _same_challenge(left: str, right: str) -> bool:
1467 """Compare two base64 challenge strings by value, not spelling.
1469 Apple emits the standard alphabet unpadded; a round-trip through raw
1470 bytes can come back padded, or urlsafe. Those are the same challenge.
1471 """
1473 def decode(value: str) -> bytes | None:
1474 try:
1475 return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
1476 except (ValueError, TypeError):
1477 return None
1479 decoded = decode(left)
1480 return decoded is not None and decoded == decode(right)
1483def _is_signin_throttled(error: Exception) -> bool:
1484 """True when Apple refused to start a sign-in rather than to finish one.
1486 Apple answers a rate-limited account with 409 on ``/signin/init``, the
1487 very first SRP call, so no challenge can be issued at all. Worth
1488 distinguishing: it is not a bad password, a bad key, or a bad
1489 assertion, and retrying is what prolongs it.
1490 """
1491 text = str(error)
1492 return "signin/init" in text and "409" in text
1495def _ceremony_cookie_dir() -> str:
1496 """A private, empty cookie jar for one security-key ceremony.
1498 icloudpy loads its jar with ``ignore_expires=True``, so cookies from a
1499 dead session are replayed on every new sign-in -- an expired
1500 ``X-APPLE-WEBAUTH-TOKEN`` and friends arriving alongside a fresh SRP
1501 handshake. Apple's reply to that contradiction is 409 Conflict, which
1502 is exactly what the assertion submit kept receiving.
1504 The ceremony therefore runs against its own jar and publishes the
1505 result only once Apple has accepted it.
1506 """
1507 return tempfile.mkdtemp(prefix="icloud-securitykey-")
1510def _publish_ceremony_session(cookie_dir: str) -> None:
1511 """Move a ceremony's session files into the directory sync reads.
1513 Called only after Apple accepts the assertion and the session is
1514 trusted, so a failed attempt can never damage a working session. The
1515 previous files are kept alongside with a ``.bak`` suffix.
1516 """
1517 try:
1518 os.makedirs(DEFAULT_COOKIE_DIRECTORY, exist_ok=True)
1519 for name in os.listdir(cookie_dir):
1520 source = os.path.join(cookie_dir, name)
1521 target = os.path.join(DEFAULT_COOKIE_DIRECTORY, name)
1522 if os.path.exists(target):
1523 shutil.copy2(target, f"{target}.bak")
1524 shutil.copy2(source, target)
1525 LOGGER.info(f"Web UI: published ceremony session to {DEFAULT_COOKIE_DIRECTORY}.")
1526 except OSError:
1527 LOGGER.exception("Web UI: could not publish ceremony session")
1530def _discard_ceremony_dir(cookie_dir: str | None) -> None:
1531 """Remove a ceremony's private jar. Best-effort."""
1532 if cookie_dir:
1533 shutil.rmtree(cookie_dir, ignore_errors=True)
1536def _render_auth(
1537 message: str | None = None,
1538 message_kind: str | None = None,
1539 security_key_blob: str | None = None,
1540):
1541 """Render auth.html with the current pending state and an optional
1542 error/info pill. Factored out so the POST endpoints can reuse it.
1544 ``security_key_blob`` carries the base64 challenge bundle for the
1545 FIDO2 ceremony; when set the template shows the signer instructions
1546 instead of the 6-digit code form."""
1547 from flask import render_template as _render
1549 config = _load_current_config()
1550 status_payload = _build_status(config=config)
1551 with _AUTH_LOCK:
1552 pending = bool(_PENDING_AUTH)
1553 return _render(
1554 "auth.html",
1555 status=status_payload,
1556 pending=pending,
1557 message=message,
1558 message_kind=message_kind,
1559 active_nav="auth",
1560 version=os.environ.get("APP_VERSION", ""),
1561 csrf_token=_get_csrf_token(),
1562 security_key_blob=security_key_blob,
1563 signer_command=_build_signer_command(security_key_blob) if security_key_blob else None,
1564 signer_source_ref=_signer_source_ref(),
1565 signer_command_short=(
1566 _build_short_signer_command(security_key_blob)
1567 if security_key_blob
1568 else None
1569 ),
1570 auth_method=_lookup_auth_method(status_payload),
1571 )
1574def start_in_thread(
1575 host: str = "127.0.0.1",
1576 port: int = 8080,
1577) -> threading.Thread:
1578 """Launch the Flask app on a daemon thread.
1580 The main sync loop owns the process; the web thread dies when the
1581 parent process exits. Returns the thread, or ``None`` if the port
1582 could not be bound (the sync loop continues regardless).
1584 Uses ``werkzeug.serving.make_server`` rather than ``Flask.run()``:
1585 binding happens synchronously here so a port conflict is reported
1586 as an error instead of a false "listening" line, and it avoids the
1587 dev-server banner. Deliberately dependency-free (no gunicorn) --
1588 this is a single-user, behind-a-proxy operator console (default
1589 host 127.0.0.1), not a public-facing API.
1590 """
1591 app = create_app()
1593 # Bind in the main thread so the "listening" log is truthful: a port
1594 # conflict raises here and is reported as a failure, instead of the
1595 # old optimistic log racing a background bind error.
1596 try:
1597 server = make_server(host, port, app, threaded=True)
1598 except OSError as e:
1599 LOGGER.error(f"Web UI failed to bind {host}:{port} — {e!s}")
1600 return None
1602 def _serve_bound():
1603 try:
1604 server.serve_forever()
1605 except Exception as e: # noqa: BLE001 -- daemon thread, never crash the sync loop
1606 LOGGER.error(f"Web UI server stopped: {e!s}")
1608 thread = threading.Thread(target=_serve_bound, name="icloud-web-ui", daemon=True)
1609 thread.start()
1610 LOGGER.info(f"Web UI listening on http://{host}:{port}/ (host={host}, port={port})")
1611 return thread