File: utils/filescfg.py

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-

# Copyright (c) 2019 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""
Operations with FILES.cfg (for portable packages)
"""

import argparse
import platform
import sys
import tarfile
import zipfile
from pathlib import Path

from _common import get_logger, add_common_params


def filescfg_generator(cfg_path, build_outputs, cpu_arch):
    """
    Generator that yields pathlib.Path relative to the build outputs according to FILES.cfg

    cfg_path is a pathlib.Path to the FILES.cfg
    build_outputs is a pathlib.Path to the build outputs directory.
    cpu_arch is a platform.architecture() string
    """
    resolved_build_outputs = build_outputs.resolve()
    exec_globals = {'__builtins__': None}
    with cfg_path.open() as cfg_file:
        exec(cfg_file.read(), exec_globals) # pylint: disable=exec-used
    for file_spec in exec_globals['FILES']:
        # Only include files for official builds
        if 'official' not in file_spec['buildtype']:
            continue
        # If a file has an 'arch' field, it must have cpu_arch to be included
        if 'arch' in file_spec and cpu_arch not in file_spec['arch']:
            continue
        # From chrome/tools/build/make_zip.py, 'filename' is actually a glob pattern
        for file_path in resolved_build_outputs.glob(file_spec['filename']):
            # Do not package Windows debugging symbols
            if file_path.suffix.lower() == '.pdb':
                continue
            yield file_path.relative_to(resolved_build_outputs)


def _get_archive_writer(output_path):
    """
    Detects and returns the appropriate archive writer

    output_path is the pathlib.Path of the archive to write
    """
    if not output_path.suffixes:
        raise ValueError('Output name has no suffix: %s' % output_path.name)
    if output_path.suffixes[-1].lower() == '.zip':
        archive_root = Path(output_path.stem)
        output_archive = zipfile.ZipFile(str(output_path), 'w', zipfile.ZIP_DEFLATED)

        def add_func(in_path, arc_path):
            """Add files to zip archive"""
            if in_path.is_dir():
                for sub_path in in_path.rglob('*'):
                    output_archive.write(str(sub_path),
                                         str(arc_path / sub_path.relative_to(in_path)))
            else:
                output_archive.write(str(in_path), str(arc_path))
    elif '.tar' in output_path.name.lower():
        if len(output_path.suffixes) >= 2 and output_path.suffixes[-2].lower() == '.tar':
            tar_mode = 'w:%s' % output_path.suffixes[-1][1:]
            archive_root = Path(output_path.with_suffix('').stem)
        elif output_path.suffixes[-1].lower() == '.tar':
            tar_mode = 'w'
            archive_root = Path(output_path.stem)
        else:
            raise ValueError('Could not detect tar format for output: %s' % output_path.name)
        output_archive = tarfile.open(str(output_path), tar_mode)
        add_func = lambda in_path, arc_path: output_archive.add(str(in_path), str(arc_path))
    else:
        raise ValueError('Unknown archive extension with name: %s' % output_path.name)
    return output_archive, add_func, archive_root


def create_archive(file_iter, include_iter, build_outputs, output_path):
    """
    Create an archive of the build outputs. Supports zip and compressed tar archives.

    file_iter is an iterable of files to include in the zip archive.
    output_path is the pathlib.Path to write the new zip archive.
    build_outputs is a pathlib.Path to the build outputs
    """
    output_archive, add_func, archive_root = _get_archive_writer(output_path)
    with output_archive:
        for relative_path in file_iter:
            add_func(build_outputs / relative_path, archive_root / relative_path)
        for include_path in include_iter:
            add_func(include_path, archive_root / include_path.name)


def _files_generator_by_args(args):
    """Returns a files_generator() instance from the CLI args"""
    # --build-outputs
    if not args.build_outputs.exists():
        get_logger().error('Could not find build outputs: %s', args.build_outputs)
        raise FileNotFoundError(args.build_outputs)

    # --cfg
    if not args.cfg.exists():
        get_logger().error('Could not find FILES.cfg at %s', args.cfg)
        raise FileNotFoundError(args.cfg)

    return filescfg_generator(args.cfg, args.build_outputs, args.cpu_arch)


def _list_callback(args):
    """List files needed to run Chromium."""
    sys.stdout.writelines('%s\n' % x for x in _files_generator_by_args(args))


def _archive_callback(args):
    """
    Create an archive of the build outputs. Supports zip and compressed tar archives.
    """
    create_archive(filescfg_generator(args.cfg, args.build_outputs, args.cpu_arch), args.include,
                   args.build_outputs, args.output)


def main():
    """CLI Entrypoint"""
    parser = argparse.ArgumentParser()
    parser.add_argument('-c',
                        '--cfg',
                        metavar='PATH',
                        type=Path,
                        required=True,
                        help=('The FILES.cfg to use. They are usually located under a '
                              'directory in chrome/tools/build/ of the source tree.'))
    parser.add_argument('--build-outputs',
                        metavar='PATH',
                        type=Path,
                        default='out/Default',
                        help=('The path to the build outputs directory relative to the '
                              'source tree. Default: %(default)s'))
    parser.add_argument('--cpu-arch',
                        metavar='ARCH',
                        default=platform.architecture()[0],
                        choices=('64bit', '32bit'),
                        help=('Filter build outputs by a target CPU. '
                              'This is the same as the "arch" key in FILES.cfg. '
                              'Default (from platform.architecture()): %(default)s'))
    add_common_params(parser)

    subparsers = parser.add_subparsers(title='filescfg actions')

    # list
    list_parser = subparsers.add_parser('list', help=_list_callback.__doc__)
    list_parser.set_defaults(callback=_list_callback)

    # archive
    archive_parser = subparsers.add_parser('archive', help=_archive_callback.__doc__)
    archive_parser.add_argument(
        '-o',
        '--output',
        type=Path,
        metavar='PATH',
        required=True,
        help=('The output path for the archive. The type of archive is selected'
              ' by the file extension. Currently supported types: .zip and'
              ' .tar.{gz,bz2,xz}'))
    archive_parser.add_argument(
        '-i',
        '--include',
        type=Path,
        metavar='PATH',
        action='append',
        default=[],
        help=('File or directory to include in the root of the archive. Specify '
              'multiple times to include multiple different items. '
              'For zip files, these contents must only be regular files.'))
    archive_parser.set_defaults(callback=_archive_callback)

    args = parser.parse_args()
    args.callback(args)


if __name__ == '__main__':
    main()



File: utils/prune_binaries.py

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-

# Copyright (c) 2019 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Prune binaries from the source tree"""

import argparse
import sys
import os
import stat
from pathlib import Path

from _common import ENCODING, get_logger, add_common_params

# List of paths to prune if they exist, excluded from domain_substitution and pruning lists
# These allow the lists to be compatible between cloned and tarball sources
CONTINGENT_PATHS = (
    # Overridable git sources
    'third_party/angle/third_party/VK-GL-CTS/src/',
    'third_party/instrumented_libs/',
    # CIPD sources
    'buildtools/linux64/',
    'buildtools/reclient/',
    'third_party/apache-linux/',
    'third_party/checkstyle/',
    'third_party/google-java-format/',
    'third_party/libei/',
    'third_party/ninja/',
    'third_party/screen-ai/',
    'third_party/siso/',
    'third_party/updater/chrome_linux64/',
    'third_party/updater/chromium_linux64/',
    'tools/luci-go/',
    'tools/resultdb/',
    'tools/skia_goldctl/linux/',
    # GCS sources
    'base/tracing/test/data',
    'build/linux/debian_bullseye_amd64-sysroot/',
    'build/linux/debian_bullseye_i386-sysroot/',
    'buildtools/linux64-format/',
    'third_party/blink/renderer/core/css/perftest_data/',
    'third_party/js_code_coverage/',
    'third_party/llvm-build/Release+Asserts/',
    'third_party/node/linux/',
    'third_party/opus/tests/resources/',
    'third_party/rust-toolchain/',
    'third_party/subresource-filter-ruleset/data',
    'third_party/test_fonts/test_fonts',
    'third_party/tfhub_models/testdata/',
    'tools/perf/page_sets/maps_perf_test/dataset/',
)


def prune_files(unpack_root, prune_list):
    """
    Delete files under unpack_root listed in prune_list. Returns an iterable of unremovable files.

    unpack_root is a pathlib.Path to the directory to be pruned
    prune_list is an iterable of files to be removed.
    """
    unremovable_files = set()
    for relative_file in prune_list:
        file_path = unpack_root / relative_file
        try:
            file_path.unlink()
        # read-only files can't be deleted on Windows
        # so remove the flag and try again.
        except PermissionError:
            os.chmod(file_path, stat.S_IWRITE)
            file_path.unlink()
        except FileNotFoundError:
            unremovable_files.add(Path(relative_file).as_posix())
    return unremovable_files


def _prune_path(path):
    """
    Delete all files and directories in path.

    path is a pathlib.Path to the directory to be pruned
    """
    for node in sorted(path.rglob('*'), key=lambda l: len(str(l)), reverse=True):
        if node.is_file() or node.is_symlink():
            try:
                node.unlink()
            except PermissionError:
                node.chmod(stat.S_IWRITE)
                node.unlink()
        elif node.is_dir() and not any(node.iterdir()):
            try:
                node.rmdir()
            except PermissionError:
                node.chmod(stat.S_IWRITE)
                node.rmdir()


def prune_dirs(unpack_root, keep_contingent_paths, sysroot):
    """
    Delete all files and directories in pycache and CONTINGENT_PATHS directories.

    unpack_root is a pathlib.Path to the source tree
    keep_contingent_paths is a boolean that determines if the contingent paths should be pruned
    sysroot is a string that optionally defines a sysroot to exempt from pruning
    """
    for pycache in unpack_root.rglob('__pycache__'):
        _prune_path(pycache)
    if keep_contingent_paths:
        get_logger().info('Keeping Contingent Paths')
    else:
        get_logger().info('Removing Contingent Paths')
        for cpath in CONTINGENT_PATHS:
            if sysroot and f'{sysroot}-sysroot' in cpath:
                get_logger().info('%s: %s', 'Exempt', cpath)
                continue
            get_logger().info('%s: %s', 'Exists' if Path(cpath).exists() else 'Absent', cpath)
            _prune_path(unpack_root / cpath)


def _callback(args):
    if not args.directory.exists():
        get_logger().error('Specified directory does not exist: %s', args.directory)
        sys.exit(1)
    if not args.pruning_list.exists():
        get_logger().error('Could not find the pruning list: %s', args.pruning_list)
    prune_dirs(args.directory, args.keep_contingent_paths, args.sysroot)
    prune_list = tuple(filter(len, args.pruning_list.read_text(encoding=ENCODING).splitlines()))
    unremovable_files = prune_files(args.directory, prune_list)
    if unremovable_files:
        get_logger().error('%d files could not be pruned.', len(unremovable_files))
        get_logger().debug('Files could not be pruned:\n%s',
                           '\n'.join(f for f in unremovable_files))
        sys.exit(1)


def main():
    """CLI Entrypoint"""
    parser = argparse.ArgumentParser()
    parser.add_argument('directory', type=Path, help='The directory to apply binary pruning.')
    parser.add_argument('pruning_list', type=Path, help='Path to pruning.list')
    parser.add_argument('--keep-contingent-paths',
                        action='store_true',
                        help=('Skip pruning the contingent paths. '
                              'Useful when building with the Google tooling is desired.'))
    parser.add_argument('--sysroot',
                        choices=('amd64', 'i386'),
                        help=('Skip pruning the sysroot for the specified architecture. '
                              'Not needed when --keep-contingent-paths is used.'))
    add_common_params(parser)
    parser.set_defaults(callback=_callback)

    args = parser.parse_args()
    args.callback(args)


if __name__ == '__main__':
    main()



File: utils/_common.py

# -*- coding: UTF-8 -*-

# Copyright (c) 2020 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Common code and constants"""
import argparse
import enum
import logging
import platform
from pathlib import Path

# Constants

ENCODING = 'UTF-8' # For config files and patches

USE_REGISTRY = '_use_registry'

LOGGER_NAME = 'ungoogled'

# Public classes


class PlatformEnum(enum.Enum):
    """Enum for platforms that need distinction for certain functionality"""
    UNIX = 'unix' # Currently covers anything that isn't Windows
    WINDOWS = 'windows'


class ExtractorEnum: #pylint: disable=too-few-public-methods
    """Enum for extraction binaries"""
    SEVENZIP = '7z'
    TAR = 'tar'
    WINRAR = 'winrar'


class SetLogLevel(argparse.Action): #pylint: disable=too-few-public-methods
    """Sets logging level based on command line arguments it receives"""
    def __init__(self, option_strings, dest, nargs=None, **kwargs):
        super().__init__(option_strings, dest, nargs=nargs, **kwargs)

    def __call__(self, parser, namespace, value, option_string=None):
        if option_string in ('--verbose', '-v'):
            value = logging.DEBUG
        elif option_string in ('--quiet', '-q'):
            value = logging.ERROR
        else:
            levels = {
                'FATAL': logging.FATAL,
                'ERROR': logging.ERROR,
                'WARNING': logging.WARNING,
                'INFO': logging.INFO,
                'DEBUG': logging.DEBUG
            }
            value = levels[value]
        set_logging_level(value)


# Public methods


def get_logger(initial_level=logging.INFO):
    """Gets the named logger"""

    logger = logging.getLogger(LOGGER_NAME)

    if logger.level == logging.NOTSET:
        logger.setLevel(initial_level)

        if not logger.hasHandlers():
            console_handler = logging.StreamHandler()
            console_handler.setLevel(initial_level)

            format_string = '%(levelname)s: %(message)s'
            formatter = logging.Formatter(format_string)
            console_handler.setFormatter(formatter)

            logger.addHandler(console_handler)
    return logger


def set_logging_level(logging_level):
    """Sets logging level of logger and all its handlers"""

    if not logging_level:
        logging_level = logging.INFO

    logger = get_logger()
    logger.setLevel(logging_level)

    if logger.hasHandlers():
        for hdlr in logger.handlers:
            hdlr.setLevel(logging_level)

    return logger


def get_running_platform():
    """
    Returns a PlatformEnum value indicating the platform that utils is running on.

    NOTE: Platform detection should only be used when no cross-platform alternative is available.
    """
    uname = platform.uname()
    # detect native python and WSL
    if uname.system == 'Windows' or 'Microsoft' in uname.release:
        return PlatformEnum.WINDOWS
    # Only Windows and UNIX-based platforms need to be distinguished right now.
    return PlatformEnum.UNIX


def get_chromium_version():
    """Returns the Chromium version."""
    return (Path(__file__).parent.parent / 'chromium_version.txt').read_text().strip()


def parse_series(series_path):
    """
    Returns an iterator of paths over the series file

    series_path is a pathlib.Path to the series file
    """
    with series_path.open(encoding=ENCODING) as series_file:
        series_lines = series_file.read().splitlines()
    # Filter blank lines
    series_lines = filter(len, series_lines)
    # Filter comment lines
    series_lines = filter((lambda x: not x.startswith('#')), series_lines)
    # Strip in-line comments
    series_lines = map((lambda x: x.strip().split(' #')[0]), series_lines)
    return series_lines


def add_common_params(parser):
    """
    Adds common command line arguments to a parser.
    """

    # Logging levels
    logging_group = parser.add_mutually_exclusive_group()
    logging_group.add_argument(
        '--log-level',
        action=SetLogLevel,
        choices=['FATAL', 'ERROR', 'WARNING', 'INFO', 'DEBUG'],
        help="Set logging level of current script. Only one of 'log-level', 'verbose',"
        " 'quiet' can be set at a time.")
    logging_group.add_argument(
        '--quiet',
        '-q',
        action=SetLogLevel,
        nargs=0,
        help="Display less outputs to console. Only one of 'log-level', 'verbose',"
        " 'quiet' can be set at a time.")
    logging_group.add_argument(
        '--verbose',
        '-v',
        action=SetLogLevel,
        nargs=0,
        help="Increase logging verbosity to include DEBUG messages. Only one of "
        "'log-level', 'verbose', 'quiet' can be set at a time.")



File: utils/__init__.py




File: utils/downloads.py

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-

