# This Source Code Form is subject to the terms of the Mozilla Public # License, v. 2.0. If a copy of the MPL was not distributed with this # file, You can obtain one at http://mozilla.org/MPL/2.0/. import hashlib import json import os import tarfile import tempfile from pathlib import Path import requests import yaml from mozlint.pathutils import expand_exclusions BROWSERTIME_FETCHES_PATH = Path("taskcluster/kinds/fetch/browsertime.yml") CUSTOMIZATIONS_PATH = Path("testing/condprofile/condprof/customization/") DOWNLOAD_TIMEOUT = 30 ERR_FETCH_TASK_MISSING = "firefox-addons taskcluster fetch config section not found" ERR_FETCH_TASK_ADDPREFIX = "firefox-addons taskcluster config 'add-prefix' attribute should be set to 'firefox-addons/'" ERR_FETCH_TASK_ARCHIVE = ( "Error downloading or opening archive from firefox-addons taskcluster fetch url" ) LINTER_NAME = "condprof-addons" MOZ_FETCHES_DIR = os.environ.get("MOZ_FETCHES_DIR") RULE_DESC = "condprof addons all listed in firefox-addons.tar fetched archive" MOZ_AUTOMATION = "MOZ_AUTOMATION" in os.environ tempdir = tempfile.gettempdir() def lint(paths, config, logger, fix=None, **lintargs): filepaths = [Path(p) for p in expand_exclusions(paths, config, lintargs["root"])] if len(filepaths) == 0: return linter = CondprofAddonsLinter(topsrcdir=lintargs["root"], logger=logger) for filepath in filepaths: linter.lint(filepath) class CondprofAddonsLinter: def __init__(self, topsrcdir, logger): self.topsrcdir = topsrcdir self.logger = logger self.BROWSERTIME_FETCHES_FULLPATH = Path( self.topsrcdir, BROWSERTIME_FETCHES_PATH ) self.CUSTOMIZATIONS_FULLPATH = Path(self.topsrcdir, CUSTOMIZATIONS_PATH) self.tar_xpi_filenames = self.get_firefox_addons_tar_names() def lint(self, filepath): data = self.read_json(filepath) if "addons" not in data: return for addon_key in data["addons"]: xpi_url = data["addons"][addon_key] xpi_filename = xpi_url.split("/")[-1] self.logger.info(f"Found addon {xpi_filename}") if xpi_filename not in self.tar_xpi_filenames: self.logger.lint_error( self.get_missing_xpi_msg(xpi_filename), lineno=0, column=None, path=str(filepath), linter=LINTER_NAME, rule=RULE_DESC, ) def get_missing_xpi_msg(self, xpi_filename): return f"{xpi_filename} is missing from the firefox-addons.tar archive" def read_json(self, filepath): with filepath.open("r") as f: return json.load(f) def read_yaml(self, filepath): with filepath.open("r") as f: return yaml.safe_load(f) def download_firefox_addons_tar(self, firefox_addons_tar_url, tar_tmp_path): self.logger.info(f"Downloading {firefox_addons_tar_url} to {tar_tmp_path}") res = requests.get( firefox_addons_tar_url, stream=True, timeout=DOWNLOAD_TIMEOUT ) res.raise_for_status() with tar_tmp_path.open("wb") as f: for chunk in res.iter_content(chunk_size=1024): if chunk is not None: f.write(chunk) f.flush() def get_firefox_addons_tar_names(self): # Get firefox-addons fetch task config. browsertime_fetches = self.read_yaml(self.BROWSERTIME_FETCHES_FULLPATH) if not ( "firefox-addons" in browsertime_fetches and "fetch" in browsertime_fetches["firefox-addons"] ): self.logger.lint_error( ERR_FETCH_TASK_MISSING, lineno=0, column=None, path=BROWSERTIME_FETCHES_PATH, linter=LINTER_NAME, rule=RULE_DESC, ) return [] fetch_config = browsertime_fetches["firefox-addons"]["fetch"] if not ( "add-prefix" in fetch_config and fetch_config["add-prefix"] == "firefox-addons/" ): self.logger.lint_error( ERR_FETCH_TASK_ADDPREFIX, lineno=0, column=None, path=BROWSERTIME_FETCHES_PATH, linter=LINTER_NAME, rule=RULE_DESC, ) return [] firefox_addons_tar_url = fetch_config["url"] firefox_addons_tar_sha256 = fetch_config["sha256"] tar_xpi_files = list() # When running on the CI, try to retrieve the list of xpi files from the target MOZ_FETCHES_DIR # subdirectory instead of downloading the archive from the fetch url. if MOZ_AUTOMATION: fetches_path = ( Path(MOZ_FETCHES_DIR) if MOZ_FETCHES_DIR is not None else None ) if fetches_path is not None and fetches_path.exists(): self.logger.info( "Detected MOZ_FETCHES_DIR, look for pre-downloaded firefox-addons fetch results" ) # add-prefix presence and value has been enforced at the start of this method. fetches_addons = Path(fetches_path, "firefox-addons/") if fetches_addons.exists(): self.logger.info( f"Retrieve list of xpi files from firefox-addons fetch result at {str(fetches_addons)}" ) for xpi_path in fetches_addons.iterdir(): if xpi_path.suffix == ".xpi": tar_xpi_files.append(xpi_path.name) return tar_xpi_files else: self.logger.warning( "No 'firefox-addons/' subdir found in MOZ_FETCHES_DIR" ) # Fallback to download the tar archive and retrieve the list of xpi file from it # (e.g. when linting the local changes on the developers environment). tar_tmp_path = Path(tempdir, "firefox-addons.tar") tar_tmp_ready = False # If the firefox-addons.tar file is found in the tempdir, check if the # file hash matches, if it does then don't download it again. if tar_tmp_path.exists(): tar_tmp_hash = hashlib.sha256() with tar_tmp_path.open("rb") as f: while chunk := f.read(1024): tar_tmp_hash.update(chunk) if tar_tmp_hash.hexdigest() == firefox_addons_tar_sha256: self.logger.info( f"Pre-downloaded file for {tar_tmp_path} found and sha256 matching" ) tar_tmp_ready = True else: self.logger.info( f"{tar_tmp_path} sha256 does not match the fetch config" ) # If the file is not found or the hash doesn't match, download it from the fetch task url. if not tar_tmp_ready: try: self.download_firefox_addons_tar(firefox_addons_tar_url, tar_tmp_path) except requests.exceptions.HTTPError as http_err: self.logger.lint_error( f"{ERR_FETCH_TASK_ARCHIVE}, {str(http_err)}", lineno=0, column=None, path=BROWSERTIME_FETCHES_PATH, linter=LINTER_NAME, rule=RULE_DESC, ) return [] # Retrieve and return the list of xpi file names. try: with tarfile.open(tar_tmp_path, "r") as tf: names = tf.getnames() for name in names: file_path = Path(name) if file_path.suffix == ".xpi": tar_xpi_files.append(file_path.name) except tarfile.ReadError as read_err: self.logger.lint_error( f"{ERR_FETCH_TASK_ARCHIVE}, {str(read_err)}", lineno=0, column=None, path=BROWSERTIME_FETCHES_PATH, linter=LINTER_NAME, rule=RULE_DESC, ) return [] return tar_xpi_files