1# Copyright 2023 The Chromium Authors 2# Use of this source code is governed by a BSD-style license that can be 3# found in the LICENSE file. 4 5from __future__ import annotations 6 7import contextlib 8import json 9import logging 10import tempfile 11import zipfile 12from typing import (TYPE_CHECKING, Dict, Final, Iterable, List, Optional, Tuple, 13 Type, Union, cast) 14 15from crossbench import helper 16from crossbench import path as pth 17from crossbench.browsers.chrome.version import ChromeVersion 18from crossbench.browsers.downloader import (DMGArchiveHelper, Downloader, 19 IncompatibleVersionError, 20 RPMArchiveHelper) 21from crossbench.browsers.version import BrowserVersion, BrowserVersionChannel 22from crossbench.plt.android_adb import AndroidAdbPlatform 23from crossbench.plt.base import SubprocessError 24 25if TYPE_CHECKING: 26 from crossbench.plt.android_adb import Adb 27 from crossbench.plt.base import Platform 28 29 30class ChromeDownloader(Downloader): 31 STORAGE_URL: str = "gs://chrome-signed/desktop-5c0tCh/" 32 VERSION_URL = ( 33 "https://versionhistory.googleapis.com/v1/" 34 "chrome/platforms/{platform}/channels/{channel}/versions?filter={filter}") 35 VERSION_URL_PLATFORM_LOOKUP: Dict[Tuple[str, str], str] = { 36 ("win", "ia32"): "win", 37 ("win", "x64"): "win64", 38 ("linux", "x64"): "linux", 39 ("macos", "x64"): "mac", 40 ("macos", "arm64"): "mac_arm64", 41 ("android", "arm64"): "android", 42 } 43 44 def __init__(self, *args, **kwargs): 45 self._gsutil: Optional[pth.AnyPath] = None 46 super().__init__(*args, **kwargs) 47 48 @classmethod 49 def is_valid_version(cls, path_or_identifier: str): 50 return ChromeVersion.is_valid_unique(path_or_identifier) 51 52 @classmethod 53 def _is_valid(cls, path_or_identifier: pth.AnyPathLike, 54 browser_platform: Platform) -> bool: 55 if cls.is_valid_version(str(path_or_identifier)): 56 return True 57 path = browser_platform.path(path_or_identifier) 58 return (browser_platform.exists(path) and 59 path.name.endswith(cls.ARCHIVE_SUFFIX)) 60 61 @classmethod 62 def _get_loader_cls(cls, 63 browser_platform: Platform) -> Type[ChromeDownloader]: 64 if browser_platform.is_macos: 65 return ChromeDownloaderMacOS 66 if browser_platform.is_linux: 67 return ChromeDownloaderLinux 68 if browser_platform.is_win: 69 return ChromeDownloaderWin 70 if browser_platform.is_android: 71 return ChromeDownloaderAndroid 72 raise ValueError( 73 "Downloading chrome is only supported on linux and macOS, " 74 f"but not on {browser_platform.name} {browser_platform.machine}") 75 76 def _pre_check(self) -> None: 77 super()._pre_check() 78 if not self._requested_version: 79 return 80 self._gsutil = self.host_platform.which("gsutil") 81 if not self._gsutil: 82 raise ValueError( 83 f"Cannot download chrome version {self._requested_version}: " 84 "please install gsutil.\n" 85 "- https://cloud.google.com/storage/docs/gsutil_install\n" 86 "- Run 'gcloud auth login' to get access to the archives " 87 "(googlers only).") 88 89 @property 90 def gsutil(self) -> pth.AnyPath: 91 assert self._gsutil, "gsutil not be found." 92 return self._gsutil 93 94 def _requested_version_validation(self) -> None: 95 pass 96 97 def _parse_version(self, version_identifier: str) -> BrowserVersion: 98 return ChromeVersion.parse_unique(version_identifier) 99 100 def _find_archive_url(self) -> Tuple[BrowserVersion, Optional[str]]: 101 # Quick probe for complete versions 102 if self._requested_version.is_complete: 103 return self._find_exact_archive_url() 104 return self._find_milestone_archive_url() 105 106 def _find_milestone_archive_url(self) -> Tuple[BrowserVersion, Optional[str]]: 107 milestone: int = self._requested_version.major 108 platform = self.VERSION_URL_PLATFORM_LOOKUP.get(self._browser_platform.key) 109 if not platform: 110 raise ValueError(f"Unsupported platform {self._browser_platform}") 111 # Version ordering is: stable < beta < dev < canary < canary_asan 112 # See https://developer.chrome.com/docs/web-platform/versionhistory/reference#filter 113 channel_filter = "channel<=canary" 114 requested_channel = BrowserVersionChannel.ANY 115 if self._requested_version.has_channel: 116 requested_channel = self._requested_version.channel 117 channel_filter = f"channel={self._requested_version.channel_name}" 118 119 url = self.VERSION_URL.format( 120 platform=platform, 121 channel="all", 122 filter=f"version>={milestone},version<{milestone+1},{channel_filter}&") 123 logging.debug("LIST ALL VERSIONS for M%s: %s", milestone, url) 124 version_urls: List[Tuple[BrowserVersion, str]] = [] 125 try: 126 with helper.urlopen(url) as response: 127 raw_infos = json.loads(response.read().decode("utf-8"))["versions"] 128 version_urls = [ 129 self._create_version_url( 130 ChromeVersion( 131 map(int, info["version"].split(".")), requested_channel)) 132 for info in raw_infos 133 ] 134 except Exception as e: 135 raise ValueError( 136 f"Could not find version {self._requested_version} " 137 f"for {self._browser_platform.name} {self._browser_platform.machine} " 138 ) from e 139 logging.debug("FILTERING %d CANDIDATES", len(version_urls)) 140 return self._filter_candidate_urls(version_urls) 141 142 def _create_version_url( 143 self, version: BrowserVersion) -> Tuple[BrowserVersion, str]: 144 # TODO: respect channel 145 assert version.has_complete_parts 146 return (version, 147 f"{self.STORAGE_URL}{version.parts_str}/{self._platform_name}/") 148 149 def _find_exact_archive_url(self) -> Tuple[BrowserVersion, Optional[str]]: 150 # TODO: respect channel 151 version, test_url = self._create_version_url(self._requested_version) 152 logging.debug("LIST VERSIONS for M%s: %s", self._requested_version, 153 test_url) 154 return self._filter_candidate_urls([(version, test_url)]) 155 156 def _filter_candidate_urls( 157 self, versions_urls: List[Tuple[BrowserVersion, str]] 158 ) -> Tuple[BrowserVersion, Optional[str]]: 159 versions_urls.sort(key=lambda x: x[1], reverse=True) 160 # Iterate from new to old version and and the first one that is older or 161 # equal than the requested version. 162 for version, url in versions_urls: 163 if not self._requested_version.contains(version): 164 logging.debug("Skipping download candidate: %s %s", version, url) 165 continue 166 for archive_version, archive_url in self._archive_urls(url, version): 167 try: 168 result = self.host_platform.sh_stdout(self.gsutil, "ls", archive_url) 169 except SubprocessError as e: 170 logging.debug("gsutil failed: %s", e) 171 continue 172 if result: 173 return archive_version, archive_url 174 return self._requested_version, None 175 176 def _download_archive(self, archive_url: str, tmp_dir: pth.LocalPath) -> None: 177 self.host_platform.sh(self.gsutil, "cp", archive_url, tmp_dir) 178 archive_candidates = list(tmp_dir.glob("*")) 179 assert len(archive_candidates) == 1, ( 180 f"Download tmp dir contains more than one file: {tmp_dir}: " 181 f"{archive_candidates}") 182 candidate = archive_candidates[0] 183 assert not self._archive_path.exists(), ( 184 f"Archive was already downloaded: {self._archive_path}") 185 candidate.replace(self._archive_path) 186 187 188class ChromeDownloaderLinux(ChromeDownloader): 189 ARCHIVE_SUFFIX: str = ".rpm" 190 191 @classmethod 192 def is_valid(cls, path_or_identifier: pth.AnyPathLike, 193 browser_platform: Platform) -> bool: 194 return cls._is_valid(path_or_identifier, browser_platform) 195 196 def __init__(self, version_identifier: Union[str, pth.LocalPath], 197 browser_type: str, platform_name: str, 198 browser_platform: Platform): 199 assert not browser_type 200 if browser_platform.is_linux and browser_platform.is_x64: 201 platform_name = "linux64" 202 else: 203 raise ValueError("Unsupported linux architecture for downloading chrome: " 204 f"got={browser_platform.machine} supported=linux.x64") 205 super().__init__(version_identifier, "chrome", platform_name, 206 browser_platform) 207 208 def _installed_app_path(self) -> pth.LocalPath: 209 dir_name = "chrome-unstable" 210 if self._requested_version.is_stable or self._requested_version.is_unknown: 211 dir_name = "chrome" 212 if self._requested_version.is_beta: 213 dir_name = "chrome-beta" 214 return self._extracted_path() / "opt/google" / dir_name / "chrome" 215 216 def _archive_urls( 217 self, folder_url: str, 218 version: BrowserVersion) -> Iterable[Tuple[BrowserVersion, str]]: 219 parts_str = version.parts_str 220 parts = version.parts 221 stable = (ChromeVersion.stable(parts), 222 f"{folder_url}google-chrome-stable-{parts_str}-1.x86_64.rpm") 223 if version.is_stable: 224 return (stable,) 225 beta = (ChromeVersion.beta(parts), 226 f"{folder_url}google-chrome-beta-{parts_str}-1.x86_64.rpm") 227 if version.is_beta: 228 return (beta,) 229 dev = (ChromeVersion.alpha(parts), 230 f"{folder_url}google-chrome-unstable-{parts_str}-1.x86_64.rpm") 231 if version.is_alpha: 232 return (dev,) 233 if version.is_pre_alpha: 234 raise ValueError(f"Canary not supported on linux: {version}") 235 return (stable, beta, dev) 236 237 def _install_archive(self, archive_path: pth.LocalPath) -> None: 238 extracted_path = self._extracted_path() 239 RPMArchiveHelper.extract(self.host_platform, archive_path, extracted_path) 240 assert extracted_path.exists() 241 242 243class ChromeDownloaderMacOS(ChromeDownloader): 244 ARCHIVE_SUFFIX: str = ".dmg" 245 MIN_MAC_ARM64_MILESTONE: Final[int] = 87 246 247 @classmethod 248 def is_valid(cls, path_or_identifier: pth.AnyPathLike, 249 browser_platform: Platform) -> bool: 250 return cls._is_valid(path_or_identifier, browser_platform) 251 252 def __init__(self, version_identifier: Union[str, pth.LocalPath], 253 browser_type: str, platform_name: str, 254 browser_platform: Platform): 255 assert not browser_type 256 assert browser_platform.is_macos, f"{type(self)} can only be used on macOS" 257 platform_name = "mac-universal" 258 super().__init__(version_identifier, "chrome", platform_name, 259 browser_platform) 260 261 def _requested_version_validation(self) -> None: 262 assert self._browser_platform.is_macos 263 major_version: int = self._requested_version.major 264 if (self._browser_platform.is_arm64 and 265 (major_version < self.MIN_MAC_ARM64_MILESTONE)): 266 raise ValueError( 267 "Native Mac arm64/m1 Chrome version is available with M87, " 268 f"but requested M{major_version}.") 269 270 def _download_archive(self, archive_url: str, tmp_dir: pth.LocalPath) -> None: 271 assert self._browser_platform.is_macos 272 if self._browser_platform.is_arm64 and (self._requested_version.major 273 < self.MIN_MAC_ARM64_MILESTONE): 274 raise ValueError( 275 "Chrome Arm64 Apple Silicon is only available starting with M87, " 276 f"but requested {self._requested_version} is too old.") 277 super()._download_archive(archive_url, tmp_dir) 278 279 def _archive_urls( 280 self, folder_url: str, 281 version: BrowserVersion) -> Iterable[Tuple[BrowserVersion, str]]: 282 # TODO: respect channel 283 version_str: str = version.parts_str 284 parts = version.parts 285 stable = (ChromeVersion.stable(parts), 286 f"{folder_url}GoogleChrome-{version_str}.dmg") 287 if version.is_stable: 288 return (stable,) 289 beta = (ChromeVersion.beta(parts), 290 f"{folder_url}GoogleChromeBeta-{version_str}.dmg") 291 if version.is_beta: 292 return (beta,) 293 dev = (ChromeVersion.alpha(parts), 294 f"{folder_url}GoogleChromeDev-{version_str}.dmg") 295 if version.is_alpha: 296 return (dev,) 297 canary = (ChromeVersion.pre_alpha(parts), 298 f"{folder_url}GoogleChromeCanary-{version_str}.dmg") 299 if version.is_pre_alpha: 300 return (canary,) 301 return (stable, beta, dev, canary) 302 303 def _extracted_path(self) -> pth.LocalPath: 304 # TODO: support local vs remote 305 return self._installed_app_path() 306 307 def _installed_app_path(self) -> pth.LocalPath: 308 return self._out_dir / f"Google Chrome {self._requested_version}.app" 309 310 def _install_archive(self, archive_path: pth.LocalPath) -> None: 311 extracted_path = self._extracted_path() 312 if archive_path.suffix == ".dmg": 313 DMGArchiveHelper.extract(self.host_platform, archive_path, extracted_path) 314 else: 315 raise ValueError(f"Unknown archive type: {archive_path}") 316 assert extracted_path.exists() 317 318 319class ChromeDownloaderAndroid(ChromeDownloader): 320 """The android downloader for Chrome pulls .apks and the 321 corresponding .apk library and installs both on the attached device.""" 322 ARCHIVE_SUFFIX: str = ".apks" 323 LIBRARY_ARCHIVE_SUFFIX: str = ".lib.apk" 324 STORAGE_URL: str = "gs://chrome-signed/android-B0urB0N/" 325 326 MIN_HIGH_ARM_64_MILESTONE: Final[int] = 104 327 ARM_32_BUILD: Final[str] = "arm" 328 ARM_64_BUILD: Final[str] = "arm_64" 329 ARM_64_HIGH_BUILD: Final[str] = "high-arm_64" 330 331 CHANNEL_PACKAGE_LOOKUP: Dict[str, Tuple[str, BrowserVersionChannel]] = { 332 "Beta": ( 333 "com.chrome.beta", 334 BrowserVersionChannel.BETA, 335 ), 336 "Dev": ("com.chrome.dev", BrowserVersionChannel.ALPHA), 337 "Canary": ("com.chrome.canary", BrowserVersionChannel.PRE_ALPHA), 338 # Let's check stable last to avoid overriding the default installation 339 # if possible. 340 "Stable": ("com.android.chrome", BrowserVersionChannel.STABLE), 341 } 342 343 @classmethod 344 def is_valid(cls, path_or_identifier: pth.AnyPathLike, 345 browser_platform: Platform) -> bool: 346 return cls._is_valid(path_or_identifier, browser_platform) 347 348 def __init__(self, version_identifier: Union[str, pth.LocalPath], 349 browser_type: str, platform_name: str, 350 browser_platform: Platform): 351 assert not browser_type 352 assert browser_platform.is_android, ( 353 f"{type(self)} can only be used on Android") 354 # TODO: support more CPU types 355 assert browser_platform.is_arm64, f"{type(self)} only supports arm64" 356 # TODO: support low-end arm_64 and high-arm_64 at the same time. 357 platform_name = "high-arm_64" 358 super().__init__(version_identifier, "chrome", platform_name, 359 browser_platform) 360 361 @property 362 def adb(self) -> Adb: 363 return cast(AndroidAdbPlatform, self._browser_platform).adb 364 365 def _pre_check(self) -> None: 366 super()._pre_check() 367 assert self._browser_platform.is_android, ( 368 f"Expected android but got {self._browser_platform}") 369 370 def _requested_version_validation(self) -> None: 371 assert self._browser_platform.is_android 372 # TODO: support custom android builds 373 if self._requested_version.major < self.MIN_HIGH_ARM_64_MILESTONE: 374 self._platform_name = self.ARM_64_BUILD 375 else: 376 self._platform_name = self.ARM_64_HIGH_BUILD 377 378 def _installed_app_version(self, app_path: pth.LocalPath) -> BrowserVersion: 379 raw_version = self._browser_platform.app_version(app_path) 380 channel = BrowserVersionChannel.STABLE 381 for value in self.CHANNEL_PACKAGE_LOOKUP.values(): 382 (package_name, package_channel) = value 383 if app_path.name == package_name: 384 channel = package_channel 385 break 386 return ChromeVersion.parse(raw_version, channel) 387 388 def _archive_urls( 389 self, folder_url: str, 390 version: BrowserVersion) -> Iterable[Tuple[BrowserVersion, str]]: 391 prefix: str = f"{folder_url}" 392 urls: List[Tuple[BrowserVersion, str]] = [] 393 # TODO: pass in correct sdk_level 394 package = self._get_chrome_package(100) 395 # TODO: respect version channel 396 for channel_name, (_, channel) in self.CHANNEL_PACKAGE_LOOKUP.items(): 397 channel_version = ChromeVersion(version.parts, channel) 398 version_url = (channel_version, 399 f"{prefix}{package}{channel_name}{self.ARCHIVE_SUFFIX}") 400 if version.matches_channel(channel_version.channel): 401 return (version_url,) 402 urls.append(version_url) 403 return tuple(urls) 404 405 def _get_chrome_package(self, sdk_level: int) -> str: 406 del sdk_level 407 # TODO support older SDKs at some point 408 # if sdk_level < 19: 409 # raise RuntimeError( 410 # f"Clank can only be installed on >= 19, not {sdk_level}") 411 # if sdk_level < 21: 412 # return "Chrome" 413 # if sdk_level < 24: 414 # return "ChromeModern" 415 # if sdk_level < 29: 416 # return "Monochrome" 417 return "TrichromeChromeGoogle6432" 418 419 def _extracted_path(self) -> pth.LocalPath: 420 return self._archive_path 421 422 def _installed_app_path(self) -> pth.LocalPath: 423 for channel, (package_name, _) in self.CHANNEL_PACKAGE_LOOKUP.items(): 424 if channel in self._archive_url: 425 logging.debug("Using package: %s", package_name) 426 return pth.LocalPath(package_name) 427 package_name, _ = self.CHANNEL_PACKAGE_LOOKUP["Stable"] 428 return pth.LocalPath(package_name) 429 430 def _find_matching_installed_version(self) -> Optional[pth.LocalPath]: 431 # TODO: we should use aapt and read the package name directly from 432 # the apk: `aapt dump badging <path-to-apk> | grep package:\ name` 433 # Iterate over all chrome versions and find any matching release 434 installed_packages = self.adb.packages() 435 for value in self.CHANNEL_PACKAGE_LOOKUP.values(): 436 (package_name, package_channel) = value 437 if not self._requested_version.matches_channel(package_channel): 438 continue 439 if package_name not in installed_packages: 440 continue 441 try: 442 package = pth.LocalPath(package_name) 443 self._validate_installed(package) 444 return package 445 except IncompatibleVersionError as e: 446 logging.debug("Ignoring installed package %s: %s", package_name, e) 447 return None 448 449 def _download_archive(self, archive_url: str, tmp_dir: pth.LocalPath) -> None: 450 super()._download_archive(archive_url, tmp_dir) 451 if "TrichromeChromeGoogle" not in archive_url: 452 return 453 # Download TrichromeLibrary.apk needed by TrichromeChromeGoogle.apks 454 with self._prepare_lib_archive_download(archive_url) as (lib_archive_url, 455 lib_tmp_dir): 456 super()._download_archive(lib_archive_url, lib_tmp_dir) 457 458 @contextlib.contextmanager 459 def _prepare_lib_archive_download(self, archive_url: str): 460 # Also download the trichrome library (such a mess) 461 main_archive_path = self._archive_path 462 lib_archive_path = main_archive_path.with_suffix( 463 self.LIBRARY_ARCHIVE_SUFFIX) 464 if lib_archive_path.exists(): 465 return 466 self._archive_path = lib_archive_path 467 lib_url = archive_url.replace("TrichromeChromeGoogle", 468 "TrichromeLibraryGoogle") 469 lib_url = lib_url.replace(self.ARCHIVE_SUFFIX, ".apk") 470 with tempfile.TemporaryDirectory(prefix="cb_download_") as tmp_dir_name: 471 lib_tmp_dir = pth.LocalPath(tmp_dir_name) 472 yield lib_url, lib_tmp_dir 473 self._archive_path = main_archive_path 474 475 def _install_archive(self, archive_path: pth.LocalPath) -> None: 476 # TODO: move browser installation to browser startup to allow 477 # multiple versions on android in a single crossbench invocation 478 package = str(self._installed_app_path()) 479 self.adb.uninstall(package, missing_ok=True) 480 lib_archive_path = archive_path.with_suffix(self.LIBRARY_ARCHIVE_SUFFIX) 481 if lib_archive_path.exists(): 482 self.adb.install(lib_archive_path, allow_downgrade=True, modules="_ALL_") 483 self.adb.install(archive_path, allow_downgrade=True, modules="_ALL_") 484 485 486class ChromeDownloaderWin(ChromeDownloader): 487 ARCHIVE_SUFFIX: str = ".zip" 488 ARCHIVE_STEM: str = "chrome-win64-clang" 489 STORAGE_URL: str = "gs://chrome-unsigned/desktop-5c0tCh/" 490 491 @classmethod 492 def is_valid(cls, path_or_identifier: pth.AnyPathLike, 493 browser_platform: Platform) -> bool: 494 return cls._is_valid(path_or_identifier, browser_platform) 495 496 def __init__(self, version_identifier: Union[str, pth.LocalPath], 497 browser_type: str, platform_name: str, 498 browser_platform: Platform): 499 assert not browser_type 500 assert browser_platform.is_win, f"{type(self)} can only be used on windows" 501 platform_name = "win64-clang" 502 super().__init__(version_identifier, "chrome", platform_name, 503 browser_platform) 504 505 def _archive_urls( 506 self, folder_url: str, 507 version: BrowserVersion) -> Iterable[Tuple[BrowserVersion, str]]: 508 parts = version.parts 509 stable = (ChromeVersion.stable(parts), 510 f"{folder_url}{self.ARCHIVE_STEM}.zip") 511 return (stable,) 512 513 def _extracted_path(self) -> pth.LocalPath: 514 # TODO: support local vs remote 515 return self._out_dir / f"Google Chrome {self._requested_version}" 516 517 def _installed_app_path(self) -> pth.LocalPath: 518 return self._extracted_path() / "chrome.exe" 519 520 def _install_archive(self, archive_path: pth.LocalPath) -> None: 521 extracted_path = self._extracted_path() 522 tmp_path = self.host_platform.mkdtemp() 523 with zipfile.ZipFile(archive_path, "r") as zip_file: 524 zip_file.extractall(tmp_path) 525 self.host_platform.rename(tmp_path / self.ARCHIVE_STEM, extracted_path) 526 assert self.host_platform.is_dir(extracted_path), "Could not extract" 527