# Copyright (c) 2019 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""
Module for the downloading, checking, and unpacking of necessary files into the source tree.
"""

import argparse
import configparser
import enum
import hashlib
import shutil
import ssl
import subprocess
import sys
import urllib.request
from pathlib import Path

from _common import ENCODING, USE_REGISTRY, ExtractorEnum, get_logger, \
    get_chromium_version, add_common_params
from _extraction import extract_tar_file, extract_with_7z, extract_with_winrar

sys.path.insert(0, str(Path(__file__).parent / 'third_party'))
import schema #pylint: disable=wrong-import-position, wrong-import-order
sys.path.pop(0)

# Constants


class HashesURLEnum(str, enum.Enum):
    """Enum for supported hash URL schemes"""
    CHROMIUM = 'chromium'


class HashMismatchError(BaseException):
    """Exception for computed hashes not matching expected hashes"""


class DownloadInfo: #pylint: disable=too-few-public-methods
    """Representation of an downloads.ini file for downloading files"""

    _hashes = ('md5', 'sha1', 'sha256', 'sha512')
    hash_url_delimiter = '|'
    _nonempty_keys = ('url', 'download_filename')
    _optional_keys = (
        'version',
        'strip_leading_dirs',
    )
    _passthrough_properties = (*_nonempty_keys, *_optional_keys, 'extractor', 'output_path')
    _ini_vars = {
        '_chromium_version': get_chromium_version(),
    }

    @staticmethod
    def _is_hash_url(value):
        return value.count(DownloadInfo.hash_url_delimiter) == 2 and value.split(
            DownloadInfo.hash_url_delimiter)[0] in iter(HashesURLEnum)

    _schema = schema.Schema({
        schema.Optional(schema.And(str, len)): {
            **{x: schema.And(str, len)
               for x in _nonempty_keys},
            'output_path': (lambda x: str(Path(x).relative_to(''))),
            **{schema.Optional(x): schema.And(str, len)
               for x in _optional_keys},
            schema.Optional('extractor'): schema.Or(ExtractorEnum.TAR, ExtractorEnum.SEVENZIP,
                                                    ExtractorEnum.WINRAR),
            schema.Optional(schema.Or(*_hashes)): schema.And(str, len),
            schema.Optional('hash_url'): lambda x: DownloadInfo._is_hash_url(x), #pylint: disable=unnecessary-lambda
        }
    })

    class _DownloadsProperties: #pylint: disable=too-few-public-methods
        def __init__(self, section_dict, passthrough_properties, hashes):
            self._section_dict = section_dict
            self._passthrough_properties = passthrough_properties
            self._hashes = hashes

        def has_hash_url(self):
            """
            Returns a boolean indicating whether the current
            download has a hash URL"""
            return 'hash_url' in self._section_dict

        def __getattr__(self, name):
            if name in self._passthrough_properties:
                return self._section_dict.get(name, fallback=None)
            if name == 'hashes':
                hashes_dict = {}
                for hash_name in (*self._hashes, 'hash_url'):
                    value = self._section_dict.get(hash_name, fallback=None)
                    if value:
                        if hash_name == 'hash_url':
                            value = value.split(DownloadInfo.hash_url_delimiter)
                        hashes_dict[hash_name] = value
                return hashes_dict
            raise AttributeError('"{}" has no attribute "{}"'.format(type(self).__name__, name))

    def _parse_data(self, path):
        """
        Parses an INI file located at path

        Raises schema.SchemaError if validation fails
        """
        def _section_generator(data):
            for section in data:
                if section == configparser.DEFAULTSECT:
                    continue
                yield section, dict(
                    filter(lambda x: x[0] not in self._ini_vars, data.items(section)))

        new_data = configparser.ConfigParser(defaults=self._ini_vars)
        with path.open(encoding=ENCODING) as ini_file:
            new_data.read_file(ini_file, source=str(path))
        try:
            self._schema.validate(dict(_section_generator(new_data)))
        except schema.SchemaError as exc:
            get_logger().error('downloads.ini failed schema validation (located in %s)', path)
            raise exc
        return new_data

    def __init__(self, ini_paths):
        """Reads an iterable of pathlib.Path to download.ini files"""
        self._data = configparser.ConfigParser()
        for path in ini_paths:
            self._data.read_dict(self._parse_data(path))

    def __getitem__(self, section):
        """
        Returns an object with keys as attributes and
        values already pre-processed strings
        """
        return self._DownloadsProperties(self._data[section], self._passthrough_properties,
                                         self._hashes)

    def __contains__(self, item):
        """
        Returns True if item is a name of a section; False otherwise.
        """
        return self._data.has_section(item)

    def __iter__(self):
        """Returns an iterator over the section names"""
        return iter(self._data.sections())

    def properties_iter(self):
        """Iterator for the download properties sorted by output path"""
        return sorted(map(lambda x: (x, self[x]), self),
                      key=(lambda x: str(Path(x[1].output_path))))


class _UrlRetrieveReportHook: #pylint: disable=too-few-public-methods
    """Hook for urllib.request.urlretrieve to log progress information to console"""
    def __init__(self):
        self._max_len_printed = 0
        self._last_percentage = None

    def __call__(self, block_count, block_size, total_size):
        # Use total_blocks to handle case total_size < block_size
        # total_blocks is ceiling of total_size / block_size
        # Ceiling division from: https://stackoverflow.com/a/17511341
        total_blocks = -(-total_size // block_size)
        if total_blocks > 0:
            # Do not needlessly update the console. Since the console is
            # updated synchronously, we don't want updating the console to
            # bottleneck downloading. Thus, only refresh the output when the
            # displayed value should change.
            percentage = round(block_count / total_blocks, ndigits=3)
            if percentage == self._last_percentage:
                return
            self._last_percentage = percentage
            print('\r' + ' ' * self._max_len_printed, end='')
            status_line = 'Progress: {:.1%} of {:,d} B'.format(percentage, total_size)
        else:
            downloaded_estimate = block_count * block_size
            status_line = 'Progress: {:,d} B of unknown size'.format(downloaded_estimate)
        self._max_len_printed = len(status_line)
        print('\r' + status_line, end='')


def _download_via_urllib(url, file_path, show_progress, disable_ssl_verification):
    reporthook = None
    if show_progress:
        reporthook = _UrlRetrieveReportHook()
    if disable_ssl_verification:
        # TODO: Remove this or properly implement disabling SSL certificate verification
        orig_https_context = ssl._create_default_https_context #pylint: disable=protected-access
        ssl._create_default_https_context = ssl._create_unverified_context #pylint: disable=protected-access
    try:
        urllib.request.urlretrieve(url, str(file_path), reporthook=reporthook)
    finally:
        # Try to reduce damage of hack by reverting original HTTPS context ASAP
        if disable_ssl_verification:
            ssl._create_default_https_context = orig_https_context #pylint: disable=protected-access
    if show_progress:
        print()


def _download_if_needed(file_path, url, show_progress, disable_ssl_verification):
    """
    Downloads a file from url to the specified path file_path if necessary.

    If show_progress is True, download progress is printed to the console.
    """
    if file_path.exists():
        get_logger().info('%s already exists. Skipping download.', file_path)
        return

    # File name for partially download file
    tmp_file_path = file_path.with_name(file_path.name + '.partial')

    if tmp_file_path.exists():
        get_logger().debug('Resuming downloading URL %s ...', url)
    else:
        get_logger().debug('Downloading URL %s ...', url)

    # Perform download
    if shutil.which('curl'):
        get_logger().debug('Using curl')
        try:
            subprocess.run(['curl', '-fL', '-o', str(tmp_file_path), '-C', '-', url], check=True)
        except subprocess.CalledProcessError as exc:
            get_logger().error('curl failed. Re-run the download command to resume downloading.')
            raise exc
    else:
        get_logger().debug('Using urllib')
        _download_via_urllib(url, tmp_file_path, show_progress, disable_ssl_verification)

    # Download complete; rename file
    tmp_file_path.rename(file_path)


def _chromium_hashes_generator(hashes_path):
    with hashes_path.open(encoding=ENCODING) as hashes_file:
        hash_lines = hashes_file.read().splitlines()
    for hash_name, hash_hex, _ in map(lambda x: x.lower().split('  '), hash_lines):
        if hash_name in hashlib.algorithms_available:
            yield hash_name, hash_hex
        else:
            get_logger().warning('Skipping unknown hash algorithm: %s', hash_name)


def _get_hash_pairs(download_properties, cache_dir):
    """Generator of (hash_name, hash_hex) for the given download"""
    for entry_type, entry_value in download_properties.hashes.items():
        if entry_type == 'hash_url':
            hash_processor, hash_filename, _ = entry_value
            if hash_processor == 'chromium':
                yield from _chromium_hashes_generator(cache_dir / hash_filename)
            else:
                raise ValueError('Unknown hash_url processor: %s' % hash_processor)
        else:
            yield entry_type, entry_value


def retrieve_downloads(download_info, cache_dir, show_progress, disable_ssl_verification=False):
    """
    Retrieve downloads into the downloads cache.

    download_info is the DowloadInfo of downloads to retrieve.
    cache_dir is the pathlib.Path to the downloads cache.
    show_progress is a boolean indicating if download progress is printed to the console.
    disable_ssl_verification is a boolean indicating if certificate verification
        should be disabled for downloads using HTTPS.

    Raises FileNotFoundError if the downloads path does not exist.
    Raises NotADirectoryError if the downloads path is not a directory.
    """
    if not cache_dir.exists():
        raise FileNotFoundError(cache_dir)
    if not cache_dir.is_dir():
        raise NotADirectoryError(cache_dir)
    for download_name, download_properties in download_info.properties_iter():
        get_logger().info('Downloading "%s" to "%s" ...', download_name,
                          download_properties.download_filename)
        download_path = cache_dir / download_properties.download_filename
        _download_if_needed(download_path, download_properties.url, show_progress,
                            disable_ssl_verification)
        if download_properties.has_hash_url():
            get_logger().info('Downloading hashes for "%s"', download_name)
            _, hash_filename, hash_url = download_properties.hashes['hash_url']
            _download_if_needed(cache_dir / hash_filename, hash_url, show_progress,
                                disable_ssl_verification)


def check_downloads(download_info, cache_dir):
    """
    Check integrity of the downloads cache.

    download_info is the DownloadInfo of downloads to unpack.
    cache_dir is the pathlib.Path to the downloads cache.

    Raises source_retrieval.HashMismatchError when the computed and expected hashes do not match.
    """
    for download_name, download_properties in download_info.properties_iter():
        get_logger().info('Verifying hashes for "%s" ...', download_name)
        download_path = cache_dir / download_properties.download_filename
        with download_path.open('rb') as file_obj:
            archive_data = file_obj.read()
        for hash_name, hash_hex in _get_hash_pairs(download_properties, cache_dir):
            get_logger().debug('Verifying %s hash...', hash_name)
            hasher = hashlib.new(hash_name, data=archive_data)
            if not hasher.hexdigest().lower() == hash_hex.lower():
                raise HashMismatchError(download_path)


def unpack_downloads(download_info, cache_dir, output_dir, skip_unused, sysroot, extractors=None):
    """
    Unpack downloads in the downloads cache to output_dir. Assumes all downloads are retrieved.

    download_info is the DownloadInfo of downloads to unpack.
    cache_dir is the pathlib.Path directory containing the download cache
    output_dir is the pathlib.Path directory to unpack the downloads to.
    skip_unused is a boolean that determines if unused paths should be extracted.
    sysroot is a string containing a sysroot to unpack if any.
    extractors is a dictionary of PlatformEnum to a command or path to the
        extractor binary. Defaults to 'tar' for tar, and '_use_registry' for 7-Zip and WinRAR.

    May raise undetermined exceptions during archive unpacking.
    """
    for download_name, download_properties in download_info.properties_iter():
        download_path = cache_dir / download_properties.download_filename
        get_logger().info('Unpacking "%s" to %s ...', download_name,
                          download_properties.output_path)
        extractor_name = download_properties.extractor or ExtractorEnum.TAR
        if extractor_name == ExtractorEnum.SEVENZIP:
            extractor_func = extract_with_7z
        elif extractor_name == ExtractorEnum.WINRAR:
            extractor_func = extract_with_winrar
        elif extractor_name == ExtractorEnum.TAR:
            extractor_func = extract_tar_file
        else:
            raise NotImplementedError(extractor_name)

        if download_properties.strip_leading_dirs is None:
            strip_leading_dirs_path = None
        else:
            strip_leading_dirs_path = Path(download_properties.strip_leading_dirs)

        extractor_func(archive_path=download_path,
                       output_dir=output_dir / Path(download_properties.output_path),
                       relative_to=strip_leading_dirs_path,
                       skip_unused=skip_unused,
                       sysroot=sysroot,
                       extractors=extractors)


def _add_common_args(parser):
    parser.add_argument(
        '-i',
        '--ini',
        type=Path,
        nargs='+',
        help='The downloads INI to parse for downloads. Can be specified multiple times.')
    parser.add_argument('-c',
                        '--cache',
                        type=Path,
                        required=True,
                        help='Path to the directory to cache downloads.')


def _retrieve_callback(args):
    retrieve_downloads(DownloadInfo(args.ini), args.cache, args.show_progress,
                       args.disable_ssl_verification)
    try:
        check_downloads(DownloadInfo(args.ini), args.cache)
    except HashMismatchError as exc:
        get_logger().error('File checksum does not match: %s', exc)
        sys.exit(1)


def _unpack_callback(args):
    extractors = {
        ExtractorEnum.SEVENZIP: args.sevenz_path,
        ExtractorEnum.WINRAR: args.winrar_path,
        ExtractorEnum.TAR: args.tar_path,
    }
    unpack_downloads(DownloadInfo(args.ini), args.cache, args.output, args.skip_unused,
                     args.sysroot, extractors)


def main():
    """CLI Entrypoint"""
    parser = argparse.ArgumentParser(description=__doc__)
    add_common_params(parser)
    subparsers = parser.add_subparsers(title='Download actions', dest='action')

    # retrieve
    retrieve_parser = subparsers.add_parser(
        'retrieve',
        help='Retrieve and check download files',
        description=('Retrieves and checks downloads without unpacking. '
                     'The downloader will attempt to use CLI command "curl". '
                     'If it is not present, Python\'s urllib will be used. However, only '
                     'the CLI-based downloaders can be resumed if the download is aborted.'))
    _add_common_args(retrieve_parser)
    retrieve_parser.add_argument('--hide-progress-bar',
                                 action='store_false',
                                 dest='show_progress',
                                 help='Hide the download progress.')
    retrieve_parser.add_argument(
        '--disable-ssl-verification',
        action='store_true',
        help='Disables certification verification for downloads using HTTPS.')
    retrieve_parser.set_defaults(callback=_retrieve_callback)

    # unpack
    unpack_parser = subparsers.add_parser(
        'unpack',
        help='Unpack download files',
        description='Verifies hashes of and unpacks download files into the specified directory.')
    _add_common_args(unpack_parser)
    unpack_parser.add_argument('--tar-path',
                               default='tar',
                               help=('(Linux and macOS only) Command or path to the BSD or GNU tar '
                                     'binary for extraction. Default: %(default)s'))
    unpack_parser.add_argument(
        '--7z-path',
        dest='sevenz_path',
        default=USE_REGISTRY,
        help=('Command or path to 7-Zip\'s "7z" binary. If "_use_registry" is '
              'specified, determine the path from the registry. Default: %(default)s'))
    unpack_parser.add_argument(
        '--winrar-path',
        dest='winrar_path',
        default=USE_REGISTRY,
        help=('Command or path to WinRAR\'s "winrar" binary. If "_use_registry" is '
              'specified, determine the path from the registry. Default: %(default)s'))
    unpack_parser.add_argument('output', type=Path, help='The directory to unpack to.')
    unpack_parser.add_argument('--skip-unused',
                               action='store_true',
                               help='Skip extraction of unused directories (CONTINGENT_PATHS).')
    unpack_parser.add_argument('--sysroot',
                               choices=('amd64', 'i386'),
                               help=('Extracts the sysroot for the given architecture '
                                     'when --skip-unused is set.'))
    unpack_parser.set_defaults(callback=_unpack_callback)

    args = parser.parse_args()
    args.callback(args)


if __name__ == '__main__':
    main()



File: utils/domain_substitution.py

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-

# Copyright (c) 2019 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""
Substitute domain names in the source tree with blockable strings.
"""

from pathlib import Path
import argparse
import collections
import contextlib
import io
import os
import stat
import re
import tarfile
import tempfile
import zlib

from _extraction import extract_tar_file
from _common import ENCODING, get_logger, add_common_params

# Encodings to try on source tree files
TREE_ENCODINGS = ('UTF-8', 'ISO-8859-1')

# Constants for domain substitution cache
_INDEX_LIST = 'cache_index.list'
_INDEX_HASH_DELIMITER = '|'
_ORIG_DIR = 'orig'

# Constants for timestamp manipulation
# Delta between all file timestamps in nanoseconds
_TIMESTAMP_DELTA = 1 * 10**9


class DomainRegexList:
    """Representation of a domain_regex.list file"""
    _regex_pair_tuple = collections.namedtuple('DomainRegexPair', ('pattern', 'replacement'))

    # Constants for format:
    _PATTERN_REPLACE_DELIM = '#'

    def __init__(self, path):
        self._data = tuple(filter(len, path.read_text().splitlines()))

        # Cache of compiled regex pairs
        self._compiled_regex = None

    def _compile_regex(self, line):
        """Generates a regex pair tuple for the given line"""
        pattern, replacement = line.split(self._PATTERN_REPLACE_DELIM)
        return self._regex_pair_tuple(re.compile(pattern), replacement)

    @property
    def regex_pairs(self):
        """
        Returns a tuple of compiled regex pairs
        """
        if not self._compiled_regex:
            self._compiled_regex = tuple(map(self._compile_regex, self._data))
        return self._compiled_regex

    @property
    def search_regex(self):
        """
        Returns a single expression to search for domains
        """
        return re.compile('|'.join(
            map(lambda x: x.split(self._PATTERN_REPLACE_DELIM, 1)[0], self._data)))


# Private Methods


def _substitute_path(path, regex_iter):
    """
    Perform domain substitution on path and add it to the domain substitution cache.

    path is a pathlib.Path to the file to be domain substituted.
    regex_iter is an iterable of regular expression namedtuple like from
        config.DomainRegexList.regex_pairs()

    Returns a tuple of the CRC32 hash of the substituted raw content and the
        original raw content; None for both entries if no substitutions were made.

    Raises FileNotFoundError if path does not exist.
    Raises UnicodeDecodeError if path's contents cannot be decoded.
    """
    if not os.access(path, os.W_OK):
        # If the patch cannot be written to, it cannot be opened for updating
        print(str(path) + " cannot be opened for writing! Adding write permission...")
        path.chmod(path.stat().st_mode | stat.S_IWUSR)
    with path.open('r+b') as input_file:
        original_content = input_file.read()
        if not original_content:
            return (None, None)
        content = None
        encoding = None
        for encoding in TREE_ENCODINGS:
            try:
                content = original_content.decode(encoding)
                break
            except UnicodeDecodeError:
                continue
        if not content:
            raise UnicodeDecodeError('Unable to decode with any encoding: %s' % path)
        file_subs = 0
        for regex_pair in regex_iter:
            content, sub_count = regex_pair.pattern.subn(regex_pair.replacement, content)
            file_subs += sub_count
        if file_subs > 0:
            substituted_content = content.encode(encoding)
            input_file.seek(0)
            input_file.write(content.encode(encoding))
            input_file.truncate()
            return (zlib.crc32(substituted_content), original_content)
        return (None, None)


def _validate_file_index(index_file, resolved_tree, cache_index_files):
    """
    Validation of file index and hashes against the source tree.
        Updates cache_index_files

    Returns True if the file index is valid; False otherwise
    """
    all_hashes_valid = True
    crc32_regex = re.compile(r'^[a-zA-Z0-9]{8}$')
    for entry in index_file.read().decode(ENCODING).splitlines():
        try:
            relative_path, file_hash = entry.split(_INDEX_HASH_DELIMITER)
        except ValueError as exc:
            get_logger().error('Could not split entry "%s": %s', entry, exc)
            continue
        if not relative_path or not file_hash:
            get_logger().error('Entry %s of domain substitution cache file index is not valid',
                               _INDEX_HASH_DELIMITER.join((relative_path, file_hash)))
            all_hashes_valid = False
            continue
        if not crc32_regex.match(file_hash):
            get_logger().error('File index hash for %s does not appear to be a CRC32 hash',
                               relative_path)
            all_hashes_valid = False
            continue
        if zlib.crc32((resolved_tree / relative_path).read_bytes()) != int(file_hash, 16):
            get_logger().error('Hashes do not match for: %s', relative_path)
            all_hashes_valid = False
            continue
        if relative_path in cache_index_files:
            get_logger().error('File %s shows up at least twice in the file index', relative_path)
            all_hashes_valid = False
            continue
        cache_index_files.add(relative_path)
    return all_hashes_valid


@contextlib.contextmanager
def _update_timestamp(path: os.PathLike, set_new: bool) -> None:
    """
    Context manager to set the timestamp of the path to plus or
    minus a fixed delta, regardless of modifications within the context.

    if set_new is True, the delta is added. Otherwise, the delta is subtracted.
    """
    stats = os.stat(path)
    if set_new:
        new_timestamp = (stats.st_atime_ns + _TIMESTAMP_DELTA, stats.st_mtime_ns + _TIMESTAMP_DELTA)
    else:
        new_timestamp = (stats.st_atime_ns - _TIMESTAMP_DELTA, stats.st_mtime_ns - _TIMESTAMP_DELTA)
    try:
        yield
    finally:
        os.utime(path, ns=new_timestamp)


# Public Methods


def apply_substitution(regex_path, files_path, source_tree, domainsub_cache):
    """
    Substitute domains in source_tree with files and substitutions,
        and save the pre-domain substitution archive to presubdom_archive.

    regex_path is a pathlib.Path to domain_regex.list
    files_path is a pathlib.Path to domain_substitution.list
    source_tree is a pathlib.Path to the source tree.
    domainsub_cache is a pathlib.Path to the domain substitution cache.

    Raises NotADirectoryError if the patches directory is not a directory or does not exist
    Raises FileNotFoundError if the source tree or required directory does not exist.
    Raises FileExistsError if the domain substitution cache already exists.
    Raises ValueError if an entry in the domain substitution list contains the file index
        hash delimiter.
    """
    if not source_tree.exists():
        raise FileNotFoundError(source_tree)
    if not regex_path.exists():
        raise FileNotFoundError(regex_path)
    if not files_path.exists():
        raise FileNotFoundError(files_path)
    if domainsub_cache and domainsub_cache.exists():
        raise FileExistsError(domainsub_cache)
    resolved_tree = source_tree.resolve()
    regex_pairs = DomainRegexList(regex_path).regex_pairs
    fileindex_content = io.BytesIO()
    with tarfile.open(str(domainsub_cache), 'w:%s' % domainsub_cache.suffix[1:],
                      compresslevel=1) if domainsub_cache else open(os.devnull, 'w') as cache_tar:
        for relative_path in filter(len, files_path.read_text().splitlines()):
            if _INDEX_HASH_DELIMITER in relative_path:
                if domainsub_cache:
                    # Cache tar will be incomplete; remove it for convenience
                    cache_tar.close()
                    domainsub_cache.unlink()
                raise ValueError(
                    'Path "%s" contains the file index hash delimiter "%s"' % relative_path,
                    _INDEX_HASH_DELIMITER)
            path = resolved_tree / relative_path
            if not path.exists():
                get_logger().warning('Skipping non-existant path: %s', path)
                continue
            if path.is_symlink():
                get_logger().warning('Skipping path that has become a symlink: %s', path)
                continue
            with _update_timestamp(path, set_new=True):
                crc32_hash, orig_content = _substitute_path(path, regex_pairs)
            if crc32_hash is None:
                get_logger().info('Path has no substitutions: %s', relative_path)
                continue
            if domainsub_cache:
                fileindex_content.write('{}{}{:08x}\n'.format(relative_path, _INDEX_HASH_DELIMITER,
                                                              crc32_hash).encode(ENCODING))
                orig_tarinfo = tarfile.TarInfo(str(Path(_ORIG_DIR) / relative_path))
                orig_tarinfo.size = len(orig_content)
                with io.BytesIO(orig_content) as orig_file:
                    cache_tar.addfile(orig_tarinfo, orig_file)
        if domainsub_cache:
            fileindex_tarinfo = tarfile.TarInfo(_INDEX_LIST)
            fileindex_tarinfo.size = fileindex_content.tell()
            fileindex_content.seek(0)
            cache_tar.addfile(fileindex_tarinfo, fileindex_content)


def revert_substitution(domainsub_cache, source_tree):
    """
    Revert domain substitution on source_tree using the pre-domain
        substitution archive presubdom_archive.
    It first checks if the hashes of the substituted files match the hashes
        computed during the creation of the domain substitution cache, raising
        KeyError if there are any mismatches. Then, it proceeds to
        reverting files in the source_tree.
    domainsub_cache is removed only if all the files from the domain substitution cache
        were relocated to the source tree.

    domainsub_cache is a pathlib.Path to the domain substitution cache.
    source_tree is a pathlib.Path to the source tree.

    Raises KeyError if:
        * There is a hash mismatch while validating the cache
        * The cache's file index is corrupt or missing
        * The cache is corrupt or is not consistent with the file index
    Raises FileNotFoundError if the source tree or domain substitution cache do not exist.
    """
    # This implementation trades disk space/wear for performance (unless a ramdisk is used
    #   for the source tree)
    # Assumptions made for this process:
    # * The correct tar file was provided (so no huge amount of space is wasted)
    # * The tar file is well-behaved (e.g. no files extracted outside of destination path)
    # * Cache file index and cache contents are already consistent (i.e. no files exclusive to
    #   one or the other)
    if not domainsub_cache:
        get_logger().error('Cache file must be specified.')
    if not domainsub_cache.exists():
        raise FileNotFoundError(domainsub_cache)
    if not source_tree.exists():
        raise FileNotFoundError(source_tree)
    resolved_tree = source_tree.resolve()

    cache_index_files = set() # All files in the file index

    with tempfile.TemporaryDirectory(prefix='domsubcache_files',
                                     dir=str(resolved_tree)) as tmp_extract_name:
        extract_path = Path(tmp_extract_name)
        get_logger().debug('Extracting domain substitution cache...')
        extract_tar_file(domainsub_cache, extract_path, None, False, None)

        # Validate source tree file hashes match
        get_logger().debug('Validating substituted files in source tree...')
        with (extract_path / _INDEX_LIST).open('rb') as index_file: #pylint: disable=no-member
            if not _validate_file_index(index_file, resolved_tree, cache_index_files):
                raise KeyError('Domain substitution cache file index is corrupt or hashes mismatch '
                               'the source tree.')

        # Move original files over substituted ones
        get_logger().debug('Moving original files over substituted ones...')
        for relative_path in cache_index_files:
            with _update_timestamp(resolved_tree / relative_path, set_new=False):
                (extract_path / _ORIG_DIR / relative_path).replace(resolved_tree / relative_path)

        # Quick check for unused files in cache
        orig_has_unused = False
        for orig_path in (extract_path / _ORIG_DIR).rglob('*'): #pylint: disable=no-member
            if orig_path.is_file():
                get_logger().warning('Unused file from cache: %s', orig_path)
                orig_has_unused = True

    if orig_has_unused:
        get_logger().warning('Cache contains unused files. Not removing.')
    else:
        domainsub_cache.unlink()


def _callback(args):
    """CLI Callback"""
    if args.reverting:
        revert_substitution(args.cache, args.directory)
    else:
        apply_substitution(args.regex, args.files, args.directory, args.cache)


def main():
    """CLI Entrypoint"""
    parser = argparse.ArgumentParser()
    add_common_params(parser)
    parser.set_defaults(callback=_callback)
    subparsers = parser.add_subparsers(title='', dest='packaging')

    # apply
    apply_parser = subparsers.add_parser(
        'apply',
        help='Apply domain substitution',
        description='Applies domain substitution and creates the domain substitution cache.')
    apply_parser.add_argument('-r',
                              '--regex',
                              type=Path,
                              required=True,
                              help='Path to domain_regex.list')
    apply_parser.add_argument('-f',
                              '--files',
                              type=Path,
                              required=True,
                              help='Path to domain_substitution.list')
    apply_parser.add_argument(
        '-c',
        '--cache',
        type=Path,
        help='The path to the domain substitution cache. The path must not already exist.')
    apply_parser.add_argument('directory',
                              type=Path,
                              help='The directory to apply domain substitution')
    apply_parser.set_defaults(reverting=False)

    # revert
    revert_parser = subparsers.add_parser(
        'revert',
        help='Revert domain substitution',
        description='Reverts domain substitution based only on the domain substitution cache.')
    revert_parser.add_argument('directory',
                               type=Path,
                               help='The directory to reverse domain substitution')
    revert_parser.add_argument('-c',
                               '--cache',
                               type=Path,
                               required=True,
                               help=('The path to the domain substitution cache. '
                                     'The path must exist and will be removed if successful.'))
    revert_parser.set_defaults(reverting=True)

    args = parser.parse_args()
    args.callback(args)


if __name__ == '__main__':
    main()



File: utils/patches.py

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-

# Copyright (c) 2020 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Applies unified diff patches"""

