#!/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()