# Copyright 2023 The Chromium Authors # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. from __future__ import annotations import contextlib import json import logging import tempfile import zipfile from typing import (TYPE_CHECKING, Dict, Final, Iterable, List, Optional, Tuple, Type, Union, cast) from crossbench import helper from crossbench import path as pth from crossbench.browsers.chrome.version import ChromeVersion from crossbench.browsers.downloader import (DMGArchiveHelper, Downloader, IncompatibleVersionError, RPMArchiveHelper) from crossbench.browsers.version import BrowserVersion, BrowserVersionChannel from crossbench.plt.android_adb import AndroidAdbPlatform from crossbench.plt.base import SubprocessError if TYPE_CHECKING: from crossbench.plt.android_adb import Adb from crossbench.plt.base import Platform class ChromeDownloader(Downloader): STORAGE_URL: str = "gs://chrome-signed/desktop-5c0tCh/" VERSION_URL = ( "https://versionhistory.googleapis.com/v1/" "chrome/platforms/{platform}/channels/{channel}/versions?filter={filter}") VERSION_URL_PLATFORM_LOOKUP: Dict[Tuple[str, str], str] = { ("win", "ia32"): "win", ("win", "x64"): "win64", ("linux", "x64"): "linux", ("macos", "x64"): "mac", ("macos", "arm64"): "mac_arm64", ("android", "arm64"): "android", } def __init__(self, *args, **kwargs): self._gsutil: Optional[pth.AnyPath] = None super().__init__(*args, **kwargs) @classmethod def is_valid_version(cls, path_or_identifier: str): return ChromeVersion.is_valid_unique(path_or_identifier) @classmethod def _is_valid(cls, path_or_identifier: pth.AnyPathLike, browser_platform: Platform) -> bool: if cls.is_valid_version(str(path_or_identifier)): return True path = browser_platform.path(path_or_identifier) return (browser_platform.exists(path) and path.name.endswith(cls.ARCHIVE_SUFFIX)) @classmethod def _get_loader_cls(cls, browser_platform: Platform) -> Type[ChromeDownloader]: if browser_platform.is_macos: return ChromeDownloaderMacOS if browser_platform.is_linux: return ChromeDownloaderLinux if browser_platform.is_win: return ChromeDownloaderWin if browser_platform.is_android: return ChromeDownloaderAndroid raise ValueError( "Downloading chrome is only supported on linux and macOS, " f"but not on {browser_platform.name} {browser_platform.machine}") def _pre_check(self) -> None: super()._pre_check() if not self._requested_version: return self._gsutil = self.host_platform.which("gsutil") if not self._gsutil: raise ValueError( f"Cannot download chrome version {self._requested_version}: " "please install gsutil.\n" "- https://cloud.google.com/storage/docs/gsutil_install\n" "- Run 'gcloud auth login' to get access to the archives " "(googlers only).") @property def gsutil(self) -> pth.AnyPath: assert self._gsutil, "gsutil not be found." return self._gsutil def _requested_version_validation(self) -> None: pass def _parse_version(self, version_identifier: str) -> BrowserVersion: return ChromeVersion.parse_unique(version_identifier) def _find_archive_url(self) -> Tuple[BrowserVersion, Optional[str]]: # Quick probe for complete versions if self._requested_version.is_complete: return self._find_exact_archive_url() return self._find_milestone_archive_url() def _find_milestone_archive_url(self) -> Tuple[BrowserVersion, Optional[str]]: milestone: int = self._requested_version.major platform = self.VERSION_URL_PLATFORM_LOOKUP.get(self._browser_platform.key) if not platform: raise ValueError(f"Unsupported platform {self._browser_platform}") # Version ordering is: stable < beta < dev < canary < canary_asan # See https://developer.chrome.com/docs/web-platform/versionhistory/reference#filter channel_filter = "channel<=canary" requested_channel = BrowserVersionChannel.ANY if self._requested_version.has_channel: requested_channel = self._requested_version.channel channel_filter = f"channel={self._requested_version.channel_name}" url = self.VERSION_URL.format( platform=platform, channel="all", filter=f"version>={milestone},version<{milestone+1},{channel_filter}&") logging.debug("LIST ALL VERSIONS for M%s: %s", milestone, url) version_urls: List[Tuple[BrowserVersion, str]] = [] try: with helper.urlopen(url) as response: raw_infos = json.loads(response.read().decode("utf-8"))["versions"] version_urls = [ self._create_version_url( ChromeVersion( map(int, info["version"].split(".")), requested_channel)) for info in raw_infos ] except Exception as e: raise ValueError( f"Could not find version {self._requested_version} " f"for {self._browser_platform.name} {self._browser_platform.machine} " ) from e logging.debug("FILTERING %d CANDIDATES", len(version_urls)) return self._filter_candidate_urls(version_urls) def _create_version_url( self, version: BrowserVersion) -> Tuple[BrowserVersion, str]: # TODO: respect channel assert version.has_complete_parts return (version, f"{self.STORAGE_URL}{version.parts_str}/{self._platform_name}/") def _find_exact_archive_url(self) -> Tuple[BrowserVersion, Optional[str]]: # TODO: respect channel version, test_url = self._create_version_url(self._requested_version) logging.debug("LIST VERSIONS for M%s: %s", self._requested_version, test_url) return self._filter_candidate_urls([(version, test_url)]) def _filter_candidate_urls( self, versions_urls: List[Tuple[BrowserVersion, str]] ) -> Tuple[BrowserVersion, Optional[str]]: versions_urls.sort(key=lambda x: x[1], reverse=True) # Iterate from new to old version and and the first one that is older or # equal than the requested version. for version, url in versions_urls: if not self._requested_version.contains(version): logging.debug("Skipping download candidate: %s %s", version, url) continue for archive_version, archive_url in self._archive_urls(url, version): try: result = self.host_platform.sh_stdout(self.gsutil, "ls", archive_url) except SubprocessError as e: logging.debug("gsutil failed: %s", e) continue if result: return archive_version, archive_url return self._requested_version, None def _download_archive(self, archive_url: str, tmp_dir: pth.LocalPath) -> None: self.host_platform.sh(self.gsutil, "cp", archive_url, tmp_dir) archive_candidates = list(tmp_dir.glob("*")) assert len(archive_candidates) == 1, ( f"Download tmp dir contains more than one file: {tmp_dir}: " f"{archive_candidates}") candidate = archive_candidates[0] assert not self._archive_path.exists(), ( f"Archive was already downloaded: {self._archive_path}") candidate.replace(self._archive_path) class ChromeDownloaderLinux(ChromeDownloader): ARCHIVE_SUFFIX: str = ".rpm" @classmethod def is_valid(cls, path_or_identifier: pth.AnyPathLike, browser_platform: Platform) -> bool: return cls._is_valid(path_or_identifier, browser_platform) def __init__(self, version_identifier: Union[str, pth.LocalPath], browser_type: str, platform_name: str, browser_platform: Platform): assert not browser_type if browser_platform.is_linux and browser_platform.is_x64: platform_name = "linux64" else: raise ValueError("Unsupported linux architecture for downloading chrome: " f"got={browser_platform.machine} supported=linux.x64") super().__init__(version_identifier, "chrome", platform_name, browser_platform) def _installed_app_path(self) -> pth.LocalPath: dir_name = "chrome-unstable" if self._requested_version.is_stable or self._requested_version.is_unknown: dir_name = "chrome" if self._requested_version.is_beta: dir_name = "chrome-beta" return self._extracted_path() / "opt/google" / dir_name / "chrome" def _archive_urls( self, folder_url: str, version: BrowserVersion) -> Iterable[Tuple[BrowserVersion, str]]: parts_str = version.parts_str parts = version.parts stable = (ChromeVersion.stable(parts), f"{folder_url}google-chrome-stable-{parts_str}-1.x86_64.rpm") if version.is_stable: return (stable,) beta = (ChromeVersion.beta(parts), f"{folder_url}google-chrome-beta-{parts_str}-1.x86_64.rpm") if version.is_beta: return (beta,) dev = (ChromeVersion.alpha(parts), f"{folder_url}google-chrome-unstable-{parts_str}-1.x86_64.rpm") if version.is_alpha: return (dev,) if version.is_pre_alpha: raise ValueError(f"Canary not supported on linux: {version}") return (stable, beta, dev) def _install_archive(self, archive_path: pth.LocalPath) -> None: extracted_path = self._extracted_path() RPMArchiveHelper.extract(self.host_platform, archive_path, extracted_path) assert extracted_path.exists() class ChromeDownloaderMacOS(ChromeDownloader): ARCHIVE_SUFFIX: str = ".dmg" MIN_MAC_ARM64_MILESTONE: Final[int] = 87 @classmethod def is_valid(cls, path_or_identifier: pth.AnyPathLike, browser_platform: Platform) -> bool: return cls._is_valid(path_or_identifier, browser_platform) def __init__(self, version_identifier: Union[str, pth.LocalPath], browser_type: str, platform_name: str, browser_platform: Platform): assert not browser_type assert browser_platform.is_macos, f"{type(self)} can only be used on macOS" platform_name = "mac-universal" super().__init__(version_identifier, "chrome", platform_name, browser_platform) def _requested_version_validation(self) -> None: assert self._browser_platform.is_macos major_version: int = self._requested_version.major if (self._browser_platform.is_arm64 and (major_version < self.MIN_MAC_ARM64_MILESTONE)): raise ValueError( "Native Mac arm64/m1 Chrome version is available with M87, " f"but requested M{major_version}.") def _download_archive(self, archive_url: str, tmp_dir: pth.LocalPath) -> None: assert self._browser_platform.is_macos if self._browser_platform.is_arm64 and (self._requested_version.major < self.MIN_MAC_ARM64_MILESTONE): raise ValueError( "Chrome Arm64 Apple Silicon is only available starting with M87, " f"but requested {self._requested_version} is too old.") super()._download_archive(archive_url, tmp_dir) def _archive_urls( self, folder_url: str, version: BrowserVersion) -> Iterable[Tuple[BrowserVersion, str]]: # TODO: respect channel version_str: str = version.parts_str parts = version.parts stable = (ChromeVersion.stable(parts), f"{folder_url}GoogleChrome-{version_str}.dmg") if version.is_stable: return (stable,) beta = (ChromeVersion.beta(parts), f"{folder_url}GoogleChromeBeta-{version_str}.dmg") if version.is_beta: return (beta,) dev = (ChromeVersion.alpha(parts), f"{folder_url}GoogleChromeDev-{version_str}.dmg") if version.is_alpha: return (dev,) canary = (ChromeVersion.pre_alpha(parts), f"{folder_url}GoogleChromeCanary-{version_str}.dmg") if version.is_pre_alpha: return (canary,) return (stable, beta, dev, canary) def _extracted_path(self) -> pth.LocalPath: # TODO: support local vs remote return self._installed_app_path() def _installed_app_path(self) -> pth.LocalPath: return self._out_dir / f"Google Chrome {self._requested_version}.app" def _install_archive(self, archive_path: pth.LocalPath) -> None: extracted_path = self._extracted_path() if archive_path.suffix == ".dmg": DMGArchiveHelper.extract(self.host_platform, archive_path, extracted_path) else: raise ValueError(f"Unknown archive type: {archive_path}") assert extracted_path.exists() class ChromeDownloaderAndroid(ChromeDownloader): """The android downloader for Chrome pulls .apks and the corresponding .apk library and installs both on the attached device.""" ARCHIVE_SUFFIX: str = ".apks" LIBRARY_ARCHIVE_SUFFIX: str = ".lib.apk" STORAGE_URL: str = "gs://chrome-signed/android-B0urB0N/" MIN_HIGH_ARM_64_MILESTONE: Final[int] = 104 ARM_32_BUILD: Final[str] = "arm" ARM_64_BUILD: Final[str] = "arm_64" ARM_64_HIGH_BUILD: Final[str] = "high-arm_64" CHANNEL_PACKAGE_LOOKUP: Dict[str, Tuple[str, BrowserVersionChannel]] = { "Beta": ( "com.chrome.beta", BrowserVersionChannel.BETA, ), "Dev": ("com.chrome.dev", BrowserVersionChannel.ALPHA), "Canary": ("com.chrome.canary", BrowserVersionChannel.PRE_ALPHA), # Let's check stable last to avoid overriding the default installation # if possible. "Stable": ("com.android.chrome", BrowserVersionChannel.STABLE), } @classmethod def is_valid(cls, path_or_identifier: pth.AnyPathLike, browser_platform: Platform) -> bool: return cls._is_valid(path_or_identifier, browser_platform) def __init__(self, version_identifier: Union[str, pth.LocalPath], browser_type: str, platform_name: str, browser_platform: Platform): assert not browser_type assert browser_platform.is_android, ( f"{type(self)} can only be used on Android") # TODO: support more CPU types assert browser_platform.is_arm64, f"{type(self)} only supports arm64" # TODO: support low-end arm_64 and high-arm_64 at the same time. platform_name = "high-arm_64" super().__init__(version_identifier, "chrome", platform_name, browser_platform) @property def adb(self) -> Adb: return cast(AndroidAdbPlatform, self._browser_platform).adb def _pre_check(self) -> None: super()._pre_check() assert self._browser_platform.is_android, ( f"Expected android but got {self._browser_platform}") def _requested_version_validation(self) -> None: assert self._browser_platform.is_android # TODO: support custom android builds if self._requested_version.major < self.MIN_HIGH_ARM_64_MILESTONE: self._platform_name = self.ARM_64_BUILD else: self._platform_name = self.ARM_64_HIGH_BUILD def _installed_app_version(self, app_path: pth.LocalPath) -> BrowserVersion: raw_version = self._browser_platform.app_version(app_path) channel = BrowserVersionChannel.STABLE for value in self.CHANNEL_PACKAGE_LOOKUP.values(): (package_name, package_channel) = value if app_path.name == package_name: channel = package_channel break return ChromeVersion.parse(raw_version, channel) def _archive_urls( self, folder_url: str, version: BrowserVersion) -> Iterable[Tuple[BrowserVersion, str]]: prefix: str = f"{folder_url}" urls: List[Tuple[BrowserVersion, str]] = [] # TODO: pass in correct sdk_level package = self._get_chrome_package(100) # TODO: respect version channel for channel_name, (_, channel) in self.CHANNEL_PACKAGE_LOOKUP.items(): channel_version = ChromeVersion(version.parts, channel) version_url = (channel_version, f"{prefix}{package}{channel_name}{self.ARCHIVE_SUFFIX}") if version.matches_channel(channel_version.channel): return (version_url,) urls.append(version_url) return tuple(urls) def _get_chrome_package(self, sdk_level: int) -> str: del sdk_level # TODO support older SDKs at some point # if sdk_level < 19: # raise RuntimeError( # f"Clank can only be installed on >= 19, not {sdk_level}") # if sdk_level < 21: # return "Chrome" # if sdk_level < 24: # return "ChromeModern" # if sdk_level < 29: # return "Monochrome" return "TrichromeChromeGoogle6432" def _extracted_path(self) -> pth.LocalPath: return self._archive_path def _installed_app_path(self) -> pth.LocalPath: for channel, (package_name, _) in self.CHANNEL_PACKAGE_LOOKUP.items(): if channel in self._archive_url: logging.debug("Using package: %s", package_name) return pth.LocalPath(package_name) package_name, _ = self.CHANNEL_PACKAGE_LOOKUP["Stable"] return pth.LocalPath(package_name) def _find_matching_installed_version(self) -> Optional[pth.LocalPath]: # TODO: we should use aapt and read the package name directly from # the apk: `aapt dump badging | grep package:\ name` # Iterate over all chrome versions and find any matching release installed_packages = self.adb.packages() for value in self.CHANNEL_PACKAGE_LOOKUP.values(): (package_name, package_channel) = value if not self._requested_version.matches_channel(package_channel): continue if package_name not in installed_packages: continue try: package = pth.LocalPath(package_name) self._validate_installed(package) return package except IncompatibleVersionError as e: logging.debug("Ignoring installed package %s: %s", package_name, e) return None def _download_archive(self, archive_url: str, tmp_dir: pth.LocalPath) -> None: super()._download_archive(archive_url, tmp_dir) if "TrichromeChromeGoogle" not in archive_url: return # Download TrichromeLibrary.apk needed by TrichromeChromeGoogle.apks with self._prepare_lib_archive_download(archive_url) as (lib_archive_url, lib_tmp_dir): super()._download_archive(lib_archive_url, lib_tmp_dir) @contextlib.contextmanager def _prepare_lib_archive_download(self, archive_url: str): # Also download the trichrome library (such a mess) main_archive_path = self._archive_path lib_archive_path = main_archive_path.with_suffix( self.LIBRARY_ARCHIVE_SUFFIX) if lib_archive_path.exists(): return self._archive_path = lib_archive_path lib_url = archive_url.replace("TrichromeChromeGoogle", "TrichromeLibraryGoogle") lib_url = lib_url.replace(self.ARCHIVE_SUFFIX, ".apk") with tempfile.TemporaryDirectory(prefix="cb_download_") as tmp_dir_name: lib_tmp_dir = pth.LocalPath(tmp_dir_name) yield lib_url, lib_tmp_dir self._archive_path = main_archive_path def _install_archive(self, archive_path: pth.LocalPath) -> None: # TODO: move browser installation to browser startup to allow # multiple versions on android in a single crossbench invocation package = str(self._installed_app_path()) self.adb.uninstall(package, missing_ok=True) lib_archive_path = archive_path.with_suffix(self.LIBRARY_ARCHIVE_SUFFIX) if lib_archive_path.exists(): self.adb.install(lib_archive_path, allow_downgrade=True, modules="_ALL_") self.adb.install(archive_path, allow_downgrade=True, modules="_ALL_") class ChromeDownloaderWin(ChromeDownloader): ARCHIVE_SUFFIX: str = ".zip" ARCHIVE_STEM: str = "chrome-win64-clang" STORAGE_URL: str = "gs://chrome-unsigned/desktop-5c0tCh/" @classmethod def is_valid(cls, path_or_identifier: pth.AnyPathLike, browser_platform: Platform) -> bool: return cls._is_valid(path_or_identifier, browser_platform) def __init__(self, version_identifier: Union[str, pth.LocalPath], browser_type: str, platform_name: str, browser_platform: Platform): assert not browser_type assert browser_platform.is_win, f"{type(self)} can only be used on windows" platform_name = "win64-clang" super().__init__(version_identifier, "chrome", platform_name, browser_platform) def _archive_urls( self, folder_url: str, version: BrowserVersion) -> Iterable[Tuple[BrowserVersion, str]]: parts = version.parts stable = (ChromeVersion.stable(parts), f"{folder_url}{self.ARCHIVE_STEM}.zip") return (stable,) def _extracted_path(self) -> pth.LocalPath: # TODO: support local vs remote return self._out_dir / f"Google Chrome {self._requested_version}" def _installed_app_path(self) -> pth.LocalPath: return self._extracted_path() / "chrome.exe" def _install_archive(self, archive_path: pth.LocalPath) -> None: extracted_path = self._extracted_path() tmp_path = self.host_platform.mkdtemp() with zipfile.ZipFile(archive_path, "r") as zip_file: zip_file.extractall(tmp_path) self.host_platform.rename(tmp_path / self.ARCHIVE_STEM, extracted_path) assert self.host_platform.is_dir(extracted_path), "Could not extract"