import argparse
import os
import shutil
import subprocess
from pathlib import Path

from _common import get_logger, parse_series, add_common_params


def _find_patch_from_env():
    patch_bin_path = None
    patch_bin_env = os.environ.get('PATCH_BIN')
    if patch_bin_env:
        patch_bin_path = Path(patch_bin_env)
        if patch_bin_path.exists():
            get_logger().debug('Found PATCH_BIN with path "%s"', patch_bin_path)
        else:
            patch_which = shutil.which(patch_bin_env)
            if patch_which:
                get_logger().debug('Found PATCH_BIN for command with path "%s"', patch_which)
                patch_bin_path = Path(patch_which)
    else:
        get_logger().debug('PATCH_BIN env variable is not set')
    return patch_bin_path


def _find_patch_from_which():
    patch_which = shutil.which('patch')
    if not patch_which:
        get_logger().debug('Did not find "patch" in PATH environment variable')
        return None
    return Path(patch_which)


def find_and_check_patch(patch_bin_path=None):
    """
    Find and/or check the patch binary is working. It finds a path to patch in this order:

    1. Use patch_bin_path if it is not None
    2. See if "PATCH_BIN" environment variable is set
    3. Do "which patch" to find GNU patch

    Then it does some sanity checks to see if the patch command is valid.

    Returns the path to the patch binary found.
    """
    if patch_bin_path is None:
        patch_bin_path = _find_patch_from_env()
    if patch_bin_path is None:
        patch_bin_path = _find_patch_from_which()
    if not patch_bin_path:
        raise ValueError('Could not find patch from PATCH_BIN env var or "which patch"')

    if not patch_bin_path.exists():
        raise ValueError('Could not find the patch binary: {}'.format(patch_bin_path))

    # Ensure patch actually runs
    cmd = [str(patch_bin_path), '--version']
    result = subprocess.run(cmd,
                            stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE,
                            check=False,
                            universal_newlines=True)
    if result.returncode:
        get_logger().error('"%s" returned non-zero exit code', ' '.join(cmd))
        get_logger().error('stdout:\n%s', result.stdout)
        get_logger().error('stderr:\n%s', result.stderr)
        raise RuntimeError('Got non-zero exit code running "{}"'.format(' '.join(cmd)))

    return patch_bin_path


def dry_run_check(patch_path, tree_path, patch_bin_path=None):
    """
    Run patch --dry-run on a patch

    tree_path is the pathlib.Path of the source tree to patch
    patch_path is a pathlib.Path to check
    reverse is whether the patches should be reversed
    patch_bin_path is the pathlib.Path of the patch binary, or None to find it automatically
        See find_and_check_patch() for logic to find "patch"

    Returns the status code, stdout, and stderr of patch --dry-run
    """
    cmd = [
        str(find_and_check_patch(patch_bin_path)), '-p1', '--ignore-whitespace', '-i',
        str(patch_path), '-d',
        str(tree_path), '--no-backup-if-mismatch', '--dry-run'
    ]
    result = subprocess.run(cmd,
                            stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE,
                            check=False,
                            universal_newlines=True)
    return result.returncode, result.stdout, result.stderr


def apply_patches(patch_path_iter, tree_path, reverse=False, patch_bin_path=None):
    """
    Applies or reverses a list of patches

    tree_path is the pathlib.Path of the source tree to patch
    patch_path_iter is a list or tuple of pathlib.Path to patch files to apply
    reverse is whether the patches should be reversed
    patch_bin_path is the pathlib.Path of the patch binary, or None to find it automatically
        See find_and_check_patch() for logic to find "patch"

    Raises ValueError if the patch binary could not be found.
    """
    patch_paths = list(patch_path_iter)
    patch_bin_path = find_and_check_patch(patch_bin_path=patch_bin_path)
    if reverse:
        patch_paths.reverse()

    logger = get_logger()
    for patch_path, patch_num in zip(patch_paths, range(1, len(patch_paths) + 1)):
        cmd = [
            str(patch_bin_path), '-p1', '--ignore-whitespace', '-i',
            str(patch_path), '-d',
            str(tree_path), '--no-backup-if-mismatch'
        ]
        if reverse:
            cmd.append('--reverse')
            log_word = 'Reversing'
        else:
            cmd.append('--forward')
            log_word = 'Applying'
        logger.info('* %s %s (%s/%s)', log_word, patch_path.name, patch_num, len(patch_paths))
        logger.debug(' '.join(cmd))
        subprocess.run(cmd, check=True)


def generate_patches_from_series(patches_dir, resolve=False):
    """Generates pathlib.Path for patches from a directory in GNU Quilt format"""
    for patch_path in parse_series(patches_dir / 'series'):
        if resolve:
            yield (patches_dir / patch_path).resolve()
        else:
            yield patch_path


def _copy_files(path_iter, source, destination):
    """Copy files from source to destination with relative paths from path_iter"""
    for path in path_iter:
        (destination / path).parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(str(source / path), str(destination / path))


def merge_patches(source_iter, destination, prepend=False):
    """
    Merges GNU quilt-formatted patches directories from sources into destination

    destination must not already exist, unless prepend is True. If prepend is True, then
    the source patches will be prepended to the destination.
    """
    series = []
    known_paths = set()
    if destination.exists():
        if prepend:
            if not (destination / 'series').exists():
                raise FileNotFoundError(
                    'Could not find series file in existing destination: {}'.format(destination /
                                                                                    'series'))
            known_paths.update(generate_patches_from_series(destination))
        else:
            raise FileExistsError('destination already exists: {}'.format(destination))
    for source_dir in source_iter:
        patch_paths = tuple(generate_patches_from_series(source_dir))
        patch_intersection = known_paths.intersection(patch_paths)
        if patch_intersection:
            raise FileExistsError(
                'Patches from {} have conflicting paths with other sources: {}'.format(
                    source_dir, patch_intersection))
        series.extend(patch_paths)
        _copy_files(patch_paths, source_dir, destination)
    if prepend and (destination / 'series').exists():
        series.extend(generate_patches_from_series(destination))
    with (destination / 'series').open('w') as series_file:
        series_file.write('\n'.join(map(str, series)))


def _apply_callback(args, parser_error):
    logger = get_logger()
    patch_bin_path = None
    if args.patch_bin is not None:
        patch_bin_path = Path(args.patch_bin)
        if not patch_bin_path.exists():
            patch_bin_path = shutil.which(args.patch_bin)
            if patch_bin_path:
                patch_bin_path = Path(patch_bin_path)
            else:
                parser_error(
                    f'--patch-bin "{args.patch_bin}" is not a command or path to executable.')
    for patch_dir in args.patches:
        logger.info('Applying patches from %s', patch_dir)
        apply_patches(generate_patches_from_series(patch_dir, resolve=True),
                      args.target,
                      patch_bin_path=patch_bin_path)


def _merge_callback(args, _):
    merge_patches(args.source, args.destination, args.prepend)


def main():
    """CLI Entrypoint"""
    parser = argparse.ArgumentParser()
    add_common_params(parser)
    subparsers = parser.add_subparsers()

    apply_parser = subparsers.add_parser(
        'apply', help='Applies patches (in GNU Quilt format) to the specified source tree')
    apply_parser.add_argument('--patch-bin',
                              help='The GNU patch command to use. Omit to find it automatically.')
    apply_parser.add_argument('target', type=Path, help='The directory tree to apply patches onto.')
    apply_parser.add_argument(
        'patches',
        type=Path,
        nargs='+',
        help='The directories containing patches to apply. They must be in GNU quilt format')
    apply_parser.set_defaults(callback=_apply_callback)

    merge_parser = subparsers.add_parser('merge',
                                         help='Merges patches directories in GNU quilt format')
    merge_parser.add_argument(
        '--prepend',
        '-p',
        action='store_true',
        help=('If "destination" exists, prepend patches from sources into it.'
              ' By default, merging will fail if the destination already exists.'))
    merge_parser.add_argument(
        'destination',
        type=Path,
        help=('The directory to write the merged patches to. '
              'The destination must not exist unless --prepend is specified.'))
    merge_parser.add_argument('source',
                              type=Path,
                              nargs='+',
                              help='The GNU quilt patches to merge.')
    merge_parser.set_defaults(callback=_merge_callback)

    args = parser.parse_args()
    if 'callback' not in args:
        parser.error('Must specify subcommand apply or merge')
    args.callback(args, parser.error)


if __name__ == '__main__':
    main()



File: utils/make_domsub_script.py

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-

# Copyright (c) 2023 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""
Generate standalone script that performs the domain substitution.
"""

from pathlib import Path
import argparse
import re


def make_domain_substitution_script(regex_path, files_path, output_path):
    """
    Generate a standalone shell script (which uses Perl) that performs
        domain substitution on the appropriate files.

    regex_path is a pathlib.Path to domain_regex.list
    files_path is a pathlib.Path to domain_substitution.list
    output_path is a pathlib.Path to the output file.

    Raises FileNotFoundError if the regex or file lists do not exist.
    Raises FileExistsError if the output file already exists.
    """
    if not regex_path.exists():
        raise FileNotFoundError(regex_path)
    if not files_path.exists():
        raise FileNotFoundError(files_path)
    if output_path.exists():
        raise FileExistsError(output_path)

    regex_list = tuple(filter(len, regex_path.read_text().splitlines()))
    files_list = tuple(filter(len, files_path.read_text().splitlines()))

    # Convert the Python-style regexes into a Perl s/// op
    perl_replace_list = ['s#' + re.sub(r'\\g<(\d+)>', r'${\1}', x) + '#g' for x in regex_list]

    files_list_str = '\n'.join(files_list)
    perl_replace_list_str = '\n'.join([f'    {x};' for x in perl_replace_list])

    with open(output_path, 'w') as out:
        out.write("""#!/bin/sh -e
#
# This script performs domain substitution on the Chromium source files.
#
# Generated by make_domsub_script.py, part of the ungoogled-chromium project:
# https://github.com/ungoogled-software/ungoogled-chromium.git
#

# Check that we are inside the Chromium source tree
test -f build/config/compiler/BUILD.gn

# These filenames may contain spaces and/or other unusual characters
print_file_list() {
	cat <<'__END__'
%s
__END__
}

echo "Creating backup archive ..."

backup=domain-substitution.orig.tar
print_file_list | tar cf $backup --verbatim-files-from --files-from=-

echo "Applying ungoogled-chromium domain substitution to %d files ..."

print_file_list | xargs -d '\\n' perl -0777 -C0 -pwi -e '
%s
'

# end
""" % (files_list_str, len(files_list), perl_replace_list_str))


def _callback(args):
    """CLI Callback"""
    make_domain_substitution_script(args.regex, args.files, args.output)


def main():
    """CLI Entrypoint"""
    parser = argparse.ArgumentParser()
    parser.set_defaults(callback=_callback)

    parser.add_argument('-r', '--regex', type=Path, required=True, help='Path to domain_regex.list')
    parser.add_argument('-f',
                        '--files',
                        type=Path,
                        required=True,
                        help='Path to domain_substitution.list')
    parser.add_argument('-o',
                        '--output',
                        type=Path,
                        required=True,
                        help='Path to script file to create')

    args = parser.parse_args()
    args.callback(args)


if __name__ == '__main__':
    main()



File: utils/clone.py

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-

# Copyright (c) 2023 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""
Module for cloning the source tree.
"""

import re
import sys
from argparse import ArgumentParser
from os import environ, pathsep
from pathlib import Path
from shutil import copytree, copy, move
from stat import S_IWRITE
from subprocess import run

from _common import add_common_params, get_chromium_version, get_logger
from prune_binaries import CONTINGENT_PATHS

# Config file for gclient
# Instances of 'src' replaced with UC_OUT, which will be replaced with the output directory
# custom_deps are set to None since they are large and unused
# target_* arguments set to match tarball rather than actual build target
GC_CONFIG = """\
solutions = [
  {
    "name": "UC_OUT",
    "url": "https://chromium.googlesource.com/chromium/src.git",
    "managed": False,
    "custom_deps": {
      "UC_OUT/third_party/angle/third_party/VK-GL-CTS/src": None,
      "UC_OUT/third_party/instrumented_libs": None,
    },
    "custom_vars": {
      "checkout_configuration": "small",
      "non_git_source": "False",
    },
  },
];
target_os = ['unix'];
target_os_only = True;
target_cpu = ['x64'];
target_cpu_only = True;
"""


def clone(args): # pylint: disable=too-many-branches, too-many-statements
    """Clones, downloads, and generates the required sources"""
    get_logger().info('Setting up cloning environment')
    iswin = sys.platform.startswith('win')
    chromium_version = get_chromium_version()
    ucstaging = args.output / 'uc_staging'
    dtpath = ucstaging / 'depot_tools'
    gnpath = ucstaging / 'gn'
    environ['GCLIENT_FILE'] = str(ucstaging / '.gclient')
    environ['PATH'] += pathsep + str(dtpath)
    environ['PYTHONPATH'] = str(dtpath)
    # Prevent gclient from auto updating depot_tools
    environ['DEPOT_TOOLS_UPDATE'] = '0'
    # Don't generate pycache files
    environ['PYTHONDONTWRITEBYTECODE'] = '1'
    # Allow usage of system python
    environ['VPYTHON_BYPASS'] = 'manually managed python not supported by chrome operations'
    # Google has some regex strings that aren't escaped properly or set as raw
    environ["PYTHONWARNINGS"] = "ignore::SyntaxWarning"

    # depth=2 since generating LASTCHANGE and gpu_lists_version.h require at least two commits
    get_logger().info('Cloning chromium source: %s', chromium_version)
    if (args.output / '.git').exists():
        run(['git', 'fetch', 'origin', 'tag', chromium_version, '--depth=2'],
            cwd=args.output,
            check=True)
        run(['git', 'reset', '--hard', 'FETCH_HEAD'], cwd=args.output, check=True)
        run(['git', 'clean', '-ffdx', '-e', 'uc_staging'], cwd=args.output, check=True)
    else:
        run([
            'git', 'clone', '-c', 'advice.detachedHead=false', '-b', chromium_version, '--depth=2',
            "https://chromium.googlesource.com/chromium/src",
            str(args.output)
        ],
            check=True)

    # Set up staging directory
    ucstaging.mkdir(exist_ok=True)

    get_logger().info('Cloning depot_tools')
    dt_commit = re.search(r"depot_tools\.git'\s*\+\s*'@'\s*\+\s*'([^']+)',",
                          Path(args.output / 'DEPS').read_text()).group(1)
    if not dt_commit:
        get_logger().error('Unable to obtain commit for depot_tools checkout')
        sys.exit(1)
    if not dtpath.exists():
        dtpath.mkdir()
        run(['git', 'init', '-q'], cwd=dtpath, check=True)
        run([
            'git', 'remote', 'add', 'origin',
            'https://chromium.googlesource.com/chromium/tools/depot_tools'
        ],
            cwd=dtpath,
            check=True)
    run(['git', 'fetch', '--depth=1', 'origin', dt_commit], cwd=dtpath, check=True)
    run(['git', 'reset', '--hard', dt_commit], cwd=dtpath, check=True)
    run(['git', 'clean', '-ffdx'], cwd=dtpath, check=True)
    if iswin:
        (dtpath / 'git.bat').write_text('git')
    # Apply changes to gclient
    run(['git', 'apply'],
        input=Path(__file__).with_name('depot_tools.patch').read_text().replace(
            'UC_OUT', str(args.output)).replace('UC_STAGING', str(ucstaging)),
        cwd=dtpath,
        check=True,
        universal_newlines=True)

    # gn requires full history to be able to generate last_commit_position.h
    get_logger().info('Cloning gn')
    if gnpath.exists():
        run(['git', 'fetch'], cwd=gnpath, check=True)
        run(['git', 'reset', '--hard', 'FETCH_HEAD'], cwd=gnpath, check=True)
        run(['git', 'clean', '-ffdx'], cwd=gnpath, check=True)
    else:
        run(['git', 'clone', "https://gn.googlesource.com/gn", str(gnpath)], check=True)

    get_logger().info('Running gsync')
    if args.custom_config:
        copy(args.custom_config, ucstaging / '.gclient').replace('UC_OUT', str(args.output))
    else:
        (ucstaging / '.gclient').write_text(GC_CONFIG.replace('UC_OUT', str(args.output)))
    gcpath = dtpath / 'gclient'
    if iswin:
        gcpath = gcpath.with_suffix('.bat')
    # -f, -D, and -R forces a hard reset on changes and deletes deps that have been removed
    run([
        str(gcpath), 'sync', '-f', '-D', '-R', '--no-history', '--nohooks',
        f'--sysroot={args.sysroot}'
    ],
        check=True)

    # Follow tarball procedure:
    # https://source.chromium.org/chromium/chromium/tools/build/+/main:recipes/recipes/publish_tarball.py
    get_logger().info('Downloading pgo profiles')
    run([
        sys.executable,
        str(args.output / 'tools' / 'update_pgo_profiles.py'), '--target=' + args.pgo, 'update',
        '--gs-url-base=chromium-optimization-profiles/pgo_profiles'
    ],
        check=True)
    # https://chromium-review.googlesource.com/c/chromium/tools/build/+/4380399
    run([
        sys.executable,
        str(args.output / 'v8' / 'tools' / 'builtins-pgo' / 'download_profiles.py'), 'download',
        '--depot-tools',
        str(dtpath)
    ],
        check=True)

    get_logger().info('Generating: DAWN_VERSION')
    run([
        sys.executable,
        str(args.output / 'build' / 'util' / 'lastchange.py'), '-s',
        str(args.output / 'third_party' / 'dawn'), '--revision',
        str(args.output / 'gpu' / 'webgpu' / 'DAWN_VERSION')
    ],
        check=True)

    get_logger().info('Generating: LASTCHANGE')
    run([
        sys.executable,
        str(args.output / 'build' / 'util' / 'lastchange.py'), '-o',
        str(args.output / 'build' / 'util' / 'LASTCHANGE')
    ],
        check=True)

    get_logger().info('Generating: gpu_lists_version.h')
    run([
        sys.executable,
        str(args.output / 'build' / 'util' / 'lastchange.py'), '-m', 'GPU_LISTS_VERSION',
        '--revision-id-only', '--header',
        str(args.output / 'gpu' / 'config' / 'gpu_lists_version.h')
    ],
        check=True)

    get_logger().info('Generating: skia_commit_hash.h')
    run([
        sys.executable,
        str(args.output / 'build' / 'util' / 'lastchange.py'), '-m', 'SKIA_COMMIT_HASH', '-s',
        str(args.output / 'third_party' / 'skia'), '--header',
        str(args.output / 'skia' / 'ext' / 'skia_commit_hash.h')
    ],
        check=True)

    get_logger().info('Generating: last_commit_position.h')
    run([sys.executable, str(gnpath / 'build' / 'gen.py')], check=True)
    for item in gnpath.iterdir():
        if not item.is_dir():
            copy(item, args.output / 'tools' / 'gn')
        elif item.name != '.git' and item.name != 'out':
            copytree(item, args.output / 'tools' / 'gn' / item.name)
    move(str(gnpath / 'out' / 'last_commit_position.h'),
         str(args.output / 'tools' / 'gn' / 'bootstrap'))

    get_logger().info('Removing uneeded files')
    # Match removals for the tarball:
    # https://source.chromium.org/chromium/chromium/tools/build/+/main:recipes/recipe_modules/chromium/resources/export_tarball.py
    remove_dirs = (
        (args.output / 'chrome' / 'test' / 'data'),
        (args.output / 'content' / 'test' / 'data'),
        (args.output / 'courgette' / 'testdata'),
        (args.output / 'extensions' / 'test' / 'data'),
        (args.output / 'media' / 'test' / 'data'),
        (args.output / 'native_client' / 'src' / 'trusted' / 'service_runtime' / 'testdata'),
        (args.output / 'third_party' / 'blink' / 'tools'),
        (args.output / 'third_party' / 'blink' / 'web_tests'),
        (args.output / 'third_party' / 'breakpad' / 'breakpad' / 'src' / 'processor' / 'testdata'),
        (args.output / 'third_party' / 'catapult' / 'tracing' / 'test_data'),
        (args.output / 'third_party' / 'hunspell' / 'tests'),
        (args.output / 'third_party' / 'hunspell_dictionaries'),
        (args.output / 'third_party' / 'jdk' / 'current'),
        (args.output / 'third_party' / 'jdk' / 'extras'),
        (args.output / 'third_party' / 'liblouis' / 'src' / 'tests' / 'braille-specs'),
        (args.output / 'third_party' / 'xdg-utils' / 'tests'),
        (args.output / 'v8' / 'test'),
    )
    keep_files = (
        (args.output / 'chrome' / 'test' / 'data' / 'webui' / 'i18n_process_css_test.html'),
        (args.output / 'chrome' / 'test' / 'data' / 'webui' / 'mojo' / 'foobar.mojom'),
        (args.output / 'chrome' / 'test' / 'data' / 'webui' / 'web_ui_test.mojom'),
        (args.output / 'v8' / 'test' / 'torque' / 'test-torque.tq'),
    )
    keep_suffix = ('.gn', '.gni', '.grd', '.gyp', '.isolate', '.pydeps')
    # Include Contingent Paths
    for cpath in CONTINGENT_PATHS:
        if args.sysroot and f'{args.sysroot}-sysroot' in cpath:
            continue
        remove_dirs += (args.output / Path(cpath), )
    for remove_dir in remove_dirs:
        for path in sorted(remove_dir.rglob('*'), key=lambda l: len(str(l)), reverse=True):
            if path.is_file() and path not in keep_files and path.suffix not in keep_suffix:
                try:
                    path.unlink()
                # read-only files can't be deleted on Windows
                # so remove the flag and try again.
                except PermissionError:
                    path.chmod(S_IWRITE)
                    path.unlink()
            elif path.is_dir() and not any(path.iterdir()):
                try:
                    path.rmdir()
                except PermissionError:
                    path.chmod(S_IWRITE)
                    path.rmdir()
    for path in sorted(args.output.rglob('*'), key=lambda l: len(str(l)), reverse=True):
        if not path.is_symlink() and '.git' not in path.parts:
            if path.is_file() and ('out' in path.parts or path.name.startswith('ChangeLog')):
                try:
                    path.unlink()
                except PermissionError:
                    path.chmod(S_IWRITE)
                    path.unlink()
            elif path.is_dir() and not any(path.iterdir()):
                try:
                    path.rmdir()
                except PermissionError:
                    path.chmod(S_IWRITE)
                    path.rmdir()

    get_logger().info('Source cloning complete')


