diff --git a/ipatool_api/services/appstore.py b/ipatool_api/services/appstore.py index e602fa2..d06f17d 100644 --- a/ipatool_api/services/appstore.py +++ b/ipatool_api/services/appstore.py @@ -4,11 +4,15 @@ import json import os import plistlib +import html +import re +from email.utils import parsedate_to_datetime from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Tuple from zipfile import ZIP_DEFLATED, ZipFile, ZipInfo +from urllib.parse import urlparse import requests @@ -29,7 +33,9 @@ HTTPRequest, HTTPResult, Payload, + ResponseDecodeError, XMLPayload, + extract_urls, ) from .keychain import FileKeychain from .machine import Machine @@ -53,6 +59,45 @@ class AppStoreConfig: verify: bool | str = True +def normalize_auth_endpoint(*endpoints: str) -> str: + for endpoint in endpoints: + endpoint = (endpoint or "").strip() + if not endpoint: + continue + + normalized = normalize_native_auth_endpoint(endpoint) + return normalized or endpoint + + return constants.DEFAULT_NATIVE_AUTH_ENDPOINT + + +def normalize_native_auth_endpoint(endpoint: str) -> str: + parsed = urlparse(endpoint) + if parsed.hostname != constants.PRIVATE_AUTH_DOMAIN: + return "" + + path = parsed.path.rstrip("/") + if not path.endswith("/fast"): + path = f"{path.rstrip('/')}/fast" + return parsed._replace(path=f"{path}/").geturl() + + +def auth_endpoint_from_response_error(exc: ResponseDecodeError) -> str: + parts = list(exc.urls) + if exc.body: + parts.append(exc.body) + return auth_endpoint_from_text(" ".join(parts)) + + +def auth_endpoint_from_text(text: str) -> str: + text = html.unescape(text.replace("\\/", "/")) + for url in extract_urls(text.encode("utf-8", errors="replace")): + endpoint = normalize_native_auth_endpoint(url.rstrip(".,;)")) + if endpoint: + return endpoint + return "" + + class AppStoreService: """Port of the Go ``appstore.AppStore`` implementation.""" @@ -60,6 +105,8 @@ def __init__(self, config: AppStoreConfig) -> None: self._keychain = config.keychain self._machine = config.machine self._http = HTTPClient(config.cookie_store, verify=config.verify) + self._verify = config.verify + self._version_history_cache: Dict[Tuple[int, str], Dict[str, str]] = {} self._storage_dir = Path(config.machine.home_directory()) / ".ipatool" self._storage_dir.mkdir(parents=True, exist_ok=True) @@ -75,7 +122,7 @@ def login(self, email: str, password: str, auth_code: str | None = None) -> Acco retry = False bag_result = self.bag() - auth_endpoint = bag_result.auth_endpoint + auth_endpoint = normalize_auth_endpoint(bag_result.auth_endpoint) for attempt in range(1, 5): payload: Payload = XMLPayload( @@ -98,7 +145,21 @@ def login(self, email: str, password: str, auth_code: str | None = None) -> Acco follow_redirects=False, ) print(f"Attempt {attempt}: POST {url}") - result = self._send_request(request) + try: + result = self._send_request(request, allow_decode_error=True) + except ResponseDecodeError as exc: + discovered_endpoint = auth_endpoint_from_response_error(exc) + if discovered_endpoint and discovered_endpoint != auth_endpoint: + auth_endpoint = discovered_endpoint + redirect_url = None + continue + raise AppStoreError("unexpected response from Apple", metadata={ + "status": exc.status_code, + "headers": exc.headers, + "contentType": exc.content_type, + "body": exc.body, + "urls": exc.urls, + }) from exc print(f"Status: {result.status_code}") print(f"Headers: {result.headers}") print(f"Data keys: {list(result.data.keys()) if isinstance(result.data, dict) else type(result.data)}") @@ -196,7 +257,10 @@ def bag(self) -> BagOutput: ) result = self._send_request(request) url_bag = result.data.get("urlBag", {}) - auth_endpoint = url_bag.get("authenticateAccount", "") + auth_endpoint = normalize_auth_endpoint( + result.data.get("authenticateAccount", ""), + url_bag.get("authenticateAccount", ""), + ) return BagOutput(auth_endpoint=auth_endpoint) # ------------------------------------------------------------------ @@ -388,19 +452,21 @@ def get_version_metadata(self, account: Account, app: App, version_id: str) -> G metadata = item.get("metadata", {}) asset_info = item.get("asset-info", {}) - release_date = metadata.get("releaseDate") - try: - release = datetime.fromisoformat(str(release_date).replace("Z", "+00:00")) - except ValueError as exc: - raise AppStoreError("failed to parse release date", metadata=metadata) from exc + display_version = str(metadata.get("bundleShortVersionString", "N/A")) + release = self._release_date_from_metadata(metadata, result.headers) + history_date = self._app_store_history_dates(account, app).get(display_version) + if history_date: + release = self._parse_apple_date(history_date) + release_display = self._release_date_display(account, app, metadata, release) age_ratings = metadata.get("appAgeRatings", {}) us_rating = age_ratings.get("US", {}) return GetVersionMetadataOutput( - display_version=str(metadata.get("bundleShortVersionString", "N/A")), + display_version=display_version, build_number=str(metadata.get("bundleVersion", "N/A")), release_date=release, + release_date_display=release_display, file_size=asset_info.get("file-size", 0), bundle_id=str(metadata.get("softwareVersionBundleId", "N/A")), artist_name=str(metadata.get("artistName", "N/A")), @@ -443,6 +509,12 @@ def _parse_login_response( if not failure_type and customer_message == constants.CUSTOMER_MESSAGE_ACCOUNT_DISABLED: raise AppStoreError("account is disabled", metadata=data) + if not failure_type and customer_message == constants.CUSTOMER_MESSAGE_ACTION_SIGN_IN_PAGE: + raise AppStoreError( + "account requires browser sign-in (2FA or Apple ID review required)", + metadata=data, + ) + if failure_type and customer_message: raise AppStoreError(customer_message, metadata=data) @@ -467,7 +539,110 @@ def _build_query_url(self, path: str, params: Dict[str, str]) -> str: query = urlencode(params) return f"https://{constants.ITUNES_API_DOMAIN}{path}?{query}" - def _send_request(self, request: HTTPRequest) -> HTTPResult: + def _release_date_from_metadata( + self, + metadata: Dict[str, Any], + headers: Dict[str, str], + ) -> Optional[datetime]: + generated_dates = self._response_generated_dates(headers) + for key in ( + "softwareVersionReleaseDate", + "versionReleaseDate", + "currentVersionReleaseDate", + ): + raw_value = metadata.get(key) + parsed = self._parse_apple_date(raw_value) + if not parsed: + continue + if any(parsed == generated or parsed.date() == generated.date() for generated in generated_dates): + continue + return parsed + return None + + def _release_date_display( + self, + account: Account, + app: App, + metadata: Dict[str, Any], + release: Optional[datetime], + ) -> str: + display_version = str(metadata.get("bundleShortVersionString", "")).strip() + history_date = self._app_store_history_dates(account, app).get(display_version) + if history_date: + return history_date + if release: + return release.date().isoformat() + + original_release = self._parse_apple_date(metadata.get("releaseDate")) + if original_release: + return f"Unknown ({original_release.date().isoformat()})" + return "Unknown" + + def _app_store_history_dates(self, account: Account, app: App) -> Dict[str, str]: + country = self._country_code_from_storefront(account.store_front).lower() + cache_key = (app.id, country) + cached = self._version_history_cache.get(cache_key) + if cached is not None: + return cached + + url = f"https://apps.apple.com/{country}/app/id{app.id}" + try: + response = requests.get( + url, + headers={"User-Agent": constants.DEFAULT_USER_AGENT}, + timeout=10, + verify=self._verify, + ) + response.raise_for_status() + except requests.RequestException as exc: + print(f"Failed to fetch App Store version history from {url}: {exc}") + self._version_history_cache[cache_key] = {} + return {} + + history = self._parse_app_store_history_dates(response.text) + self._version_history_cache[cache_key] = history + return history + + def _parse_app_store_history_dates(self, page_html: str) -> Dict[str, str]: + history: Dict[str, str] = {} + pattern = re.compile( + r']*class="[^"]*\bmetadata\b[^"]*"[^>]*>\s*' + r']*>\s*([^<]+?)\s*\s*' + r']*datetime="(\d{4}-\d{2}-\d{2})"[^>]*>', + re.IGNORECASE | re.DOTALL, + ) + for version, release_date in pattern.findall(page_html): + version = html.unescape(version).strip() + if version and version not in history: + history[version] = release_date + return history + + def _response_generated_dates(self, headers: Dict[str, str]) -> List[datetime]: + dates: List[datetime] = [] + for key in ("x-apple-date-generated", "Date", "Expires"): + parsed = self._parse_apple_date(headers.get(key)) + if parsed: + dates.append(parsed) + return dates + + def _parse_apple_date(self, value: Any) -> Optional[datetime]: + if not value: + return None + if isinstance(value, datetime): + return value + text = str(value).strip() + if not text: + return None + try: + return datetime.fromisoformat(text.replace("Z", "+00:00")) + except ValueError: + pass + try: + return parsedate_to_datetime(text) + except (TypeError, ValueError): + return None + + def _send_request(self, request: HTTPRequest, allow_decode_error: bool = False) -> HTTPResult: try: print(f"Sending {request.method} request to {request.url}") response = self._http.send(request) @@ -488,6 +663,17 @@ def _send_request(self, request: HTTPRequest) -> HTTPResult: "body": body_text, } raise AppStoreError("unexpected response from Apple", metadata=metadata) from exc + except ResponseDecodeError as exc: + if allow_decode_error: + raise + metadata = { + "status": exc.status_code, + "headers": exc.headers, + "contentType": exc.content_type, + "body": exc.body, + "urls": exc.urls, + } + raise AppStoreError("unexpected response from Apple", metadata=metadata) from exc except requests.RequestException as exc: raise AppStoreError("network request failed", metadata={"error": str(exc)}) from exc @@ -663,4 +849,3 @@ def _replicate_sinf_from_info( info = ZipInfo(full_path) info.compress_type = ZIP_DEFLATED zip_file.writestr(info, sinfs[0].data) - diff --git a/ipatool_api/services/constants.py b/ipatool_api/services/constants.py index 463408d..5ab3662 100644 --- a/ipatool_api/services/constants.py +++ b/ipatool_api/services/constants.py @@ -15,6 +15,7 @@ CUSTOMER_MESSAGE_BAD_LOGIN = "MZFinance.BadLogin.Configurator_message" CUSTOMER_MESSAGE_ACCOUNT_DISABLED = "Your account is disabled." CUSTOMER_MESSAGE_SUBSCRIPTION_REQUIRED = "Subscription Required" +CUSTOMER_MESSAGE_ACTION_SIGN_IN_PAGE = "AMD-Action::SP" ITUNES_API_DOMAIN = "itunes.apple.com" ITUNES_API_SEARCH_PATH = "/search" @@ -27,6 +28,10 @@ PRIVATE_PURCHASE_PATH = "/WebObjects/MZFinance.woa/wa/buyProduct" PRIVATE_DOWNLOAD_PATH = "/WebObjects/MZFinance.woa/wa/volumeStoreDownloadProduct" +PRIVATE_AUTH_DOMAIN = f"auth.{ITUNES_API_DOMAIN}" +PRIVATE_AUTH_NATIVE_PATH = "/auth/v1/native/fast/" +DEFAULT_NATIVE_AUTH_ENDPOINT = f"https://{PRIVATE_AUTH_DOMAIN}{PRIVATE_AUTH_NATIVE_PATH}" + HTTP_HEADER_STOREFRONT = "X-Set-Apple-Store-Front" HTTP_HEADER_POD = "pod" diff --git a/ipatool_api/services/http_client.py b/ipatool_api/services/http_client.py index 9e533e8..aaef301 100644 --- a/ipatool_api/services/http_client.py +++ b/ipatool_api/services/http_client.py @@ -12,6 +12,18 @@ _DOCUMENT_XML_RE = re.compile(r"(?is)]*>(.*)") _PLIST_XML_RE = re.compile(r"(?is)]*>.*?") _DICT_XML_RE = re.compile(r"(?is)]*>.*") +_URL_RE = re.compile(rb"""https?://[^\s"'<>]+""") + + +def extract_urls(body: bytes) -> list[str]: + return [match.decode("utf-8", errors="replace") for match in _URL_RE.findall(body)] + + +def _truncate_body(body: bytes, max_chars: int = 500) -> str: + text = body.decode("utf-8", errors="replace").strip() + if len(text) <= max_chars: + return text + return text[:max_chars] + "..." def _normalize_plist_body(body: bytes) -> bytes: @@ -75,6 +87,28 @@ def __init__(self, message: str, *, status_code: int, headers: Dict[str, str], b self.body = body +class ResponseDecodeError(RuntimeError): + """Raised when Apple's XML/plist response cannot be decoded.""" + + def __init__( + self, + message: str, + *, + cause: Exception, + status_code: int, + headers: Dict[str, str], + body: bytes, + ) -> None: + super().__init__(message) + self.__cause__ = cause + self.cause = cause + self.status_code = status_code + self.headers = headers + self.content_type = headers.get("Content-Type", "") + self.body = _truncate_body(body) + self.urls = extract_urls(body) + + @dataclass(slots=True) class HTTPRequest: method: str @@ -179,8 +213,9 @@ def send(self, request: HTTPRequest) -> HTTPResult: body=raw_body, ) except (ValueError, plistlib.InvalidFileException) as exc: - raise HTTPClientResponseError( - "Failed to parse server response", + raise ResponseDecodeError( + f"failed to unmarshal xml: {exc}", + cause=exc, status_code=response.status_code, headers=raw_headers, body=raw_body, diff --git a/ipatool_api/services/models.py b/ipatool_api/services/models.py index f435484..edc7e7b 100644 --- a/ipatool_api/services/models.py +++ b/ipatool_api/services/models.py @@ -105,7 +105,8 @@ class ListVersionsOutput: class GetVersionMetadataOutput: display_version: str build_number: str - release_date: datetime + release_date: Optional[datetime] + release_date_display: str file_size: int bundle_id: str artist_name: str @@ -120,7 +121,8 @@ def to_dict(self) -> Dict[str, Any]: return { "displayVersion": self.display_version, "buildNumber": self.build_number, - "releaseDate": self.release_date.isoformat(), + "releaseDate": self.release_date.isoformat() if self.release_date else None, + "releaseDateDisplay": self.release_date_display, "fileSize": self.file_size, "bundleId": self.bundle_id, "artistName": self.artist_name, diff --git a/static/js/app.js b/static/js/app.js index 8527a86..c3f2961 100644 --- a/static/js/app.js +++ b/static/js/app.js @@ -520,7 +520,9 @@ class IPAToolUI { renderVersionCards(versions) { this.versionsList.innerHTML = versions.map((version, index) => { const fileSizeMB = (version.fileSize / (1024 * 1024)).toFixed(2); - const releaseDate = new Date(version.releaseDate).toLocaleDateString(); + const releaseDate = version.releaseDateDisplay || (version.releaseDate + ? new Date(version.releaseDate).toLocaleDateString() + : 'Unknown'); return `