tor-browser

The Tor Browser
git clone https://git.dasho.dev/tor-browser.git
Log | Files | Refs | README | LICENSE

compatible_utils.py (4895B)

      1 # Copyright 2022 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 """Functions used in both v1 and v2 scripts."""
      5 
      6 import json
      7 import os
      8 import platform
      9 import stat
     10 
     11 from typing import Iterable, List, Tuple
     12 
     13 
     14 _FILTER_DIR = 'testing/buildbot/filters'
     15 _SSH_KEYS = os.path.expanduser('~/.ssh/fuchsia_authorized_keys')
     16 _CHROME_HEADLESS = 'CHROME_HEADLESS'
     17 _SWARMING_SERVER = 'SWARMING_SERVER'
     18 
     19 class VersionNotFoundError(Exception):
     20    """Thrown when version info cannot be retrieved from device."""
     21 
     22 
     23 def get_ssh_keys() -> str:
     24    """Returns path of Fuchsia ssh keys."""
     25 
     26    return _SSH_KEYS
     27 
     28 
     29 def running_unattended() -> bool:
     30    """Returns true if running non-interactively.
     31 
     32    When running unattended, confirmation prompts and the like are suppressed.
     33    """
     34 
     35    # TODO(crbug.com/40884247): Change to mixin based approach.
     36    # And remove SWARMING_SERVER check when it's no longer needed by dart,
     37    # eureka and flutter to partially revert https://crrev.com/c/4112522.
     38    return _CHROME_HEADLESS in os.environ or _SWARMING_SERVER in os.environ
     39 
     40 
     41 def force_running_unattended() -> None:
     42    """Treats everything as running non-interactively."""
     43    if not running_unattended():
     44        os.environ[_CHROME_HEADLESS] = '1'
     45        assert running_unattended()
     46 
     47 
     48 def force_running_attended() -> None:
     49    """Treats everything as running interactively."""
     50    if running_unattended():
     51        os.environ.pop(_CHROME_HEADLESS, None)
     52        os.environ.pop(_SWARMING_SERVER, None)
     53        assert not running_unattended()
     54 
     55 
     56 def get_host_arch() -> str:
     57    """Retrieve CPU architecture of the host machine. """
     58    host_arch = platform.machine()
     59    # platform.machine() returns AMD64 on 64-bit Windows.
     60    if host_arch in ['x86_64', 'AMD64']:
     61        return 'x64'
     62    if host_arch in ['aarch64', 'arm64']:
     63        return 'arm64'
     64    raise NotImplementedError('Unsupported host architecture: %s' % host_arch)
     65 
     66 
     67 def add_exec_to_file(file: str) -> None:
     68    """Add execution bits to a file.
     69 
     70    Args:
     71        file: path to the file.
     72    """
     73    file_stat = os.stat(file)
     74    os.chmod(file, file_stat.st_mode | stat.S_IXUSR)
     75 
     76 
     77 def get_ssh_prefix(host_port_pair: Tuple[str, int]) -> List[str]:
     78    """Get the prefix of a barebone ssh command."""
     79    return [
     80        'ssh', '-F',
     81        os.path.join(os.path.dirname(__file__), 'sshconfig'),
     82        host_port_pair[0], '-p',
     83        str(host_port_pair[1])
     84    ]
     85 
     86 
     87 def install_symbols(package_paths: Iterable[str],
     88                    fuchsia_out_dir: str) -> None:
     89    """Installs debug symbols for a package into the GDB-standard symbol
     90    directory located in fuchsia_out_dir."""
     91 
     92    symbol_root = os.path.join(fuchsia_out_dir, '.build-id')
     93    for path in package_paths:
     94        package_dir = os.path.dirname(path)
     95        ids_txt_path = os.path.join(package_dir, 'ids.txt')
     96        with open(ids_txt_path, 'r') as f:
     97            for entry in f:
     98                build_id, binary_relpath = entry.strip().split(' ')
     99                binary_abspath = os.path.abspath(
    100                    os.path.join(package_dir, binary_relpath))
    101                symbol_dir = os.path.join(symbol_root, build_id[:2])
    102                symbol_file = os.path.join(symbol_dir, build_id[2:] + '.debug')
    103                if not os.path.exists(symbol_dir):
    104                    os.makedirs(symbol_dir)
    105 
    106                if os.path.islink(symbol_file) or os.path.exists(symbol_file):
    107                    # Clobber the existing entry to ensure that the symlink's
    108                    # target is up to date.
    109                    os.unlink(symbol_file)
    110                os.symlink(os.path.relpath(binary_abspath, symbol_dir),
    111                           symbol_file)
    112 
    113 
    114 # TODO(crbug.com/42050403): Until one can send files to the device when running
    115 # a test, filter files must be read from the test package.
    116 def map_filter_file_to_package_file(filter_file: str) -> str:
    117    """Returns the path to |filter_file| within the test component's package."""
    118 
    119    if not _FILTER_DIR in filter_file:
    120        raise ValueError('CFv2 tests only support registered filter files '
    121                         'present in the test package')
    122    return '/pkg/' + filter_file[filter_file.index(_FILTER_DIR):]
    123 
    124 
    125 # TODO(crbug.com/40938340): Rename to get_product_version.
    126 def get_sdk_hash(system_image_dir: str) -> Tuple[str, str]:
    127    """Read version of hash in pre-installed package directory.
    128    Returns:
    129        Tuple of (product, version) of image to be installed.
    130    """
    131 
    132    with open(os.path.join(system_image_dir,
    133                           'product_bundle.json')) as product:
    134        # The product_name in the json file does not match the name of the image
    135        # flashed to the device.
    136        return (os.path.basename(os.path.normpath(system_image_dir)),
    137                json.load(product)['product_version'])