def main():
    """CLI Entrypoint"""
    parser = ArgumentParser(description=__doc__)
    parser.add_argument('-o',
                        '--output',
                        type=Path,
                        metavar='DIRECTORY',
                        default='chromium',
                        help='Output directory for the cloned sources. Default: %(default)s')
    parser.add_argument('-c',
                        '--custom-config',
                        type=Path,
                        metavar='FILE',
                        help='Supply a replacement for the default gclient config.')
    parser.add_argument('-p',
                        '--pgo',
                        default='linux',
                        choices=('linux', 'mac', 'mac-arm', 'win32', 'win64'),
                        help='Specifiy which pgo profile to download.  Default: %(default)s')
    parser.add_argument('-s',
                        '--sysroot',
                        choices=('amd64', 'arm64', 'armhf', 'i386', 'mips64el', 'mipsel'),
                        help='Download a linux sysroot for the given architecture')
    add_common_params(parser)
    args = parser.parse_args()
    clone(args)


if __name__ == '__main__':
    main()



File: utils/_extraction.py

# -*- coding: UTF-8 -*-

# Copyright (c) 2019 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""
Archive extraction utilities
"""

import os
import shutil
import subprocess
import tarfile
from pathlib import Path, PurePosixPath

from _common import (USE_REGISTRY, PlatformEnum, ExtractorEnum, get_logger, get_running_platform)
from prune_binaries import CONTINGENT_PATHS

DEFAULT_EXTRACTORS = {
    ExtractorEnum.SEVENZIP: USE_REGISTRY,
    ExtractorEnum.TAR: 'tar',
    ExtractorEnum.WINRAR: USE_REGISTRY,
}


def _find_7z_by_registry():
    """
    Return a string to 7-zip's 7z.exe from the Windows Registry.
    """
    import winreg #pylint: disable=import-error, import-outside-toplevel
    sub_key_7zfm = 'SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\App Paths\\7zFM.exe'
    try:
        with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, sub_key_7zfm) as key_handle:
            sevenzipfm_dir = winreg.QueryValueEx(key_handle, 'Path')[0]
    except OSError:
        get_logger().exception('Unable to locate 7-zip from the Windows Registry')
        raise
    sevenzip_path = Path(sevenzipfm_dir, '7z.exe')
    if not sevenzip_path.is_file():
        get_logger().error('7z.exe not found at path from registry: %s', sevenzip_path)
    return sevenzip_path


def _find_winrar_by_registry():
    """
    Return a string to WinRAR's WinRAR.exe from the Windows Registry.
    """
    import winreg #pylint: disable=import-error, import-outside-toplevel
    sub_key_winrar = 'SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\App Paths\\WinRAR.exe'
    try:
        with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, sub_key_winrar) as key_handle:
            winrar_dir = winreg.QueryValueEx(key_handle, 'Path')[0]
    except OSError:
        get_logger().exception('Unable to locale WinRAR from the Windows Registry')
        raise
    winrar_path = Path(winrar_dir, 'WinRAR.exe')
    if not winrar_path.is_file():
        get_logger().error('WinRAR.exe not found at path from registry: %s', winrar_path)
    return winrar_path


def _find_extractor_by_cmd(extractor_cmd):
    """Returns a string path to the binary; None if it couldn't be found"""
    if not extractor_cmd:
        return None
    if Path(extractor_cmd).is_file():
        return extractor_cmd
    return shutil.which(extractor_cmd)


def _process_relative_to(unpack_root, relative_to):
    """
    For an extractor that doesn't support an automatic transform, move the extracted
    contents from the relative_to/ directory to the unpack_root

    If relative_to is None, nothing is done.
    """
    if relative_to is None:
        return
    relative_root = unpack_root / relative_to
    if not relative_root.is_dir():
        get_logger().error('Could not find relative_to directory in extracted files: %s',
                           relative_to)
        raise Exception()
    for src_path in relative_root.iterdir():
        dest_path = unpack_root / src_path.name
        src_path.rename(dest_path)
    relative_root.rmdir()


def _extract_tar_with_7z(binary, archive_path, output_dir, relative_to, skip_unused, sysroot):
    get_logger().debug('Using 7-zip extractor')
    if not relative_to is None and (output_dir / relative_to).exists():
        get_logger().error('Temporary unpacking directory already exists: %s',
                           output_dir / relative_to)
        raise Exception()
    cmd1 = (binary, 'x', str(archive_path), '-so')
    cmd2 = (binary, 'x', '-si', '-aoa', '-ttar', '-o{}'.format(str(output_dir)))
    if skip_unused:
        for cpath in CONTINGENT_PATHS:
            if sysroot and f'{sysroot}-sysroot' in cpath:
                continue
            cmd2 += ('-x!%s/%s' % (str(relative_to), cpath[:-1]), )
    get_logger().debug('7z command line: %s | %s', ' '.join(cmd1), ' '.join(cmd2))

    proc1 = subprocess.Popen(cmd1, stdout=subprocess.PIPE)
    proc2 = subprocess.Popen(cmd2, stdin=proc1.stdout, stdout=subprocess.PIPE)
    proc1.stdout.close()
    (stdout_data, stderr_data) = proc2.communicate()
    if proc2.returncode != 0:
        get_logger().error('7z commands returned non-zero status: %s', proc2.returncode)
        get_logger().debug('stdout: %s', stdout_data)
        get_logger().debug('stderr: %s', stderr_data)
        raise Exception()

    _process_relative_to(output_dir, relative_to)


def _extract_tar_with_tar(binary, archive_path, output_dir, relative_to, skip_unused, sysroot):
    get_logger().debug('Using BSD or GNU tar extractor')
    output_dir.mkdir(exist_ok=True)
    cmd = (binary, '-xf', str(archive_path), '-C', str(output_dir))
    if skip_unused:
        for cpath in CONTINGENT_PATHS:
            if sysroot and f'{sysroot}-sysroot' in cpath:
                continue
            cmd += ('--exclude=%s/%s' % (str(relative_to), cpath[:-1]), )
    get_logger().debug('tar command line: %s', ' '.join(cmd))
    result = subprocess.run(cmd, check=False)
    if result.returncode != 0:
        get_logger().error('tar command returned %s', result.returncode)
        raise Exception()

    # for gnu tar, the --transform option could be used. but to keep compatibility with
    # bsdtar on macos, we just do this ourselves
    _process_relative_to(output_dir, relative_to)


def _extract_tar_with_winrar(binary, archive_path, output_dir, relative_to, skip_unused, sysroot):
    get_logger().debug('Using WinRAR extractor')
    output_dir.mkdir(exist_ok=True)
    cmd = (binary, 'x', '-o+', str(archive_path), str(output_dir))
    if skip_unused:
        for cpath in CONTINGENT_PATHS:
            if sysroot and f'{sysroot}-sysroot' in cpath:
                continue
            cmd += ('-x%s%s%s' % (str(relative_to), os.sep, cpath[:-1].replace('/')), )
    get_logger().debug('WinRAR command line: %s', ' '.join(cmd))
    result = subprocess.run(cmd, check=False)
    if result.returncode != 0:
        get_logger().error('WinRAR command returned %s', result.returncode)
        raise Exception()

    _process_relative_to(output_dir, relative_to)


def _extract_tar_with_python(archive_path, output_dir, relative_to, skip_unused, sysroot):
    get_logger().debug('Using pure Python tar extractor')

    class NoAppendList(list):
        """Hack to workaround memory issues with large tar files"""
        def append(self, obj):
            pass

    # Simple hack to check if symlinks are supported
    try:
        os.symlink('', '')
    except FileNotFoundError:
        # Symlinks probably supported
        symlink_supported = True
    except OSError:
        # Symlinks probably not supported
        get_logger().info('System does not support symlinks. Ignoring them.')
        symlink_supported = False
    except BaseException:
        # Unexpected exception
        get_logger().exception('Unexpected exception during symlink support check.')
        raise

    with tarfile.open(str(archive_path), 'r|%s' % archive_path.suffix[1:]) as tar_file_obj:
        tar_file_obj.members = NoAppendList()
        for tarinfo in tar_file_obj:
            try:
                if skip_unused and [
                        cpath for cpath in CONTINGENT_PATHS
                        if tarinfo.name.startswith(str(relative_to) + '/' + cpath)
                        and not (sysroot and f'{sysroot}-sysroot' in cpath)
                ]:
                    continue
                if relative_to is None:
                    destination = output_dir / PurePosixPath(tarinfo.name)
                else:
                    destination = output_dir / PurePosixPath(tarinfo.name).relative_to(relative_to)
                if tarinfo.issym() and not symlink_supported:
                    # In this situation, TarFile.makelink() will try to create a copy of the
                    # target. But this fails because TarFile.members is empty
                    # But if symlinks are not supported, it's safe to assume that symlinks
                    # aren't needed. The only situation where this happens is on Windows.
                    continue
                if tarinfo.islnk():
                    # Derived from TarFile.extract()
                    new_target = output_dir / PurePosixPath(
                        tarinfo.linkname).relative_to(relative_to)
                    tarinfo._link_target = new_target.as_posix() # pylint: disable=protected-access
                if destination.is_symlink():
                    destination.unlink()
                tar_file_obj._extract_member(tarinfo, str(destination)) # pylint: disable=protected-access
            except BaseException:
                get_logger().exception('Exception thrown for tar member: %s', tarinfo.name)
                raise


def extract_tar_file(archive_path, output_dir, relative_to, skip_unused, sysroot, extractors=None):
    """
    Extract regular or compressed tar archive into the output directory.

    archive_path is the pathlib.Path to the archive to unpack
    output_dir is a pathlib.Path to the directory to unpack. It must already exist.

    relative_to is a pathlib.Path for directories that should be stripped relative to the
        root of the archive, or None if no path components should be stripped.
    extractors is a dictionary of PlatformEnum to a command or path to the
        extractor binary. Defaults to 'tar' for tar, and '_use_registry' for 7-Zip and WinRAR.
    """
    if extractors is None:
        extractors = DEFAULT_EXTRACTORS

    current_platform = get_running_platform()
    if current_platform == PlatformEnum.WINDOWS:
        # Try to use 7-zip first
        sevenzip_cmd = extractors.get(ExtractorEnum.SEVENZIP)
        if sevenzip_cmd == USE_REGISTRY:
            sevenzip_cmd = str(_find_7z_by_registry())
        sevenzip_bin = _find_extractor_by_cmd(sevenzip_cmd)
        if sevenzip_bin is not None:
            _extract_tar_with_7z(sevenzip_bin, archive_path, output_dir, relative_to, skip_unused,
                                 sysroot)
            return

        # Use WinRAR if 7-zip is not found
        winrar_cmd = extractors.get(ExtractorEnum.WINRAR)
        if winrar_cmd == USE_REGISTRY:
            winrar_cmd = str(_find_winrar_by_registry())
        winrar_bin = _find_extractor_by_cmd(winrar_cmd)
        if winrar_bin is not None:
            _extract_tar_with_winrar(winrar_bin, archive_path, output_dir, relative_to, skip_unused,
                                     sysroot)
            return
        get_logger().warning(
            'Neither 7-zip nor WinRAR were found. Falling back to Python extractor...')
    elif current_platform == PlatformEnum.UNIX:
        # NOTE: 7-zip isn't an option because it doesn't preserve file permissions
        tar_bin = _find_extractor_by_cmd(extractors.get(ExtractorEnum.TAR))
        if not tar_bin is None:
            _extract_tar_with_tar(tar_bin, archive_path, output_dir, relative_to, skip_unused,
                                  sysroot)
            return
    else:
        # This is not a normal code path, so make it clear.
        raise NotImplementedError(current_platform)
    # Fallback to Python-based extractor on all platforms
    _extract_tar_with_python(archive_path, output_dir, relative_to, skip_unused, sysroot)


def extract_with_7z(archive_path, output_dir, relative_to, skip_unused, sysroot, extractors=None):
    """
    Extract archives with 7-zip into the output directory.
    Only supports archives with one layer of unpacking, so compressed tar archives don't work.

    archive_path is the pathlib.Path to the archive to unpack
    output_dir is a pathlib.Path to the directory to unpack. It must already exist.

    relative_to is a pathlib.Path for directories that should be stripped relative to the
    root of the archive.
    extractors is a dictionary of PlatformEnum to a command or path to the
    extractor binary. Defaults to 'tar' for tar, and '_use_registry' for 7-Zip.
    """
    # TODO: It would be nice to extend this to support arbitrary standard IO chaining of 7z
    # instances, so _extract_tar_with_7z and other future formats could use this.
    if extractors is None:
        extractors = DEFAULT_EXTRACTORS
    sevenzip_cmd = extractors.get(ExtractorEnum.SEVENZIP)
    if sevenzip_cmd == USE_REGISTRY:
        if not get_running_platform() == PlatformEnum.WINDOWS:
            get_logger().error('"%s" for 7-zip is only available on Windows', sevenzip_cmd)
            raise Exception()
        sevenzip_cmd = str(_find_7z_by_registry())
    sevenzip_bin = _find_extractor_by_cmd(sevenzip_cmd)

    if not relative_to is None and (output_dir / relative_to).exists():
        get_logger().error('Temporary unpacking directory already exists: %s',
                           output_dir / relative_to)
        raise Exception()
    cmd = (sevenzip_bin, 'x', str(archive_path), '-aoa', '-o{}'.format(str(output_dir)))
    if skip_unused:
        for cpath in CONTINGENT_PATHS:
            if sysroot and f'{sysroot}-sysroot' in cpath:
                continue
            cmd += ('-x!%s/%s' % (str(relative_to), cpath[:-1]), )
    get_logger().debug('7z command line: %s', ' '.join(cmd))

    result = subprocess.run(cmd, check=False)
    if result.returncode != 0:
        get_logger().error('7z command returned %s', result.returncode)
        raise Exception()

    _process_relative_to(output_dir, relative_to)


def extract_with_winrar(archive_path,
                        output_dir,
                        relative_to,
                        skip_unused,
                        sysroot,
                        extractors=None):
    """
    Extract archives with WinRAR into the output directory.
    Only supports archives with one layer of unpacking, so compressed tar archives don't work.

    archive_path is the pathlib.Path to the archive to unpack
    output_dir is a pathlib.Path to the directory to unpack. It must already exist.

    relative_to is a pathlib.Path for directories that should be stripped relative to the
    root of the archive.
    extractors is a dictionary of PlatformEnum to a command or path to the
    extractor binary. Defaults to 'tar' for tar, and '_use_registry' for WinRAR.
    """
    if extractors is None:
        extractors = DEFAULT_EXTRACTORS
    winrar_cmd = extractors.get(ExtractorEnum.WINRAR)
    if winrar_cmd == USE_REGISTRY:
        if not get_running_platform() == PlatformEnum.WINDOWS:
            get_logger().error('"%s" for WinRAR is only available on Windows', winrar_cmd)
            raise Exception()
        winrar_cmd = str(_find_winrar_by_registry())
    winrar_bin = _find_extractor_by_cmd(winrar_cmd)

    if not relative_to is None and (output_dir / relative_to).exists():
        get_logger().error('Temporary unpacking directory already exists: %s',
                           output_dir / relative_to)
        raise Exception()
    cmd = (winrar_bin, 'x', '-o+', str(archive_path), str(output_dir))
    if skip_unused:
        for cpath in CONTINGENT_PATHS:
            if sysroot and f'{sysroot}-sysroot' in cpath:
                continue
            cmd += ('-x%s%s%s' % (str(relative_to), os.sep, cpath[:-1].replace('/', os.sep)), )
    get_logger().debug('WinRAR command line: %s', ' '.join(cmd))

    result = subprocess.run(cmd, check=False)
    if result.returncode != 0:
        get_logger().error('WinRAR command returned %s', result.returncode)
        raise Exception()

    _process_relative_to(output_dir, relative_to)



File: utils/third_party/__init__.py




File: utils/third_party/schema.py

"""schema is a library for validating Python data structures, such as those
obtained from config-files, forms, external services or command-line
parsing, converted from JSON/YAML (or something else) to Python data-types."""

import re

__version__ = '0.6.7'
__all__ = ['Schema',
           'And', 'Or', 'Regex', 'Optional', 'Use', 'Forbidden', 'Const',
           'SchemaError',
           'SchemaWrongKeyError',
           'SchemaMissingKeyError',
           'SchemaForbiddenKeyError',
           'SchemaUnexpectedTypeError']


class SchemaError(Exception):
    """Error during Schema validation."""

    def __init__(self, autos, errors=None):
        self.autos = autos if type(autos) is list else [autos]
        self.errors = errors if type(errors) is list else [errors]
        Exception.__init__(self, self.code)

    @property
    def code(self):
        """
        Removes duplicates values in auto and error list.
        parameters.
        """
        def uniq(seq):
            """
            Utility function that removes duplicate.
            """
            seen = set()
            seen_add = seen.add
            # This way removes duplicates while preserving the order.
            return [x for x in seq if x not in seen and not seen_add(x)]
        data_set = uniq(i for i in self.autos if i is not None)
        error_list = uniq(i for i in self.errors if i is not None)
        if error_list:
            return '\n'.join(error_list)
        return '\n'.join(data_set)


class SchemaWrongKeyError(SchemaError):
    """Error Should be raised when an unexpected key is detected within the
    data set being."""
    pass


class SchemaMissingKeyError(SchemaError):
    """Error should be raised when a mandatory key is not found within the
    data set being vaidated"""
    pass


class SchemaForbiddenKeyError(SchemaError):
    """Error should be raised when a forbidden key is found within the
    data set being validated, and its value matches the value that was specified"""
    pass


class SchemaUnexpectedTypeError(SchemaError):
    """Error should be raised when a type mismatch is detected within the
    data set being validated."""
    pass


class And:
    """
    Utility function to combine validation directives in AND Boolean fashion.
    """
    def __init__(self, *args, **kw):
        self._args = args
        assert set(kw).issubset(['error', 'schema', 'ignore_extra_keys'])
        self._error = kw.get('error')
        self._ignore_extra_keys = kw.get('ignore_extra_keys', False)
        # You can pass your inherited Schema class.
        self._schema = kw.get('schema', Schema)

    def __repr__(self):
        return '%s(%s)' % (self.__class__.__name__,
                           ', '.join(repr(a) for a in self._args))

    def validate(self, data):
        """
        Validate data using defined sub schema/expressions ensuring all
        values are valid.
        :param data: to be validated with sub defined schemas.
        :return: returns validated data
        """
        for s in [self._schema(s, error=self._error,
                               ignore_extra_keys=self._ignore_extra_keys)
                  for s in self._args]:
            data = s.validate(data)
        return data


class Or(And):
    """Utility function to combine validation directives in a OR Boolean
    fashion."""
    def validate(self, data):
        """
        Validate data using sub defined schema/expressions ensuring at least
        one value is valid.
        :param data: data to be validated by provided schema.
        :return: return validated data if not validation
        """
        x = SchemaError([], [])
        for s in [self._schema(s, error=self._error,
                               ignore_extra_keys=self._ignore_extra_keys)
                  for s in self._args]:
            try:
                return s.validate(data)
            except SchemaError as _x:
                x = _x
        raise SchemaError(['%r did not validate %r' % (self, data)] + x.autos,
                          [self._error.format(data) if self._error else None] +
                          x.errors)


class Regex:
    """
    Enables schema.py to validate string using regular expressions.
    """
    # Map all flags bits to a more readable description
    NAMES = ['re.ASCII', 're.DEBUG', 're.VERBOSE', 're.UNICODE', 're.DOTALL',
             're.MULTILINE', 're.LOCALE', 're.IGNORECASE', 're.TEMPLATE']

    def __init__(self, pattern_str, flags=0, error=None):
        self._pattern_str = pattern_str
        flags_list = [Regex.NAMES[i] for i, f in  # Name for each bit
                      enumerate('{0:09b}'.format(flags)) if f != '0']

        if flags_list:
            self._flags_names = ', flags=' + '|'.join(flags_list)
        else:
            self._flags_names = ''

        self._pattern = re.compile(pattern_str, flags=flags)
        self._error = error

    def __repr__(self):
        return '%s(%r%s)' % (
            self.__class__.__name__, self._pattern_str, self._flags_names
        )

    def validate(self, data):
        """
        Validated data using defined regex.
        :param data: data to be validated
        :return: return validated data.
        """
        e = self._error

        try:
            if self._pattern.search(data):
                return data
            else:
                raise SchemaError('%r does not match %r' % (self, data), e)
        except TypeError:
            raise SchemaError('%r is not string nor buffer' % data, e)


class Use:
    """
    For more general use cases, you can use the Use class to transform
    the data while it is being validate.
    """
    def __init__(self, callable_, error=None):
        assert callable(callable_)
        self._callable = callable_
        self._error = error

    def __repr__(self):
        return '%s(%r)' % (self.__class__.__name__, self._callable)

    def validate(self, data):
        try:
            return self._callable(data)
        except SchemaError as x:
            raise SchemaError([None] + x.autos,
                              [self._error.format(data)
                               if self._error else None] + x.errors)
        except BaseException as x:
            f = _callable_str(self._callable)
            raise SchemaError('%s(%r) raised %r' % (f, data, x),
                              self._error.format(data)
                              if self._error else None)


COMPARABLE, CALLABLE, VALIDATOR, TYPE, DICT, ITERABLE = range(6)


def _priority(s):
    """Return priority for a given object."""
    if type(s) in (list, tuple, set, frozenset):
        return ITERABLE
    if type(s) is dict:
        return DICT
    if issubclass(type(s), type):
        return TYPE
    if hasattr(s, 'validate'):
        return VALIDATOR
    if callable(s):
        return CALLABLE
    else:
        return COMPARABLE


class Schema:
    """
    Entry point of the library, use this class to instantiate validation
    schema for the data that will be validated.
    """
    def __init__(self, schema, error=None, ignore_extra_keys=False):
        self._schema = schema
        self._error = error
        self._ignore_extra_keys = ignore_extra_keys

    def __repr__(self):
        return '%s(%r)' % (self.__class__.__name__, self._schema)

    @staticmethod
    def _dict_key_priority(s):
        """Return priority for a given key object."""
        if isinstance(s, Forbidden):
            return _priority(s._schema) - 0.5
        if isinstance(s, Optional):
            return _priority(s._schema) + 0.5
        return _priority(s)

    def validate(self, data):
        Schema = self.__class__
        s = self._schema
        e = self._error
        i = self._ignore_extra_keys
        flavor = _priority(s)
        if flavor == ITERABLE:
            data = Schema(type(s), error=e).validate(data)
            o = Or(*s, error=e, schema=Schema, ignore_extra_keys=i)
            return type(data)(o.validate(d) for d in data)
        if flavor == DICT:
            data = Schema(dict, error=e).validate(data)
            new = type(data)()  # new - is a dict of the validated values
            coverage = set()  # matched schema keys
            # for each key and value find a schema entry matching them, if any
            sorted_skeys = sorted(s, key=self._dict_key_priority)
            for key, value in data.items():
                for skey in sorted_skeys:
                    svalue = s[skey]
                    try:
                        nkey = Schema(skey, error=e).validate(key)
                    except SchemaError:
                        pass
                    else:
                        if isinstance(skey, Forbidden):
                            # As the content of the value makes little sense for
                            # forbidden keys, we reverse its meaning:
                            # we will only raise the SchemaErrorForbiddenKey
                            # exception if the value does match, allowing for
                            # excluding a key only if its value has a certain type,
                            # and allowing Forbidden to work well in combination
                            # with Optional.
                            try:
                                nvalue = Schema(svalue, error=e).validate(value)
                            except SchemaError:
                                continue
                            raise SchemaForbiddenKeyError(
                                    'Forbidden key encountered: %r in %r' %
                                    (nkey, data), e)
                        try:
                            nvalue = Schema(svalue, error=e,
                                            ignore_extra_keys=i).validate(value)
                        except SchemaError as x:
                            k = "Key '%s' error:" % nkey
                            raise SchemaError([k] + x.autos, [e] + x.errors)
                        else:
                            new[nkey] = nvalue
                            coverage.add(skey)
                            break
            required = {k for k in s if type(k) not in [Optional, Forbidden]}
            if not required.issubset(coverage):
                missing_keys = required - coverage
                s_missing_keys = \
                    ', '.join(repr(k) for k in sorted(missing_keys, key=repr))
                raise \
                    SchemaMissingKeyError('Missing keys: ' + s_missing_keys, e)
            if not self._ignore_extra_keys and (len(new) != len(data)):
                wrong_keys = set(data.keys()) - set(new.keys())
                s_wrong_keys = \
                    ', '.join(repr(k) for k in sorted(wrong_keys, key=repr))
                raise \
                    SchemaWrongKeyError(
                        'Wrong keys %s in %r' % (s_wrong_keys, data),
                        e.format(data) if e else None)

            # Apply default-having optionals that haven't been used:
            defaults = {k for k in s if type(k) is Optional and
                           hasattr(k, 'default')} - coverage
            for default in defaults:
                new[default.key] = default.default

            return new
        if flavor == TYPE:
            if isinstance(data, s):
                return data
            else:
                raise SchemaUnexpectedTypeError(
                    '%r should be instance of %r' % (data, s.__name__),
                    e.format(data) if e else None)
        if flavor == VALIDATOR:
            try:
                return s.validate(data)
            except SchemaError as x:
                raise SchemaError([None] + x.autos, [e] + x.errors)
            except BaseException as x:
                raise SchemaError(
                    '%r.validate(%r) raised %r' % (s, data, x),
                    self._error.format(data) if self._error else None)
        if flavor == CALLABLE:
            f = _callable_str(s)
            try:
                if s(data):
                    return data
            except SchemaError as x:
                raise SchemaError([None] + x.autos, [e] + x.errors)
            except BaseException as x:
                raise SchemaError(
                    '%s(%r) raised %r' % (f, data, x),
                    self._error.format(data) if self._error else None)
            raise SchemaError('%s(%r) should evaluate to True' % (f, data), e)
        if s == data:
            return data
        else:
            raise SchemaError('%r does not match %r' % (s, data),
                              e.format(data) if e else None)


class Optional(Schema):
    """Marker for an optional part of the validation Schema."""
    _MARKER = object()

    def __init__(self, *args, **kwargs):
        default = kwargs.pop('default', self._MARKER)
        super(Optional, self).__init__(*args, **kwargs)
        if default is not self._MARKER:
            # See if I can come up with a static key to use for myself:
            if _priority(self._schema) != COMPARABLE:
                raise TypeError(
                    'Optional keys with defaults must have simple, '
                    'predictable values, like literal strings or ints. '
                    '"%r" is too complex.' % (self._schema,))
            self.default = default
            self.key = self._schema

    def __hash__(self):
        return hash(self._schema)

    def __eq__(self, other):
        return (self.__class__ is other.__class__ and
                getattr(self, 'default', self._MARKER) ==
                getattr(other, 'default', self._MARKER) and
                self._schema == other._schema)


class Forbidden(Schema):
    def __init__(self, *args, **kwargs):
        super(Forbidden, self).__init__(*args, **kwargs)
        self.key = self._schema


class Const(Schema):
    def validate(self, data):
        super(Const, self).validate(data)
        return data


def _callable_str(callable_):
    if hasattr(callable_, '__name__'):
        return callable_.__name__
    return str(callable_)



File: devutils/validate_patches.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Copyright (c) 2020 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""
Validates that all patches apply cleanly against the source tree.

The required source tree files can be retrieved from Google directly.
"""

import argparse
import ast
import base64
import email.utils
import json
import logging
import sys
import tempfile
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent / 'third_party'))
import unidiff
from unidiff.constants import LINE_TYPE_EMPTY, LINE_TYPE_NO_NEWLINE
sys.path.pop(0)

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / 'utils'))
from domain_substitution import TREE_ENCODINGS
from _common import ENCODING, get_logger, get_chromium_version, parse_series, add_common_params
from patches import dry_run_check
sys.path.pop(0)

try:
    import requests
    import requests.adapters
    import urllib3.util

    class _VerboseRetry(urllib3.util.Retry):
        """A more verbose version of HTTP Adatper about retries"""
        def sleep_for_retry(self, response=None):
            """Sleeps for Retry-After, and logs the sleep time"""
            if response:
                retry_after = self.get_retry_after(response)
                if retry_after:
                    get_logger().info(
                        'Got HTTP status %s with Retry-After header. Retrying after %s seconds...',
                        response.status, retry_after)
                else:
                    get_logger().info(
                        'Could not find Retry-After header for HTTP response %s. Status reason: %s',
                        response.status, response.reason)
            return super().sleep_for_retry(response)

        def _sleep_backoff(self):
            """Log info about backoff sleep"""
            get_logger().info('Running HTTP request sleep backoff')
            super()._sleep_backoff()

    def _get_requests_session():
        session = requests.Session()
        http_adapter = requests.adapters.HTTPAdapter(
            max_retries=_VerboseRetry(total=10,
                                      read=10,
                                      connect=10,
                                      backoff_factor=8,
                                      status_forcelist=urllib3.Retry.RETRY_AFTER_STATUS_CODES,
                                      raise_on_status=False))
        session.mount('http://', http_adapter)
        session.mount('https://', http_adapter)
        return session
except ImportError:

    def _get_requests_session():
        raise RuntimeError('The Python module "requests" is required for remote'
                           'file downloading. It can be installed from PyPI.')


_ROOT_DIR = Path(__file__).resolve().parent.parent
_SRC_PATH = Path('src')


class _PatchValidationError(Exception):
    """Raised when patch validation fails"""


class _UnexpectedSyntaxError(RuntimeError):
    """Raised when unexpected syntax is used in DEPS"""


class _NotInRepoError(RuntimeError):
    """Raised when the remote file is not present in the given repo"""


class _DepsNodeVisitor(ast.NodeVisitor):
    _valid_syntax_types = (ast.mod, ast.expr_context, ast.boolop, ast.Assign, ast.Add, ast.Name,
                           ast.Dict, ast.Str, ast.NameConstant, ast.List, ast.BinOp)
    _allowed_callables = ('Var', )

    def visit_Call(self, node): #pylint: disable=invalid-name
        """Override Call syntax handling"""
        if node.func.id not in self._allowed_callables:
            raise _UnexpectedSyntaxError('Unexpected call of "%s" at line %s, column %s' %
                                         (node.func.id, node.lineno, node.col_offset))

    def generic_visit(self, node):
        for ast_type in self._valid_syntax_types:
            if isinstance(node, ast_type):
                super().generic_visit(node)
                return
        raise _UnexpectedSyntaxError('Unexpected {} at line {}, column {}'.format(
            type(node).__name__, node.lineno, node.col_offset))


def _validate_deps(deps_text):
    """Returns True if the DEPS file passes validation; False otherwise"""
    try:
        _DepsNodeVisitor().visit(ast.parse(deps_text))
    except _UnexpectedSyntaxError as exc:
        get_logger().error('%s', exc)
        return False
    return True


def _deps_var(deps_globals):
    """Return a function that implements DEPS's Var() function"""
    def _var_impl(var_name):
        """Implementation of Var() in DEPS"""
        return deps_globals['vars'][var_name]

    return _var_impl


def _parse_deps(deps_text):
    """Returns a dict of parsed DEPS data"""
    deps_globals = {'__builtins__': None}
    deps_globals['Var'] = _deps_var(deps_globals)
    exec(deps_text, deps_globals) #pylint: disable=exec-used
    return deps_globals


def _download_googlesource_file(download_session, repo_url, version, relative_path):
    """
    Returns the contents of the text file with path within the given
    googlesource.com repo as a string.
    """
    if 'googlesource.com' not in repo_url:
        raise ValueError('Repository URL is not a googlesource.com URL: {}'.format(repo_url))
    full_url = repo_url + '/+/{}/{}?format=TEXT'.format(version, str(relative_path))
    get_logger().debug('Downloading: %s', full_url)
    response = download_session.get(full_url)
    if response.status_code == 404:
        raise _NotInRepoError()
    response.raise_for_status()
    # Assume all files that need patching are compatible with UTF-8
    return base64.b64decode(response.text, validate=True).decode('UTF-8')


def _get_dep_value_url(deps_globals, dep_value):
    """Helper for _process_deps_entries"""
    if isinstance(dep_value, str):
        url = dep_value
    elif isinstance(dep_value, dict):
        if 'url' not in dep_value:
            # Ignore other types like CIPD since
            # it probably isn't necessary
            return None
        url = dep_value['url']
    else:
        raise NotImplementedError()
    if '{' in url:
        # Probably a Python format string
        url = url.format(**deps_globals['vars'])
    if url.count('@') != 1:
        raise _PatchValidationError('Invalid number of @ symbols in URL: {}'.format(url))
    return url


def _process_deps_entries(deps_globals, child_deps_tree, child_path, deps_use_relative_paths):
    """Helper for _get_child_deps_tree"""
    for dep_path_str, dep_value in deps_globals.get('deps', dict()).items():
        url = _get_dep_value_url(deps_globals, dep_value)
        if url is None:
            continue
        dep_path = Path(dep_path_str)
        if not deps_use_relative_paths:
            try:
                dep_path = Path(dep_path_str).relative_to(child_path)
            except ValueError:
                # Not applicable to the current DEPS tree path
                continue
        grandchild_deps_tree = None # Delaying creation of dict() until it's needed
        for recursedeps_item in deps_globals.get('recursedeps', tuple()):
            if isinstance(recursedeps_item, str):
                if recursedeps_item == str(dep_path):
                    grandchild_deps_tree = 'DEPS'
            else: # Some sort of iterable
                recursedeps_item_path, recursedeps_item_depsfile = recursedeps_item
                if recursedeps_item_path == str(dep_path):
                    grandchild_deps_tree = recursedeps_item_depsfile
        if grandchild_deps_tree is None:
            # This dep is not recursive; i.e. it is fully loaded
            grandchild_deps_tree = dict()
        child_deps_tree[dep_path] = (*url.split('@'), grandchild_deps_tree)


def _get_child_deps_tree(download_session, current_deps_tree, child_path, deps_use_relative_paths):
    """Helper for _download_source_file"""
    repo_url, version, child_deps_tree = current_deps_tree[child_path]
    if isinstance(child_deps_tree, str):
        # Load unloaded DEPS
        deps_globals = _parse_deps(
            _download_googlesource_file(download_session, repo_url, version, child_deps_tree))
        child_deps_tree = dict()
        current_deps_tree[child_path] = (repo_url, version, child_deps_tree)
        deps_use_relative_paths = deps_globals.get('use_relative_paths', False)
        _process_deps_entries(deps_globals, child_deps_tree, child_path, deps_use_relative_paths)
    return child_deps_tree, deps_use_relative_paths


def _get_last_chromium_modification():
    """Returns the last modification date of the chromium-browser-official tar file"""
    with _get_requests_session() as session:
        response = session.head(
            'https://storage.googleapis.com/chromium-browser-official/chromium-{}.tar.xz'.format(
                get_chromium_version()))
        response.raise_for_status()
        return email.utils.parsedate_to_datetime(response.headers['Last-Modified'])


def _get_gitiles_git_log_date(log_entry):
    """Helper for _get_gitiles_git_log_date"""
    return email.utils.parsedate_to_datetime(log_entry['committer']['time'])


def _get_gitiles_commit_before_date(repo_url, target_branch, target_datetime):
    """Returns the hexadecimal hash of the closest commit before target_datetime"""
    json_log_url = '{repo}/+log/{branch}?format=JSON'.format(repo=repo_url, branch=target_branch)
    with _get_requests_session() as session:
        response = session.get(json_log_url)
        response.raise_for_status()
        git_log = json.loads(response.text[5:]) # Trim closing delimiters for various structures
    assert len(git_log) == 2 # 'log' and 'next' entries
    assert 'log' in git_log
    assert git_log['log']
    git_log = git_log['log']
    # Check boundary conditions
    if _get_gitiles_git_log_date(git_log[0]) < target_datetime:
        # Newest commit is older than target datetime
        return git_log[0]['commit']
    if _get_gitiles_git_log_date(git_log[-1]) > target_datetime:
        # Oldest commit is newer than the target datetime; assume oldest is close enough.
        get_logger().warning('Oldest entry in gitiles log for repo "%s" is newer than target; '
                             'continuing with oldest entry...')
        return git_log[-1]['commit']
    # Do binary search
    low_index = 0
    high_index = len(git_log) - 1
    mid_index = high_index
    while low_index != high_index:
        mid_index = low_index + (high_index - low_index) // 2
        if _get_gitiles_git_log_date(git_log[mid_index]) > target_datetime:
            low_index = mid_index + 1
        else:
            high_index = mid_index
    return git_log[mid_index]['commit']


class _FallbackRepoManager:
    """Retrieves fallback repos and caches data needed for determining repos"""

    _GN_REPO_URL = 'https://gn.googlesource.com/gn.git'

    def __init__(self):
        self._cache_gn_version = None

    @property
    def gn_version(self):
        """
        Returns the version of the GN repo for the Chromium version used by this code
        """
        if not self._cache_gn_version:
            # Because there seems to be no reference to the logic for generating the
            # chromium-browser-official tar file, it's possible that it is being generated
            # by an internal script that manually injects the GN repository files.
            # Therefore, assume that the GN version used in the chromium-browser-official tar
            # files correspond to the latest commit in the master branch of the GN repository
            # at the time of the tar file's generation. We can get an approximation for the
            # generation time by using the last modification date of the tar file on
            # Google's file server.
            self._cache_gn_version = _get_gitiles_commit_before_date(
                self._GN_REPO_URL, 'master', _get_last_chromium_modification())
        return self._cache_gn_version

    def get_fallback(self, current_relative_path, current_node, root_deps_tree):
        """
        Helper for _download_source_file

        It returns a new (repo_url, version, new_relative_path) to attempt a file download with
        """
        assert len(current_node) == 3
        # GN special processing
        try:
            new_relative_path = current_relative_path.relative_to('tools/gn')
        except ValueError:
            pass
        else:
            if current_node is root_deps_tree[_SRC_PATH]:
                get_logger().info('Redirecting to GN repo version %s for path: %s', self.gn_version,
                                  current_relative_path)
                return (self._GN_REPO_URL, self.gn_version, new_relative_path)
        return None, None, None


def _get_target_file_deps_node(download_session, root_deps_tree, target_file):
    """
    Helper for _download_source_file

    Returns the corresponding repo containing target_file based on the DEPS tree
    """
    # The "deps" from the current DEPS file
    current_deps_tree = root_deps_tree
    current_node = None
    # Path relative to the current node (i.e. DEPS file)
    current_relative_path = Path('src', target_file)
    previous_relative_path = None
    deps_use_relative_paths = False
    child_path = None
    while current_relative_path != previous_relative_path:
        previous_relative_path = current_relative_path
        for child_path in current_deps_tree:
            try:
                current_relative_path = previous_relative_path.relative_to(child_path)
            except ValueError:
                # previous_relative_path does not start with child_path
                continue
            current_node = current_deps_tree[child_path]
            # current_node will match with current_deps_tree after the following statement
            current_deps_tree, deps_use_relative_paths = _get_child_deps_tree(
                download_session, current_deps_tree, child_path, deps_use_relative_paths)
            break
    assert not current_node is None
    return current_node, current_relative_path


def _download_source_file(download_session, root_deps_tree, fallback_repo_manager, target_file):
    """
    Downloads the source tree file from googlesource.com

    download_session is an active requests.Session() object
    deps_dir is a pathlib.Path to the directory containing a DEPS file.
    """
    current_node, current_relative_path = _get_target_file_deps_node(download_session,
                                                                     root_deps_tree, target_file)
    # Attempt download with potential fallback logic
    repo_url, version, _ = current_node
    try:
        # Download with DEPS-provided repo
        return _download_googlesource_file(download_session, repo_url, version,
                                           current_relative_path)
    except _NotInRepoError:
        pass
    get_logger().debug(
        'Path "%s" (relative: "%s") not found using DEPS tree; finding fallback repo...',
        target_file, current_relative_path)
    repo_url, version, current_relative_path = fallback_repo_manager.get_fallback(
        current_relative_path, current_node, root_deps_tree)
    if not repo_url:
        get_logger().error('No fallback repo found for "%s" (relative: "%s")', target_file,
                           current_relative_path)
        raise _NotInRepoError()
    try:
        # Download with fallback repo
        return _download_googlesource_file(download_session, repo_url, version,
                                           current_relative_path)
    except _NotInRepoError:
        pass
    get_logger().error('File "%s" (relative: "%s") not found in fallback repo "%s", version "%s"',
                       target_file, current_relative_path, repo_url, version)
    raise _NotInRepoError()


def _initialize_deps_tree():
    """
    Initializes and returns a dependency tree for DEPS files

    The DEPS tree is a dict has the following format:
    key - pathlib.Path relative to the DEPS file's path
    value - tuple(repo_url, version, recursive dict here)
        repo_url is the URL to the dependency's repository root
        If the recursive dict is a string, then it is a string to the DEPS file to load
            if needed

    download_session is an active requests.Session() object
    """
    root_deps_tree = {
        _SRC_PATH: ('https://chromium.googlesource.com/chromium/src.git', get_chromium_version(),
                    'DEPS')
    }
    return root_deps_tree


def _retrieve_remote_files(file_iter):
    """
    Retrieves all file paths in file_iter from Google

    file_iter is an iterable of strings that are relative UNIX paths to
        files in the Chromium source.

    Returns a dict of relative UNIX path strings to a list of lines in the file as strings
    """

    files = dict()

    root_deps_tree = _initialize_deps_tree()

    try:
        total_files = len(file_iter)
    except TypeError:
        total_files = None

    logger = get_logger()
    if total_files is None:
        logger.info('Downloading remote files...')
    else:
        logger.info('Downloading %d remote files...', total_files)
    last_progress = 0
    file_count = 0
    fallback_repo_manager = _FallbackRepoManager()
    with _get_requests_session() as download_session:
        download_session.stream = False # To ensure connection to Google can be reused
        for file_path in file_iter:
            if total_files:
                file_count += 1
                current_progress = file_count * 100 // total_files // 5 * 5
                if current_progress != last_progress:
                    last_progress = current_progress
                    logger.info('%d%% downloaded', current_progress)
            else:
                current_progress = file_count // 20 * 20
                if current_progress != last_progress:
                    last_progress = current_progress
                    logger.info('%d files downloaded', current_progress)
            try:
                files[file_path] = _download_source_file(download_session, root_deps_tree,
                                                         fallback_repo_manager,
                                                         file_path).split('\n')
            except _NotInRepoError:
                get_logger().warning('Could not find "%s" remotely. Skipping...', file_path)
    return files


def _retrieve_local_files(file_iter, source_dir):
    """
    Retrieves all file paths in file_iter from the local source tree

    file_iter is an iterable of strings that are relative UNIX paths to
        files in the Chromium source.

    Returns a dict of relative UNIX path strings to a list of lines in the file as strings
    """
    files = dict()
    for file_path in file_iter:
        try:
            raw_content = (source_dir / file_path).read_bytes()
        except FileNotFoundError:
            get_logger().warning('Missing file from patches: %s', file_path)
            continue
        for encoding in TREE_ENCODINGS:
            try:
                content = raw_content.decode(encoding)
                break
            except UnicodeDecodeError:
                continue
        if not content:
            raise UnicodeDecodeError('Unable to decode with any encoding: %s' % file_path)
        files[file_path] = content.split('\n')
    if not files:
        get_logger().error('All files used by patches are missing!')
    return files


def _modify_file_lines(patched_file, file_lines):
    """Helper for _apply_file_unidiff"""
    # Cursor for keeping track of the current line during hunk application
    # NOTE: The cursor is based on the line list index, not the line number!
    line_cursor = None
    for hunk in patched_file:
        # Validate hunk will match
        if not hunk.is_valid():
            raise _PatchValidationError('Hunk is not valid: {}'.format(repr(hunk)))
        line_cursor = hunk.target_start - 1
        for line in hunk:
            normalized_line = line.value.rstrip('\n')
            if line.is_added:
                file_lines[line_cursor:line_cursor] = (normalized_line, )
                line_cursor += 1
            elif line.is_removed:
                if normalized_line != file_lines[line_cursor]:
                    raise _PatchValidationError(
                        "Line '{}' does not match removal line '{}' from patch".format(
                            file_lines[line_cursor], normalized_line))
                del file_lines[line_cursor]
            elif line.is_context:
                if not normalized_line and line_cursor == len(file_lines):
                    # We reached the end of the file
                    break
                if normalized_line != file_lines[line_cursor]:
                    raise _PatchValidationError(
                        "Line '{}' does not match context line '{}' from patch".format(
                            file_lines[line_cursor], normalized_line))
                line_cursor += 1
            else:
                assert line.line_type in (LINE_TYPE_EMPTY, LINE_TYPE_NO_NEWLINE)


def _apply_file_unidiff(patched_file, files_under_test):
    """Applies the unidiff.PatchedFile to the source files under testing"""
    patched_file_path = Path(patched_file.path)
    if patched_file.is_added_file:
        if patched_file_path in files_under_test:
            assert files_under_test[patched_file_path] is None
        assert len(patched_file) == 1 # Should be only one hunk
        assert patched_file[0].removed == 0
        assert patched_file[0].target_start == 1
        files_under_test[patched_file_path] = [x.value.rstrip('\n') for x in patched_file[0]]
    elif patched_file.is_removed_file:
        # Remove lines to see if file to be removed matches patch
        _modify_file_lines(patched_file, files_under_test[patched_file_path])
        files_under_test[patched_file_path] = None
    else: # Patching an existing file
        assert patched_file.is_modified_file
        _modify_file_lines(patched_file, files_under_test[patched_file_path])


def _dry_check_patched_file(patched_file, orig_file_content):
    """Run "patch --dry-check" on a unidiff.PatchedFile for diagnostics"""
    with tempfile.TemporaryDirectory() as tmpdirname:
        tmp_dir = Path(tmpdirname)
        # Write file to patch
        patched_file_path = tmp_dir / patched_file.path
        patched_file_path.parent.mkdir(parents=True, exist_ok=True)
        patched_file_path.write_text(orig_file_content)
        # Write patch
        patch_path = tmp_dir / 'broken_file.patch'
        patch_path.write_text(str(patched_file))
        # Dry run
        _, dry_stdout, _ = dry_run_check(patch_path, tmp_dir)
        return dry_stdout


def _test_patches(series_iter, patch_cache, files_under_test):
    """
    Tests the patches specified in the iterable series_iter

    Returns a boolean indicating if any of the patches have failed
    """
    for patch_path_str in series_iter:
        for patched_file in patch_cache[patch_path_str]:
            orig_file_content = None
            if get_logger().isEnabledFor(logging.DEBUG):
                orig_file_content = files_under_test.get(Path(patched_file.path))
                if orig_file_content:
                    orig_file_content = ' '.join(orig_file_content)
            try:
                _apply_file_unidiff(patched_file, files_under_test)
            except _PatchValidationError as exc:
                get_logger().warning('Patch failed validation: %s', patch_path_str)
                get_logger().debug('Specifically, file "%s" failed validation: %s',
                                   patched_file.path, exc)
                if get_logger().isEnabledFor(logging.DEBUG):
                    # _PatchValidationError cannot be thrown when a file is added
                    assert patched_file.is_modified_file or patched_file.is_removed_file
                    assert orig_file_content is not None
                    get_logger().debug(
                        'Output of "patch --dry-run" for this patch on this file:\n%s',
                        _dry_check_patched_file(patched_file, orig_file_content))
                return True
            except: #pylint: disable=bare-except
                get_logger().warning('Patch failed validation: %s', patch_path_str)
                get_logger().debug('Specifically, file "%s" caused exception while applying:',
                                   patched_file.path,
                                   exc_info=True)
                return True
    return False


def _load_all_patches(series_iter, patches_dir):
    """
    Returns a tuple of the following:
    - boolean indicating success or failure of reading files
    - dict of relative UNIX path strings to unidiff.PatchSet
    """
    had_failure = False
    unidiff_dict = dict()
    for relative_path in series_iter:
        if relative_path in unidiff_dict:
            continue
        unidiff_dict[relative_path] = unidiff.PatchSet.from_filename(str(patches_dir /
                                                                         relative_path),
                                                                     encoding=ENCODING)
        if not (patches_dir / relative_path).read_text(encoding=ENCODING).endswith('\n'):
            had_failure = True
            get_logger().warning('Patch file does not end with newline: %s',
                                 str(patches_dir / relative_path))
    return had_failure, unidiff_dict


def _get_required_files(patch_cache):
    """Returns an iterable of pathlib.Path files needed from the source tree for patching"""
    new_files = set() # Files introduced by patches
    file_set = set()
    for patch_set in patch_cache.values():
        for patched_file in patch_set:
            if patched_file.is_added_file:
                new_files.add(patched_file.path)
            elif patched_file.path not in new_files:
                file_set.add(Path(patched_file.path))
    return file_set


def _get_files_under_test(args, required_files, parser):
    """
    Helper for main to get files_under_test

    Exits the program if --cache-remote debugging option is used
    """
    if args.local:
        files_under_test = _retrieve_local_files(required_files, args.local)
    else: # --remote and --cache-remote
        files_under_test = _retrieve_remote_files(required_files)
        if args.cache_remote:
            for file_path, file_content in files_under_test.items():
                if not (args.cache_remote / file_path).parent.exists():
                    (args.cache_remote / file_path).parent.mkdir(parents=True)
                with (args.cache_remote / file_path).open('w', encoding=ENCODING) as cache_file:
                    cache_file.write('\n'.join(file_content))
            parser.exit()
    return files_under_test


def main():
    """CLI Entrypoint"""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('-s',
                        '--series',
                        type=Path,
                        metavar='FILE',
                        default=str(Path('patches', 'series')),
                        help='The series file listing patches to apply. Default: %(default)s')
    parser.add_argument('-p',
                        '--patches',
                        type=Path,
                        metavar='DIRECTORY',
                        default='patches',
                        help='The patches directory to read from. Default: %(default)s')
    add_common_params(parser)

    file_source_group = parser.add_mutually_exclusive_group(required=True)
    file_source_group.add_argument(
        '-l',
        '--local',
        type=Path,
        metavar='DIRECTORY',
        help=
        'Use a local source tree. It must be UNMODIFIED, otherwise the results will not be valid.')
    file_source_group.add_argument(
        '-r',
        '--remote',
        action='store_true',
        help=('Download the required source tree files from Google. '
              'This feature requires the Python module "requests". If you do not want to '
              'install this, consider using --local instead.'))
    file_source_group.add_argument(
        '-c',
        '--cache-remote',
        type=Path,
        metavar='DIRECTORY',
        help='(For debugging) Store the required remote files in an empty local directory')
    args = parser.parse_args()
    if args.cache_remote and not args.cache_remote.exists():
        if args.cache_remote.parent.exists():
            args.cache_remote.mkdir()
        else:
            parser.error('Parent of cache path {} does not exist'.format(args.cache_remote))

    if not args.series.is_file():
        parser.error('--series path is not a file or not found: {}'.format(args.series))
    if not args.patches.is_dir():
        parser.error('--patches path is not a directory or not found: {}'.format(args.patches))

    series_iterable = tuple(parse_series(args.series))
    had_failure, patch_cache = _load_all_patches(series_iterable, args.patches)
    required_files = _get_required_files(patch_cache)
    files_under_test = _get_files_under_test(args, required_files, parser)
    had_failure |= _test_patches(series_iterable, patch_cache, files_under_test)
    if had_failure:
        get_logger().error('***FAILED VALIDATION; SEE ABOVE***')
        if not args.verbose:
            get_logger().info('(For more error details, re-run with the "-v" flag)')
        parser.exit(status=1)
    else:
        get_logger().info('Passed validation (%d patches total)', len(series_iterable))


if __name__ == '__main__':
    main()



File: devutils/run_other_pylint.py

#!/usr/bin/env python3

# Copyright (c) 2019 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Run Pylint over any module"""

import argparse
import os
import shutil
import sys
from pathlib import Path

from pylint import lint


class ChangeDir:
    """
    Changes directory to path in with statement
    """
    def __init__(self, path):
        self._path = path
        self._orig_path = os.getcwd()

    def __enter__(self):
        os.chdir(str(self._path))

    def __exit__(self, *_):
        os.chdir(self._orig_path)


def run_pylint(module_path, pylint_options, ignore_prefixes=tuple()):
    """Runs Pylint. Returns a boolean indicating success"""
    pylint_stats = Path('/run/user/{}/pylint_stats'.format(os.getuid()))
    if not pylint_stats.parent.is_dir(): #pylint: disable=no-member
        pylint_stats = Path('/run/shm/pylint_stats')
    os.environ['PYLINTHOME'] = str(pylint_stats)

    input_paths = list()
    if not module_path.exists():
        print('ERROR: Cannot find', module_path)
        sys.exit(1)
    if module_path.is_dir():
        for path in module_path.rglob('*.py'):
            ignore_matched = False
            for prefix in ignore_prefixes:
                if path.parts[:len(prefix)] == prefix:
                    ignore_matched = True
                    break
            if ignore_matched:
                continue
            input_paths.append(str(path))
    else:
        input_paths.append(str(module_path))
    runner = lint.Run((*input_paths, *pylint_options), do_exit=False)

    if pylint_stats.is_dir():
        shutil.rmtree(str(pylint_stats))

    if runner.linter.msg_status != 0:
        print('WARNING: Non-zero exit status:', runner.linter.msg_status)
        return False
    return True


def main():
    """CLI entrypoint"""

    parser = argparse.ArgumentParser(description='Run Pylint over arbitrary module')
    parser.add_argument('--hide-fixme', action='store_true', help='Hide "fixme" Pylint warnings.')
    parser.add_argument('--show-locally-disabled',
                        action='store_true',
                        help='Show "locally-disabled" Pylint warnings.')
    parser.add_argument('module_path', type=Path, help='Path to the module to check')
    args = parser.parse_args()

    if not args.module_path.exists():
        print('ERROR: Module path "{}" does not exist'.format(args.module_path))
        sys.exit(1)

    disables = [
        'wrong-import-position',
        'bad-continuation',
    ]

    if args.hide_fixme:
        disables.append('fixme')
    if not args.show_locally_disabled:
        disables.append('locally-disabled')

    pylint_options = [
        '--disable={}'.format(','.join(disables)),
        '--jobs=4',
        '--score=n',
        '--persistent=n',
    ]

    if not run_pylint(args.module_path, pylint_options):
        sys.exit(1)
    sys.exit(0)


if __name__ == '__main__':
    main()



File: devutils/update_lists.py

#!/usr/bin/env python3

# Copyright (c) 2019 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""
Update binary pruning and domain substitution lists automatically.

It will download and unpack into the source tree as necessary.
No binary pruning or domain substitution will be applied to the source tree after
the process has finished.
"""

import argparse
import os
import sys

from itertools import repeat
from multiprocessing import Pool
from pathlib import Path, PurePosixPath

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / 'utils'))
from _common import get_logger
from domain_substitution import DomainRegexList, TREE_ENCODINGS
from prune_binaries import CONTINGENT_PATHS
sys.path.pop(0)

# Encoding for output files
_ENCODING = 'UTF-8'

# NOTE: Include patterns have precedence over exclude patterns
# pathlib.Path.match() paths to include in binary pruning
PRUNING_INCLUDE_PATTERNS = [
    'components/domain_reliability/baked_in_configs/*',
    # Removals for patches/core/ungoogled-chromium/remove-unused-preferences-fields.patch
    'components/safe_browsing/core/common/safe_browsing_prefs.cc',
    'components/safe_browsing/core/common/safe_browsing_prefs.h',
    'components/signin/public/base/signin_pref_names.cc',
    'components/signin/public/base/signin_pref_names.h',
]

# pathlib.Path.match() paths to exclude from binary pruning
PRUNING_EXCLUDE_PATTERNS = [
    'chrome/common/win/eventlog_messages.mc', # TODO: False positive textfile
    # Exclusions for DOM distiller (contains model data only)
    'components/dom_distiller/core/data/distillable_page_model_new.bin',
    'components/dom_distiller/core/data/long_page_model.bin',
    # Exclusions for GeoLanguage data
    # Details: https://docs.google.com/document/d/18WqVHz5F9vaUiE32E8Ge6QHmku2QSJKvlqB9JjnIM-g/edit
    # Introduced with: https://chromium.googlesource.com/chromium/src/+/6647da61
    'components/language/content/browser/ulp_language_code_locator/geolanguage-data_rank0.bin',
    'components/language/content/browser/ulp_language_code_locator/geolanguage-data_rank1.bin',
    'components/language/content/browser/ulp_language_code_locator/geolanguage-data_rank2.bin',
    # Exclusion for required prebuilt object for Windows arm64 builds
    'third_party/crashpad/crashpad/util/misc/capture_context_win_arm64.obj',
    'third_party/icu/common/icudtl.dat', # Exclusion for ICU data
    # Exclusion for Android
    'build/android/chromium-debug.keystore',
    'third_party/icu/android/icudtl.dat',
    'third_party/icu/common/icudtb.dat',
    # Exclusion for performance tracing
    'third_party/perfetto/src/trace_processor/importers/proto/atoms.descriptor',
    # Exclusions for safe file extensions
    '*.avif',
    '*.ttf',
    '*.png',
    '*.jpg',
    '*.webp',
    '*.gif',
    '*.ico',
    '*.mp3',
    '*.wav',
    '*.flac',
    '*.icns',
    '*.woff',
    '*.woff2',
    '*makefile',
    '*.profdata',
    '*.xcf',
    '*.cur',
    '*.pdf',
    '*.ai',
    '*.h',
    '*.c',
    '*.cpp',
    '*.cc',
    '*.mk',
    '*.bmp',
    '*.py',
    '*.xml',
    '*.html',
    '*.js',
    '*.json',
    '*.txt',
    '*.xtb'
]

# NOTE: Domain substitution path prefix exclusion has precedence over inclusion patterns
# Paths to exclude by prefixes of the POSIX representation for domain substitution
DOMAIN_EXCLUDE_PREFIXES = [
    'components/test/',
    'net/http/transport_security_state_static.json',
    'net/http/transport_security_state_static_pins.json',
    # Exclusions for Visual Studio Project generation with GN (PR #445)
    'tools/gn/',
    # Exclusions for files covered with other patches/unnecessary
    'components/search_engines/prepopulated_engines.json',
    'third_party/blink/renderer/core/dom/document.cc',
]

# pathlib.Path.match() patterns to include in domain substitution
DOMAIN_INCLUDE_PATTERNS = [
    '*.h', '*.hh', '*.hpp', '*.hxx', '*.cc', '*.cpp', '*.cxx', '*.c', '*.h', '*.json', '*.js',
    '*.html', '*.htm', '*.css', '*.py*', '*.grd*', '*.sql', '*.idl', '*.mk', '*.gyp*', 'makefile',
    '*.ts', '*.txt', '*.xml', '*.mm', '*.jinja*', '*.gn', '*.gni'
]

# Binary-detection constant
_TEXTCHARS = bytearray({7, 8, 9, 10, 12, 13, 27} | set(range(0x20, 0x100)) - {0x7f})


class UnusedPatterns: #pylint: disable=too-few-public-methods
    """Tracks unused prefixes and patterns"""

    _all_names = ('pruning_include_patterns', 'pruning_exclude_patterns', 'domain_include_patterns',
                  'domain_exclude_prefixes')

    def __init__(self):
        # Initialize all tracked patterns and prefixes in sets
        # Users will discard elements that are used
        for name in self._all_names:
            setattr(self, name, set(globals()[name.upper()]))

    def log_unused(self, error=True):
        """
        Logs unused patterns and prefixes

        Returns True if there are unused patterns or prefixes; False otherwise
        """
        have_unused = False
        log = get_logger().error if error else get_logger().info
        for name in self._all_names:
            current_set = getattr(self, name, None)
            if current_set:
                log('Unused from %s: %s', name.upper(), current_set)
                have_unused = True
        return have_unused


def _is_binary(bytes_data):
    """
    Returns True if the data seems to be binary data (i.e. not human readable); False otherwise
    """
    # From: https://stackoverflow.com/a/7392391
    return bool(bytes_data.translate(None, _TEXTCHARS))


def _dir_empty(path):
    """
    Returns True if the directory is empty; False otherwise

    path is a pathlib.Path or string to a directory to test.
    """
    try:
        next(os.scandir(str(path)))
    except StopIteration:
        return True
    return False


def should_prune(path, relative_path, used_pep_set, used_pip_set):
    """
    Returns True if a path should be pruned from the source tree; False otherwise

    path is the pathlib.Path to the file from the current working directory.
    relative_path is the pathlib.Path to the file from the source tree
    used_pep_set is a list of PRUNING_EXCLUDE_PATTERNS that have been matched
    used_pip_set is a list of PRUNING_INCLUDE_PATTERNS that have been matched
    """
    # Match against include patterns
    for pattern in filter(relative_path.match, PRUNING_INCLUDE_PATTERNS):
        used_pip_set.add(pattern)
        return True

    # Match against exclude patterns
    for pattern in filter(Path(str(relative_path).lower()).match, PRUNING_EXCLUDE_PATTERNS):
        used_pep_set.add(pattern)
        return False

    # Do binary data detection
    with path.open('rb') as file_obj:
        if _is_binary(file_obj.read()):
            return True

    # Passed all filtering; do not prune
    return False


def _check_regex_match(file_path, search_regex):
    """
    Returns True if a regex pattern matches a file; False otherwise

    file_path is a pathlib.Path to the file to test
    search_regex is a compiled regex object to search for domain names
    """
    with file_path.open("rb") as file_obj:
        file_bytes = file_obj.read()
        content = None
        for encoding in TREE_ENCODINGS:
            try:
                content = file_bytes.decode(encoding)
                break
            except UnicodeDecodeError:
                continue
        if not search_regex.search(content) is None:
            return True
    return False


def should_domain_substitute(path, relative_path, search_regex, used_dep_set, used_dip_set):
    """
    Returns True if a path should be domain substituted in the source tree; False otherwise

    path is the pathlib.Path to the file from the current working directory.
    relative_path is the pathlib.Path to the file from the source tree.
    used_dep_set is a list of DOMAIN_EXCLUDE_PREFIXES that have been matched
    used_dip_set is a list of DOMAIN_INCLUDE_PATTERNS that have been matched
    """
    relative_path_posix = relative_path.as_posix().lower()
    for include_pattern in DOMAIN_INCLUDE_PATTERNS:
        if PurePosixPath(relative_path_posix).match(include_pattern):
            used_dip_set.add(include_pattern)
            for exclude_prefix in DOMAIN_EXCLUDE_PREFIXES:
                if relative_path_posix.startswith(exclude_prefix):
                    used_dep_set.add(exclude_prefix)
                    return False
            return _check_regex_match(path, search_regex)
    return False


def compute_lists_proc(path, source_tree, search_regex):
    """
    Adds the path to appropriate lists to be used by compute_lists.

    path is the pathlib.Path to the file from the current working directory.
    source_tree is a pathlib.Path to the source tree
    search_regex is a compiled regex object to search for domain names
    """
    used_pep_set = set() # PRUNING_EXCLUDE_PATTERNS
    used_pip_set = set() # PRUNING_INCLUDE_PATTERNS
    used_dep_set = set() # DOMAIN_EXCLUDE_PREFIXES
    used_dip_set = set() # DOMAIN_INCLUDE_PATTERNS
    pruning_set = set()
    domain_substitution_set = set()
    symlink_set = set()
    if path.is_file():
        relative_path = path.relative_to(source_tree)
        if not any(cpath in str(relative_path.as_posix()) for cpath in CONTINGENT_PATHS):
            if path.is_symlink():
                try:
                    resolved_relative_posix = path.resolve().relative_to(source_tree).as_posix()
                    symlink_set.add((resolved_relative_posix, relative_path.as_posix()))
                except ValueError:
                    # Symlink leads out of the source tree
                    pass
            elif not any(skip in ('.git', '__pycache__', 'uc_staging') for skip in path.parts):
                try:
                    if should_prune(path, relative_path, used_pep_set, used_pip_set):
                        pruning_set.add(relative_path.as_posix())
                    elif should_domain_substitute(path, relative_path, search_regex, used_dep_set,
                                                  used_dip_set):
                        domain_substitution_set.add(relative_path.as_posix())
                except: #pylint: disable=bare-except
                    get_logger().exception('Unhandled exception while processing %s', relative_path)
    return (used_pep_set, used_pip_set, used_dep_set, used_dip_set, pruning_set,
            domain_substitution_set, symlink_set)


def compute_lists(source_tree, search_regex, processes): # pylint: disable=too-many-locals
    """
    Compute the binary pruning and domain substitution lists of the source tree.
    Returns a tuple of three items in the following order:
    1. The sorted binary pruning list
    2. The sorted domain substitution list
    3. An UnusedPatterns object

    source_tree is a pathlib.Path to the source tree
    search_regex is a compiled regex object to search for domain names
    processes is the maximum number of worker processes to create
    """
    pruning_set = set()
    domain_substitution_set = set()
    symlink_set = set() # POSIX resolved path -> set of POSIX symlink paths
    source_tree = source_tree.resolve()
    unused_patterns = UnusedPatterns()

    # Launch multiple processes iterating over the source tree
    with Pool(processes) as procpool:
        returned_data = procpool.starmap(
            compute_lists_proc,
            zip(source_tree.rglob('*'), repeat(source_tree), repeat(search_regex)))

    # Handle the returned data
    for (used_pep_set, used_pip_set, used_dep_set, used_dip_set, returned_pruning_set,
         returned_domain_sub_set, returned_symlink_set) in returned_data:
        # pragma pylint: disable=no-member
        unused_patterns.pruning_exclude_patterns.difference_update(used_pep_set)
        unused_patterns.pruning_include_patterns.difference_update(used_pip_set)
        unused_patterns.domain_exclude_prefixes.difference_update(used_dep_set)
        unused_patterns.domain_include_patterns.difference_update(used_dip_set)
        # pragma pylint: enable=no-member
        pruning_set.update(returned_pruning_set)
        domain_substitution_set.update(returned_domain_sub_set)
        symlink_set.update(returned_symlink_set)

    # Prune symlinks for pruned files
    for (resolved, symlink) in symlink_set:
        if resolved in pruning_set:
            pruning_set.add(symlink)

    return sorted(pruning_set), sorted(domain_substitution_set), unused_patterns


def main(args_list=None):
    """CLI entrypoint"""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--pruning',
                        metavar='PATH',
                        type=Path,
                        default='pruning.list',
                        help='The path to store pruning.list. Default: %(default)s')
    parser.add_argument('--domain-substitution',
                        metavar='PATH',
                        type=Path,
                        default='domain_substitution.list',
                        help='The path to store domain_substitution.list. Default: %(default)s')
    parser.add_argument('--domain-regex',
                        metavar='PATH',
                        type=Path,
                        default='domain_regex.list',
                        help='The path to domain_regex.list. Default: %(default)s')
    parser.add_argument('-t',
                        '--tree',
                        metavar='PATH',
                        type=Path,
                        required=True,
                        help='The path to the source tree to use.')
    parser.add_argument(
        '--processes',
        metavar='NUM',
        type=int,
        default=None,
        help=
        'The maximum number of worker processes to create. Defaults to the number of system CPUs.')
    parser.add_argument('--domain-exclude-prefix',
                        metavar='PREFIX',
                        type=str,
                        action='append',
                        help='Additional exclusion for domain_substitution.list.')
    parser.add_argument('--no-error-unused',
                        action='store_false',
                        dest='error_unused',
                        help='Do not treat unused patterns/prefixes as an error.')
    args = parser.parse_args(args_list)
    if args.domain_exclude_prefix is not None:
        DOMAIN_EXCLUDE_PREFIXES.extend(args.domain_exclude_prefix)
    if args.tree.exists() and not _dir_empty(args.tree):
        get_logger().info('Using existing source tree at %s', args.tree)
    else:
        get_logger().error('No source tree found. Aborting.')
        sys.exit(1)
    get_logger().info('Computing lists...')
    pruning_set, domain_substitution_set, unused_patterns = compute_lists(
        args.tree,
        DomainRegexList(args.domain_regex).search_regex, args.processes)
    with args.pruning.open('w', encoding=_ENCODING) as file_obj:
        file_obj.writelines('%s\n' % line for line in pruning_set)
    with args.domain_substitution.open('w', encoding=_ENCODING) as file_obj:
        file_obj.writelines('%s\n' % line for line in domain_substitution_set)
    if unused_patterns.log_unused(args.error_unused) and args.error_unused:
        get_logger().error('Please update or remove unused patterns and/or prefixes. '
                           'The lists have still been updated with the remaining valid entries.')
        sys.exit(1)


if __name__ == "__main__":
    main()



File: devutils/check_gn_flags.py

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-

# Copyright (c) 2019 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Run sanity checking algorithms over GN flags

It checks the following:

    * GN flags in flags.gn are sorted and not duplicated

Exit codes:
    * 0 if no problems detected
    * 1 if warnings or errors occur
"""

import argparse
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / 'utils'))
from _common import ENCODING, get_logger
sys.path.pop(0)


def check_gn_flags(gn_flags_path):
    """
    Checks if GN flags are sorted and not duplicated.

    gn_flags_path is a pathlib.Path to the GN flags file to check

    Returns True if warnings were logged; False otherwise
    """
    keys_seen = set()
    warnings = False
    with gn_flags_path.open(encoding=ENCODING) as file_obj:
        iterator = iter(file_obj.read().splitlines())
    try:
        previous = next(iterator)
    except StopIteration:
        return warnings
    for current in iterator:
        gn_key = current.split('=')[0]
        if gn_key in keys_seen:
            get_logger().warning('In GN flags %s, "%s" appears at least twice', gn_flags_path,
                                 gn_key)
            warnings = True
        else:
            keys_seen.add(gn_key)
        if current < previous:
            get_logger().warning('In GN flags %s, "%s" should be sorted before "%s"', gn_flags_path,
                                 current, previous)
            warnings = True
        previous = current
    return warnings


def main():
    """CLI entrypoint"""

    root_dir = Path(__file__).resolve().parent.parent
    default_flags_gn = root_dir / 'flags.gn'

    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('-f',
                        '--flags-gn',
                        type=Path,
                        default=default_flags_gn,
                        help='Path to the GN flags to use. Default: %(default)s')
    args = parser.parse_args()

    if check_gn_flags(args.flags_gn):
        sys.exit(1)
    sys.exit(0)


if __name__ == '__main__':
    main()



File: devutils/check_files_exist.py

#!/usr/bin/env python3

# Copyright (c) 2019 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""
Checks if files in a list exist.

Used for quick validation of lists in CI checks.
"""

import argparse
import sys
from pathlib import Path


def main():
    """CLI entrypoint"""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('root_dir', type=Path, help='The directory to check from')
    parser.add_argument('input_files', type=Path, nargs='+', help='The files lists to check')
    args = parser.parse_args()

    for input_name in args.input_files:
        file_iter = filter(
            len, map(str.strip,
                     Path(input_name).read_text(encoding='UTF-8').splitlines()))
        for file_name in file_iter:
            if not Path(args.root_dir, file_name).exists():
                print('ERROR: Path "{}" from file "{}" does not exist.'.format(
                    file_name, input_name),
                      file=sys.stderr)
                sys.exit(1)


if __name__ == "__main__":
    main()



File: devutils/__init__.py




File: devutils/check_downloads_ini.py

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-

# Copyright (c) 2019 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Run sanity checking algorithms over downloads.ini files

It checks the following:

    * downloads.ini has the correct format (i.e. conforms to its schema)

Exit codes:
    * 0 if no problems detected
    * 1 if warnings or errors occur
"""

import argparse
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / 'utils'))
from downloads import DownloadInfo, schema
sys.path.pop(0)


def check_downloads_ini(downloads_ini_iter):
    """
    Combines and checks if the the downloads.ini files provided are valid.

    downloads_ini_iter must be an iterable of strings to downloads.ini files.

    Returns True if errors occured, False otherwise.
    """
    try:
        DownloadInfo(downloads_ini_iter)
    except schema.SchemaError:
        return True
    return False


def main():
    """CLI entrypoint"""

    root_dir = Path(__file__).resolve().parent.parent
    default_downloads_ini = [str(root_dir / 'downloads.ini')]

    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('-d',
                        '--downloads-ini',
                        type=Path,
                        nargs='*',
                        default=default_downloads_ini,
                        help='List of downloads.ini files to check. Default: %(default)s')
    args = parser.parse_args()

    if check_downloads_ini(args.downloads_ini):
        sys.exit(1)
    sys.exit(0)


if __name__ == '__main__':
    main()



File: devutils/check_patch_files.py

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-

# Copyright (c) 2019 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Run sanity checking algorithms over ungoogled-chromium's patch files

It checks the following:

    * All patches exist
    * All patches are referenced by the patch order

Exit codes:
    * 0 if no problems detected
    * 1 if warnings or errors occur
"""

import argparse
import sys
from pathlib import Path

from third_party import unidiff

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / 'utils'))
from _common import ENCODING, get_logger, parse_series # pylint: disable=wrong-import-order
sys.path.pop(0)

# File suffixes to ignore for checking unused patches
_PATCHES_IGNORE_SUFFIXES = {'.md'}


def _read_series_file(patches_dir, series_file, join_dir=False):
    """
    Returns a generator over the entries in the series file

    patches_dir is a pathlib.Path to the directory of patches
    series_file is a pathlib.Path relative to patches_dir

    join_dir indicates if the patches_dir should be joined with the series entries
    """
    for entry in parse_series(patches_dir / series_file):
        if join_dir:
            yield patches_dir / entry
        else:
            yield entry


def check_patch_readability(patches_dir, series_path=Path('series')):
    """
    Check if the patches from iterable patch_path_iter are readable.
        Patches that are not are logged to stdout.

    Returns True if warnings occured, False otherwise.
    """
    warnings = False
    for patch_path in _read_series_file(patches_dir, series_path, join_dir=True):
        if patch_path.exists():
            with patch_path.open(encoding=ENCODING) as file_obj:
                try:
                    unidiff.PatchSet(file_obj.read())
                except unidiff.errors.UnidiffParseError:
                    get_logger().exception('Could not parse patch: %s', patch_path)
                    warnings = True
                    continue
        else:
            get_logger().warning('Patch not found: %s', patch_path)
            warnings = True
    return warnings


def check_unused_patches(patches_dir, series_path=Path('series')):
    """
    Checks if there are unused patches in patch_dir from series file series_path.
        Unused patches are logged to stdout.

    patches_dir is a pathlib.Path to the directory of patches
    series_path is a pathlib.Path to the series file relative to the patches_dir

    Returns True if there are unused patches; False otherwise.
    """
    unused_patches = set()
    for path in patches_dir.rglob('*'):
        if path.is_dir():
            continue
        if path.suffix in _PATCHES_IGNORE_SUFFIXES:
            continue
        unused_patches.add(str(path.relative_to(patches_dir)))
    unused_patches -= set(_read_series_file(patches_dir, series_path))
    unused_patches.remove(str(series_path))
    logger = get_logger()
    for entry in sorted(unused_patches):
        logger.warning('Unused patch: %s', entry)
    return bool(unused_patches)


def check_series_duplicates(patches_dir, series_path=Path('series')):
    """
    Checks if there are duplicate entries in the series file

    series_path is a pathlib.Path to the series file relative to the patches_dir

    returns True if there are duplicate entries; False otherwise.
    """
    entries_seen = set()
    for entry in _read_series_file(patches_dir, series_path):
        if entry in entries_seen:
            get_logger().warning('Patch appears more than once in series: %s', entry)
            return True
        entries_seen.add(entry)
    return False


def main():
    """CLI entrypoint"""

    root_dir = Path(__file__).resolve().parent.parent
    default_patches_dir = root_dir / 'patches'

    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('-p',
                        '--patches',
                        type=Path,
                        default=default_patches_dir,
                        help='Path to the patches directory to use. Default: %(default)s')
    args = parser.parse_args()

    warnings = False
    warnings |= check_patch_readability(args.patches)
    warnings |= check_series_duplicates(args.patches)
    warnings |= check_unused_patches(args.patches)

    if warnings:
        sys.exit(1)
    sys.exit(0)


if __name__ == '__main__':
    main()



File: devutils/update_platform_patches.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Copyright (c) 2019 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""
Utility to ease the updating of platform patches against ungoogled-chromium's patches
"""

import argparse
import os
import shutil
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / 'utils'))
from _common import ENCODING, get_logger
from patches import merge_patches
sys.path.pop(0)

_SERIES = 'series'
_SERIES_ORIG = 'series.orig'
_SERIES_PREPEND = 'series.prepend'
_SERIES_MERGED = 'series.merged'


def merge_platform_patches(platform_patches_dir, prepend_patches_dir):
    '''
    Prepends prepend_patches_dir into platform_patches_dir

    Returns True if successful, False otherwise
    '''
    if not (platform_patches_dir / _SERIES).exists():
        get_logger().error('Unable to find platform series file: %s',
                           platform_patches_dir / _SERIES)
        return False

    # Make series.orig file
    shutil.copyfile(str(platform_patches_dir / _SERIES), str(platform_patches_dir / _SERIES_ORIG))

    # Make series.prepend
    shutil.copyfile(str(prepend_patches_dir / _SERIES), str(platform_patches_dir / _SERIES_PREPEND))

    # Merge patches
    merge_patches([prepend_patches_dir], platform_patches_dir, prepend=True)
    (platform_patches_dir / _SERIES).replace(platform_patches_dir / _SERIES_MERGED)

    return True


def _dir_empty(path):
    '''
    Returns True if the directory exists and is empty; False otherwise
    '''
    try:
        next(os.scandir(str(path)))
    except StopIteration:
        return True
    except FileNotFoundError:
        pass
    return False


def _remove_files_with_dirs(root_dir, sorted_file_iter):
    '''
    Deletes a list of sorted files relative to root_dir, removing empty directories along the way
    '''
    past_parent = None
    for partial_path in sorted_file_iter:
        complete_path = Path(root_dir, partial_path)
        try:
            complete_path.unlink()
        except FileNotFoundError:
            get_logger().warning('Could not remove prepended patch: %s', complete_path)
        if past_parent != complete_path.parent:
            while past_parent and _dir_empty(past_parent):
                past_parent.rmdir()
                past_parent = past_parent.parent
            past_parent = complete_path.parent
    # Handle last path's directory
    while _dir_empty(complete_path.parent):
        complete_path.parent.rmdir()
        complete_path = complete_path.parent


def unmerge_platform_patches(platform_patches_dir):
    '''
    Undo merge_platform_patches(), adding any new patches from series.merged as necessary

    Returns True if successful, False otherwise
    '''
    if not (platform_patches_dir / _SERIES_PREPEND).exists():
        get_logger().error('Unable to find series.prepend at: %s',
                           platform_patches_dir / _SERIES_PREPEND)
        return False
    prepend_series = set(
        filter(len,
               (platform_patches_dir / _SERIES_PREPEND).read_text(encoding=ENCODING).splitlines()))

    # Remove prepended files with directories
    _remove_files_with_dirs(platform_patches_dir, sorted(prepend_series))

    # Determine positions of blank spaces in series.orig
    if not (platform_patches_dir / _SERIES_ORIG).exists():
        get_logger().error('Unable to find series.orig at: %s', platform_patches_dir / _SERIES_ORIG)
        return False
    orig_series = (platform_patches_dir / _SERIES_ORIG).read_text(encoding=ENCODING).splitlines()
    # patch path -> list of lines after patch path and before next patch path
    path_comments = dict()
    # patch path -> inline comment for patch
    path_inline_comments = dict()
    previous_path = None
    for partial_path in orig_series:
        if not partial_path or partial_path.startswith('#'):
            if partial_path not in path_comments:
                path_comments[previous_path] = list()
            path_comments[previous_path].append(partial_path)
        else:
            path_parts = partial_path.split(' #', maxsplit=1)
            previous_path = path_parts[0]
            if len(path_parts) == 2:
                path_inline_comments[path_parts[0]] = path_parts[1]

    # Apply changes on series.merged into a modified version of series.orig
    if not (platform_patches_dir / _SERIES_MERGED).exists():
        get_logger().error('Unable to find series.merged at: %s',
                           platform_patches_dir / _SERIES_MERGED)
        return False
    new_series = filter(len, (platform_patches_dir /
                              _SERIES_MERGED).read_text(encoding=ENCODING).splitlines())
    new_series = filter((lambda x: x not in prepend_series), new_series)
    new_series = list(new_series)
    series_index = 0
    while series_index < len(new_series):
        current_path = new_series[series_index]
        if current_path in path_inline_comments:
            new_series[series_index] = current_path + ' #' + path_inline_comments[current_path]
        if current_path in path_comments:
            new_series.insert(series_index + 1, '\n'.join(path_comments[current_path]))
            series_index += 1
        series_index += 1

    # Write series file
    with (platform_patches_dir / _SERIES).open('w', encoding=ENCODING) as series_file:
        series_file.write('\n'.join(new_series))
        series_file.write('\n')

    # All other operations are successful; remove merging intermediates
    (platform_patches_dir / _SERIES_MERGED).unlink()
    (platform_patches_dir / _SERIES_ORIG).unlink()
    (platform_patches_dir / _SERIES_PREPEND).unlink()

    return True


def main():
    """CLI Entrypoint"""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('command',
                        choices=('merge', 'unmerge'),
                        help='Merge or unmerge ungoogled-chromium patches with platform patches')
    parser.add_argument('platform_patches',
                        type=Path,
                        help='The path to the platform patches in GNU Quilt format to merge into')
    args = parser.parse_args()

    repo_dir = Path(__file__).resolve().parent.parent

    success = False
    if args.command == 'merge':
        success = merge_platform_patches(args.platform_patches, repo_dir / 'patches')
    elif args.command == 'unmerge':
        success = unmerge_platform_patches(args.platform_patches)
    else:
        raise NotImplementedError(args.command)

    if success:
        return 0
    return 1


if __name__ == '__main__':
    sys.exit(main())



File: devutils/run_devutils_pylint.py

#!/usr/bin/env python3

# Copyright (c) 2019 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Run Pylint over devutils"""

import argparse
import sys
from pathlib import Path

from run_other_pylint import ChangeDir, run_pylint


def main():
    """CLI entrypoint"""
    parser = argparse.ArgumentParser(description='Run Pylint over devutils')
    parser.add_argument('--hide-fixme', action='store_true', help='Hide "fixme" Pylint warnings.')
    parser.add_argument('--show-locally-disabled',
                        action='store_true',
                        help='Show "locally-disabled" Pylint warnings.')
    args = parser.parse_args()

    disables = [
        'wrong-import-position',
        'bad-continuation',
        'duplicate-code',
    ]

    if args.hide_fixme:
        disables.append('fixme')
    if not args.show_locally_disabled:
        disables.append('locally-disabled')

    pylint_options = [
        '--disable={}'.format(','.join(disables)),
        '--jobs=4',
        '--score=n',
        '--persistent=n',
    ]

    ignore_prefixes = [
        ('third_party', ),
    ]

    sys.path.insert(1, str(Path(__file__).resolve().parent.parent / 'utils'))
    sys.path.insert(2, str(Path(__file__).resolve().parent.parent / 'devutils' / 'third_party'))
    with ChangeDir(Path(__file__).parent):
        result = run_pylint(
            Path(),
            pylint_options,
            ignore_prefixes=ignore_prefixes,
        )
    sys.path.pop(2)
    sys.path.pop(1)
    if not result:
        sys.exit(1)
    sys.exit(0)


if __name__ == '__main__':
    main()



File: devutils/validate_config.py

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-

# Copyright (c) 2019 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Run sanity checking algorithms over ungoogled-chromium's config files

NOTE: This script is hardcoded to run over ungoogled-chromium's config files only.
To check other files, use the other scripts imported by this script.

It checks the following:

    * All patches exist
    * All patches are referenced by the patch order
    * Each patch is used only once
    * GN flags in flags.gn are sorted and not duplicated
    * downloads.ini has the correct format (i.e. conforms to its schema)

Exit codes:
    * 0 if no problems detected
    * 1 if warnings or errors occur
"""

import sys
from pathlib import Path

from check_downloads_ini import check_downloads_ini
from check_gn_flags import check_gn_flags
from check_patch_files import (check_patch_readability, check_series_duplicates,
                               check_unused_patches)


def main():
    """CLI entrypoint"""

    warnings = False
    root_dir = Path(__file__).resolve().parent.parent
    patches_dir = root_dir / 'patches'

    # Check patches
    warnings |= check_patch_readability(patches_dir)
    warnings |= check_series_duplicates(patches_dir)
    warnings |= check_unused_patches(patches_dir)

    # Check GN flags
    warnings |= check_gn_flags(root_dir / 'flags.gn')

    # Check downloads.ini
    warnings |= check_downloads_ini([root_dir / 'downloads.ini'])

    if warnings:
        sys.exit(1)
    sys.exit(0)


if __name__ == '__main__':
    if sys.argv[1:]:
        print(__doc__)
    else:
        main()



File: devutils/run_utils_pylint.py

#!/usr/bin/env python3

# Copyright (c) 2019 The ungoogled-chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Run Pylint over utils"""

import argparse
import sys
from pathlib import Path

from run_other_pylint import ChangeDir, run_pylint


def main():
    """CLI entrypoint"""
    parser = argparse.ArgumentParser(description='Run Pylint over utils')
    parser.add_argument('--hide-fixme', action='store_true', help='Hide "fixme" Pylint warnings.')
    parser.add_argument('--show-locally-disabled',
                        action='store_true',
                        help='Show "locally-disabled" Pylint warnings.')
    args = parser.parse_args()

    disable = ['bad-continuation']

    if args.hide_fixme:
        disable.append('fixme')
    if not args.show_locally_disabled:
        disable.append('locally-disabled')

    pylint_options = [
        '--disable={}'.format(','.join(disable)),
        '--jobs=4',
        '--max-args=6',
        '--score=n',
        '--persistent=n',
    ]

    ignore_prefixes = [
        ('third_party', ),
        ('tests', ),
    ]

    sys.path.insert(1, str(Path(__file__).resolve().parent.parent / 'utils' / 'third_party'))
    sys.path.append(Path(__file__).resolve().parent.parent / 'utils')
    with ChangeDir(Path(__file__).resolve().parent.parent / 'utils'):
        result = run_pylint(
            Path(),
            pylint_options,
            ignore_prefixes=ignore_prefixes,
        )
    sys.path.pop(1)
    if not result:
        sys.exit(1)
    sys.exit(0)


if __name__ == '__main__':
    main()



File: devutils/third_party/__init__.py




File: devutils/third_party/unidiff/patch.py

# -*- coding: utf-8 -*-

# The MIT License (MIT)
# Copyright (c) 2014-2017 Matias Bordese
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
# DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE
# OR OTHER DEALINGS IN THE SOFTWARE.


"""Classes used by the unified diff parser to keep the diff data."""

from __future__ import unicode_literals

import codecs
import sys

from .constants import (
    DEFAULT_ENCODING,
    LINE_TYPE_ADDED,
    LINE_TYPE_CONTEXT,
    LINE_TYPE_EMPTY,
    LINE_TYPE_REMOVED,
    LINE_TYPE_NO_NEWLINE,
    LINE_VALUE_NO_NEWLINE,
    RE_HUNK_BODY_LINE,
    RE_HUNK_EMPTY_BODY_LINE,
    RE_HUNK_HEADER,
    RE_SOURCE_FILENAME,
    RE_TARGET_FILENAME,
    RE_NO_NEWLINE_MARKER,
)
from .errors import UnidiffParseError


PY2 = sys.version_info[0] == 2
if PY2:
    from StringIO import StringIO
    open_file = codecs.open
    make_str = lambda x: x.encode(DEFAULT_ENCODING)

    def implements_to_string(cls):
        cls.__unicode__ = cls.__str__
        cls.__str__ = lambda x: x.__unicode__().encode(DEFAULT_ENCODING)
        return cls
else:
    from io import StringIO
    open_file = open
    make_str = str
    implements_to_string = lambda x: x
    unicode = str
    basestring = str


@implements_to_string
class Line(object):
    """A diff line."""

    def __init__(self, value, line_type,
                 source_line_no=None, target_line_no=None, diff_line_no=None):
        super(Line, self).__init__()
        self.source_line_no = source_line_no
        self.target_line_no = target_line_no
        self.diff_line_no = diff_line_no
        self.line_type = line_type
        self.value = value

    def __repr__(self):
        return make_str("<Line: %s%s>") % (self.line_type, self.value)

    def __str__(self):
        return "%s%s" % (self.line_type, self.value)

    def __eq__(self, other):
        return (self.source_line_no == other.source_line_no and
                self.target_line_no == other.target_line_no and
                self.diff_line_no == other.diff_line_no and
                self.line_type == other.line_type and
                self.value == other.value)

    @property
    def is_added(self):
        return self.line_type == LINE_TYPE_ADDED

    @property
    def is_removed(self):
        return self.line_type == LINE_TYPE_REMOVED

    @property
    def is_context(self):
        return self.line_type == LINE_TYPE_CONTEXT


@implements_to_string
class PatchInfo(list):
    """Lines with extended patch info.

    Format of this info is not documented and it very much depends on
    patch producer.

    """

    def __repr__(self):
        value = "<PatchInfo: %s>" % self[0].strip()
        return make_str(value)

    def __str__(self):
        return ''.join(unicode(line) for line in self)


@implements_to_string
class Hunk(list):
    """Each of the modified blocks of a file."""

    def __init__(self, src_start=0, src_len=0, tgt_start=0, tgt_len=0,
                 section_header=''):
        if src_len is None:
            src_len = 1
        if tgt_len is None:
            tgt_len = 1
        self.added = 0  # number of added lines
        self.removed = 0  # number of removed lines
        self.source = []
        self.source_start = int(src_start)
        self.source_length = int(src_len)
        self.target = []
        self.target_start = int(tgt_start)
        self.target_length = int(tgt_len)
        self.section_header = section_header

    def __repr__(self):
        value = "<Hunk: @@ %d,%d %d,%d @@ %s>" % (self.source_start,
                                                  self.source_length,
                                                  self.target_start,
                                                  self.target_length,
                                                  self.section_header)
        return make_str(value)

    def __str__(self):
        # section header is optional and thus we output it only if it's present
        head = "@@ -%d,%d +%d,%d @@%s\n" % (
            self.source_start, self.source_length,
            self.target_start, self.target_length,
            ' ' + self.section_header if self.section_header else '')
        content = ''.join(unicode(line) for line in self)
        return head + content

    def append(self, line):
        """Append the line to hunk, and keep track of source/target lines."""
        super(Hunk, self).append(line)
        s = str(line)
        if line.is_added:
            self.added += 1
            self.target.append(s)
        elif line.is_removed:
            self.removed += 1
            self.source.append(s)
        elif line.is_context:
            self.target.append(s)
            self.source.append(s)

    def is_valid(self):
        """Check hunk header data matches entered lines info."""
        return (len(self.source) == self.source_length and
                len(self.target) == self.target_length)

    def source_lines(self):
        """Hunk lines from source file (generator)."""
        return (l for l in self if l.is_context or l.is_removed)

    def target_lines(self):
        """Hunk lines from target file (generator)."""
        return (l for l in self if l.is_context or l.is_added)


class PatchedFile(list):
    """Patch updated file, it is a list of Hunks."""

    def __init__(self, patch_info=None, source='', target='',
                 source_timestamp=None, target_timestamp=None):
        super(PatchedFile, self).__init__()
        self.patch_info = patch_info
        self.source_file = source
        self.source_timestamp = source_timestamp
        self.target_file = target
        self.target_timestamp = target_timestamp

    def __repr__(self):
        return make_str("<PatchedFile: %s>") % make_str(self.path)

    def __str__(self):
        # patch info is optional
        info = '' if self.patch_info is None else str(self.patch_info)
        source = "--- %s%s\n" % (
            self.source_file,
            '\t' + self.source_timestamp if self.source_timestamp else '')
        target = "+++ %s%s\n" % (
            self.target_file,
            '\t' + self.target_timestamp if self.target_timestamp else '')
        hunks = ''.join(unicode(hunk) for hunk in self)
        return info + source + target + hunks

    def _parse_hunk(self, header, diff, encoding):
        """Parse hunk details."""
        header_info = RE_HUNK_HEADER.match(header)
        hunk_info = header_info.groups()
        hunk = Hunk(*hunk_info)

        source_line_no = hunk.source_start
        target_line_no = hunk.target_start
        expected_source_end = source_line_no + hunk.source_length
        expected_target_end = target_line_no + hunk.target_length

        for diff_line_no, line in diff:
            if encoding is not None:
                line = line.decode(encoding)

            valid_line = RE_HUNK_EMPTY_BODY_LINE.match(line)
            if not valid_line:
                valid_line = RE_HUNK_BODY_LINE.match(line)

            if not valid_line:
                raise UnidiffParseError('Hunk diff line expected: %s' % line)

            line_type = valid_line.group('line_type')
            if line_type == LINE_TYPE_EMPTY:
                line_type = LINE_TYPE_CONTEXT
            value = valid_line.group('value')
            original_line = Line(value, line_type=line_type)
            if line_type == LINE_TYPE_ADDED:
                original_line.target_line_no = target_line_no
                target_line_no += 1
            elif line_type == LINE_TYPE_REMOVED:
                original_line.source_line_no = source_line_no
                source_line_no += 1
            elif line_type == LINE_TYPE_CONTEXT:
                original_line.target_line_no = target_line_no
                target_line_no += 1
                original_line.source_line_no = source_line_no
                source_line_no += 1
            elif line_type == LINE_TYPE_NO_NEWLINE:
                pass
            else:
                original_line = None

            # stop parsing if we got past expected number of lines
            if (source_line_no > expected_source_end or
                    target_line_no > expected_target_end):
                raise UnidiffParseError('Hunk is longer than expected')

            if original_line:
                original_line.diff_line_no = diff_line_no
                hunk.append(original_line)

            # if hunk source/target lengths are ok, hunk is complete
            if (source_line_no == expected_source_end and
                    target_line_no == expected_target_end):
                break

        # report an error if we haven't got expected number of lines
        if (source_line_no < expected_source_end or
                target_line_no < expected_target_end):
            raise UnidiffParseError('Hunk is shorter than expected')

        self.append(hunk)

    def _add_no_newline_marker_to_last_hunk(self):
        if not self:
            raise UnidiffParseError(
                'Unexpected marker:' + LINE_VALUE_NO_NEWLINE)
        last_hunk = self[-1]
        last_hunk.append(
            Line(LINE_VALUE_NO_NEWLINE + '\n', line_type=LINE_TYPE_NO_NEWLINE))

    def _append_trailing_empty_line(self):
        if not self:
            raise UnidiffParseError('Unexpected trailing newline character')
        last_hunk = self[-1]
        last_hunk.append(Line('\n', line_type=LINE_TYPE_EMPTY))

    @property
    def path(self):
        """Return the file path abstracted from VCS."""
        if (self.source_file.startswith('a/') and
                self.target_file.startswith('b/')):
            filepath = self.source_file[2:]
        elif (self.source_file.startswith('a/') and
              self.target_file == '/dev/null'):
            filepath = self.source_file[2:]
        elif (self.target_file.startswith('b/') and
              self.source_file == '/dev/null'):
            filepath = self.target_file[2:]
        else:
            filepath = self.source_file
        return filepath

    @property
    def added(self):
        """Return the file total added lines."""
        return sum([hunk.added for hunk in self])

    @property
    def removed(self):
        """Return the file total removed lines."""
        return sum([hunk.removed for hunk in self])

    @property
    def is_added_file(self):
        """Return True if this patch adds the file."""
        return (len(self) == 1 and self[0].source_start == 0 and
                self[0].source_length == 0)

    @property
    def is_removed_file(self):
        """Return True if this patch removes the file."""
        return (len(self) == 1 and self[0].target_start == 0 and
                self[0].target_length == 0)

    @property
    def is_modified_file(self):
        """Return True if this patch modifies the file."""
        return not (self.is_added_file or self.is_removed_file)


@implements_to_string
class PatchSet(list):
    """A list of PatchedFiles."""

    def __init__(self, f, encoding=None):
        super(PatchSet, self).__init__()

        # convert string inputs to StringIO objects
        if isinstance(f, basestring):
            f = self._convert_string(f, encoding)

        # make sure we pass an iterator object to parse
        data = iter(f)
        # if encoding is None, assume we are reading unicode data
        self._parse(data, encoding=encoding)

    def __repr__(self):
        return make_str('<PatchSet: %s>') % super(PatchSet, self).__repr__()

    def __str__(self):
        return ''.join(unicode(patched_file) for patched_file in self)

    def _parse(self, diff, encoding):
        current_file = None
        patch_info = None

        diff = enumerate(diff, 1)
        for unused_diff_line_no, line in diff:
            if encoding is not None:
                line = line.decode(encoding)

            # check for source file header
            is_source_filename = RE_SOURCE_FILENAME.match(line)
            if is_source_filename:
                source_file = is_source_filename.group('filename')
                source_timestamp = is_source_filename.group('timestamp')
                # reset current file
                current_file = None
                continue

            # check for target file header
            is_target_filename = RE_TARGET_FILENAME.match(line)
            if is_target_filename:
                if current_file is not None:
                    raise UnidiffParseError('Target without source: %s' % line)
                target_file = is_target_filename.group('filename')
                target_timestamp = is_target_filename.group('timestamp')
                # add current file to PatchSet
                current_file = PatchedFile(
                    patch_info, source_file, target_file,
                    source_timestamp, target_timestamp)
                self.append(current_file)
                patch_info = None
                continue

            # check for hunk header
            is_hunk_header = RE_HUNK_HEADER.match(line)
            if is_hunk_header:
                if current_file is None:
                    raise UnidiffParseError('Unexpected hunk found: %s' % line)
                current_file._parse_hunk(line, diff, encoding)
                continue

            # check for no newline marker
            is_no_newline = RE_NO_NEWLINE_MARKER.match(line)
            if is_no_newline:
                if current_file is None:
                    raise UnidiffParseError('Unexpected marker: %s' % line)
                current_file._add_no_newline_marker_to_last_hunk()
                continue

            # sometimes hunks can be followed by empty lines
            if line == '\n' and current_file is not None:
                current_file._append_trailing_empty_line()
                continue

            # if nothing has matched above then this line is a patch info
            if patch_info is None:
                current_file = None
                patch_info = PatchInfo()
            patch_info.append(line)

    @classmethod
    def from_filename(cls, filename, encoding=DEFAULT_ENCODING, errors=None):
        """Return a PatchSet instance given a diff filename."""
        with open_file(filename, 'r', encoding=encoding, errors=errors) as f:
            instance = cls(f)
        return instance

    @staticmethod
    def _convert_string(data, encoding=None, errors='strict'):
        if encoding is not None:
            # if encoding is given, assume bytes and decode
            data = unicode(data, encoding=encoding, errors=errors)
        return StringIO(data)

    @classmethod
    def from_string(cls, data, encoding=None, errors='strict'):
        """Return a PatchSet instance given a diff string."""
        return cls(cls._convert_string(data, encoding, errors))

    @property
    def added_files(self):
        """Return patch added files as a list."""
        return [f for f in self if f.is_added_file]

    @property
    def removed_files(self):
        """Return patch removed files as a list."""
        return [f for f in self if f.is_removed_file]

    @property
    def modified_files(self):
        """Return patch modified files as a list."""
        return [f for f in self if f.is_modified_file]

    @property
    def added(self):
        """Return the patch total added lines."""
        return sum([f.added for f in self])

    @property
    def removed(self):
        """Return the patch total removed lines."""
        return sum([f.removed for f in self])



File: devutils/third_party/unidiff/constants.py

# -*- coding: utf-8 -*-

# The MIT License (MIT)
# Copyright (c) 2014-2017 Matias Bordese
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
# DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE
# OR OTHER DEALINGS IN THE SOFTWARE.


"""Useful constants and regexes used by the package."""

from __future__ import unicode_literals

import re


RE_SOURCE_FILENAME = re.compile(
    r'^--- (?P<filename>[^\t\n]+)(?:\t(?P<timestamp>[^\n]+))?')
RE_TARGET_FILENAME = re.compile(
    r'^\+\+\+ (?P<filename>[^\t\n]+)(?:\t(?P<timestamp>[^\n]+))?')

# @@ (source offset, length) (target offset, length) @@ (section header)
RE_HUNK_HEADER = re.compile(
    r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))?\ @@[ ]?(.*)")

#    kept line (context)
# \n empty line (treat like context)
# +  added line
# -  deleted line
# \  No newline case
RE_HUNK_BODY_LINE = re.compile(
    r'^(?P<line_type>[- \+\\])(?P<value>.*)', re.DOTALL)
RE_HUNK_EMPTY_BODY_LINE = re.compile(
    r'^(?P<line_type>[- \+\\]?)(?P<value>[\r\n]{1,2})', re.DOTALL)

RE_NO_NEWLINE_MARKER = re.compile(r'^\\ No newline at end of file')

DEFAULT_ENCODING = 'UTF-8'

LINE_TYPE_ADDED = '+'
LINE_TYPE_REMOVED = '-'
LINE_TYPE_CONTEXT = ' '
LINE_TYPE_EMPTY = ''
LINE_TYPE_NO_NEWLINE = '\\'
LINE_VALUE_NO_NEWLINE = ' No newline at end of file'



File: devutils/third_party/unidiff/__init__.py

# -*- coding: utf-8 -*-

# The MIT License (MIT)
# Copyright (c) 2014-2017 Matias Bordese
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
# DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE
# OR OTHER DEALINGS IN THE SOFTWARE.


"""Unidiff parsing library."""

from __future__ import unicode_literals

from . import __version__
from .patch import (
    DEFAULT_ENCODING,
    LINE_TYPE_ADDED,
    LINE_TYPE_CONTEXT,
    LINE_TYPE_REMOVED,
    Hunk,
    PatchedFile,
    PatchSet,
    UnidiffParseError,
)

VERSION = __version__.__version__



File: devutils/third_party/unidiff/__version__.py

# -*- coding: utf-8 -*-

# The MIT License (MIT)
# Copyright (c) 2014-2017 Matias Bordese
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
# DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE
# OR OTHER DEALINGS IN THE SOFTWARE.

__version__ = '0.5.5'



File: devutils/third_party/unidiff/errors.py

# -*- coding: utf-8 -*-

# The MIT License (MIT)
# Copyright (c) 2014-2017 Matias Bordese
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
# DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
# OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE
# OR OTHER DEALINGS IN THE SOFTWARE.


"""Errors and exceptions raised by the package."""

from __future__ import unicode_literals


class UnidiffParseError(Exception):
    """Exception when parsing the unified diff data."""



# ungoogled-chromium

*A lightweight approach to removing Google web service dependency*

**Help is welcome!** See the [docs/contributing.md](docs/contributing.md) document for more information.

## Objectives

In descending order of significance (i.e. most important objective first):

1. **ungoogled-chromium is Google Chromium, sans dependency on Google web services**.
2. **ungoogled-chromium retains the default Chromium experience as closely as possible**. Unlike other Chromium forks that have their own visions of a web browser, ungoogled-chromium is essentially a drop-in replacement for Chromium.
3. **ungoogled-chromium features tweaks to enhance privacy, control, and transparency**. However, almost all of these features must be manually activated or enabled. For more details, see [Feature Overview](#feature-overview).

In scenarios where the objectives conflict, the objective of higher significance should take precedence.

## Content Overview

* [Objectives](#objectives)
* [Motivation and Philosophy](#motivation-and-philosophy)
* [Feature Overview](#feature-overview)
* [**Downloads**](#downloads)
* [Source Code](#source-code)
* [**FAQ**](#faq)
* [Building Instructions](#building-instructions)
* [Design Documentation](#design-documentation)
* [**Contributing, Reporting, Contacting**](#contributing-reporting-contacting)
* [Credits](#credits)
* [Related Projects](#related-projects)
* [License](#license)

## Motivation and Philosophy

Without signing in to a Google Account, Chromium does pretty well in terms of security and privacy. However, Chromium still has some dependency on Google web services and binaries. In addition, Google designed Chromium to be easy and intuitive for users, which means they compromise on transparency and control of internal operations.

ungoogled-chromium addresses these issues in the following ways:

1. Remove all remaining background requests to any web services while building and running the browser
2. Remove all code specific to Google web services
3. Remove all uses of pre-made binaries from the source code, and replace them with user-provided alternatives when possible.
4. Disable features that inhibit control and transparency, and add or modify features that promote them (these changes will almost always require manual activation or enabling).

These features are implemented as configuration flags, patches, and custom scripts. For more details, consult the [Design Documentation](docs/design.md).

## Feature Overview

*This section overviews the features of ungoogled-chromium. For more detailed information, it is best to consult the source code.*

Contents of this section:

* [Key Features](#key-features)
* [Enhancing Features](#enhancing-features)
* [Borrowed Features](#borrowed-features)
* [Supported Platforms and Distributions](#supported-platforms-and-distributions)

### Key Features

*These are the core features introduced by ungoogled-chromium.*

* Disable functionality specific to Google domains (e.g. Google Host Detector, Google URL Tracker, Google Cloud Messaging, Google Hotwording, etc.)
    * This includes disabling [Safe Browsing](https://en.wikipedia.org/wiki/Google_Safe_Browsing). Consult [the FAQ for the rationale](https://ungoogled-software.github.io/ungoogled-chromium-wiki/faq#why-is-safe-browsing-disabled).
* Block internal requests to Google at runtime. This feature is a fail-safe measure for the above, in case Google changes or introduces new components that our patches do not disable. This feature is implemented by replacing many Google web domains in the source code with non-existent alternatives ending in `qjz9zk` (known as domain substitution; [see docs/design.md](docs/design.md#source-file-processors) for details), then [modifying Chromium to block its own requests with such domains](patches/core/ungoogled-chromium/block-trk-and-subdomains.patch). In other words, no connections are attempted to the `qjz9zk` domain.
* Strip binaries from the source code (known as binary pruning; [see docs/design.md](docs/design.md#source-file-processors) for details)

### Enhancing Features

*These are the non-essential features introduced by ungoogled-chromium.*

* Add many new command-line switches and `chrome://flags` entries to configure new features (which are disabled by default). See [docs/flags.md](docs/flags.md) for the exhaustive list.
* Add *Suggestions URL* text field in the search engine editor (`chrome://settings/searchEngines`) for customizing search engine suggestions.
* Add more URL schemes allowed to save page schemes.
* Add Omnibox search provider "No Search" to allow disabling of searching
* Add a custom cross-platform build configuration and packaging wrapper for Chromium. It currently supports many Linux distributions, macOS, and Windows. (See [docs/design.md](docs/design.md) for details on the system.)
* Force all pop-ups into tabs
* Disable automatic formatting of URLs in Omnibox (e.g. stripping `http://`, hiding certain parameters)
* Disable intranet redirect detector (extraneous DNS requests)
    * This breaks captive portal detection, but captive portals still work.
* (Iridium Browser feature change) Prevent URLs with the `trk:` scheme from connecting to the Internet
    * Also prevents any URLs with the top-level domain `qjz9zk` (as used in domain substitution) from attempting a connection.
* (Windows-specific) Do not set the Zone Identifier on downloaded files

### Borrowed Features

In addition to the features introduced by ungoogled-chromium, ungoogled-chromium selectively borrows many features from the following projects (in approximate order of significance):

* [Inox patchset](https://github.com/gcarq/inox-patchset)
* [Bromite](https://github.com/bromite/bromite)
* [Debian](https://tracker.debian.org/pkg/chromium)
* [Iridium Browser](https://iridiumbrowser.de/)

### Supported Platforms and Distributions

[See docs/platforms.md for a list of supported platforms](docs/platforms.md).

Other platforms are discussed and tracked in this repository's Issue Tracker. Learn more about using the Issue Tracker under the section [Contributing, Reporting, Contacting](#contributing-reporting-contacting).

## Downloads

### Automated or maintained builds

ungoogled-chromium is available in the following **software repositories**:

* Arch: Available in AUR & OBS, [see instructions in ungoogled-chromium-archlinux](https://github.com/ungoogled-software/ungoogled-chromium-archlinux)
* Debian & Ubuntu: Available in OBS, find your [distribution specific instructions](https://github.com/ungoogled-software/ungoogled-chromium-debian) in the Installing section
* Fedora: Available in [COPR](https://copr.fedorainfracloud.org/coprs/) as [`wojnilowicz/ungoogled-chromium`](https://copr.fedorainfracloud.org/coprs/wojnilowicz/ungoogled-chromium/). Also available in [RPM Fusion](https://rpmfusion.org/Configuration) as `chromium-browser-privacy` (outdated).
* Gentoo: Available in [`::pf4public`](https://github.com/PF4Public/gentoo-overlay) overlay as [`ungoogled-chromium`](https://github.com/PF4Public/gentoo-overlay/tree/master/www-client/ungoogled-chromium) and [`ungoogled-chromium-bin`](https://github.com/PF4Public/gentoo-overlay/tree/master/www-client/ungoogled-chromium-bin) ebuilds
* [OpenMandriva](https://openmandriva.org/) includes ungoogled-chromium as its main browser. The `chromium` package includes all ungoogling patches.
* macOS: Available in [Homebrew](https://brew.sh/) as [`eloston-chromium`](https://formulae.brew.sh/cask/eloston-chromium). Just run `brew install --cask eloston-chromium`. Chromium will appear in your `/Applications` directory.

If your GNU/Linux distribution is not listed, there are distro-independent builds available via the following **package managers**:

* Flatpak: Available [in the Flathub repo](https://flathub.org/apps/details/io.github.ungoogled_software.ungoogled_chromium) as `io.github.ungoogled_software.ungoogled_chromium`
* GNU Guix: Available as `ungoogled-chromium`
* NixOS/nixpkgs: Available as `ungoogled-chromium`

### Third-party binaries

If your operating system is not listed above, you can also try to [**Download binaries from here**](https://ungoogled-software.github.io/ungoogled-chromium-binaries/)

*NOTE: These binaries are provided by anyone who are willing to build and submit them. Because these binaries are not necessarily [reproducible](https://reproducible-builds.org/), authenticity cannot be guaranteed; In other words, there is always a non-zero probability that these binaries may have been tampered with. In the unlikely event that this has happened to you, please [report it in a new issue](#contributing-reporting-contacting).*

These binaries are known as **contributor binaries**.

## Source Code

This repository only contains the common code for all platforms; it does not contain all the configuration and scripts necessary to build ungoogled-chromium. Most users will want to use platform-specific repos, where all the remaining configuration and scripts are provided for specific platforms:

[**Find the repo for a specific platform here**](docs/platforms.md).

If you wish to include ungoogled-chromium code in your own build process, consider using [the tags in this repo](https://github.com/ungoogled-software/ungoogled-chromium/tags). These tags follow the format `{chromium_version}-{revision}` where

* `chromium_version` is the version of Chromium used in `x.x.x.x` format, and
* `revision` is a number indicating the version of ungoogled-chromium for the corresponding Chromium version.

Additionally, most platform-specific repos extend their tag scheme upon this one.

**Building the source code**: [See docs/building.md](docs/building.md)

### Mirrors

List of mirrors:

* [Codeberg](https://codeberg.org): [main repo](https://codeberg.org/ungoogled-software/ungoogled-chromium) and [ungoogled-software](https://codeberg.org/ungoogled-software)

## FAQ

[See the frequently-asked questions (FAQ) on the Wiki](https://ungoogled-software.github.io/ungoogled-chromium-wiki/faq)

## Building Instructions

[See docs/building.md](docs/building.md)

## Design Documentation

[See docs/design.md](docs/design.md)

## Contributing, Reporting, Contacting

* For reporting and contacting, see [SUPPORT.md](SUPPORT.md)
* If you're willing to help, check out the [Issue Tracker](https://github.com/ungoogled-software/ungoogled-chromium/issues) and especially issues, which [need help](https://github.com/ungoogled-software/ungoogled-chromium/issues?q=is%3Aopen+is%3Aissue+label%3A%22help+wanted%22)
* For contributing (e.g. how to help, submitting changes, criteria for new features), see [docs/contributing.md](docs/contributing.md)
* If you have some small contributions that don't fit our criteria, consider adding them to [ungoogled-software/contrib](https://github.com/ungoogled-software/contrib) or [our Wiki](https://github.com/ungoogled-software/ungoogled-chromium-wiki) instead.

## Credits

* [The Chromium Project](https://www.chromium.org/)
* [Inox patchset](https://github.com/gcarq/inox-patchset)
* [Debian](https://tracker.debian.org/pkg/chromium-browser)
* [Bromite](https://github.com/bromite/bromite)
* [Iridium Browser](https://iridiumbrowser.de/)
* The users for testing and debugging, [contributing code](https://github.com/ungoogled-software/ungoogled-chromium/graphs/contributors), providing feedback, or simply using ungoogled-chromium in some capacity.

## Related Projects

List of known projects that fork or use changes from ungoogled-chromium:

* [Bromite](https://github.com/bromite/bromite) (Borrows some patches. Features builds for Android)
* [ppc64le fork](https://github.com/leo-lb/ungoogled-chromium) (Fork with changes to build for ppc64le CPUs)

## License

BSD-3-clause. See [LICENSE](LICENSE)
