release_checker.py

#!/usr/bin/env python3
# Copyright 2019, The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import argparse
import os
import shlex
import subprocess
import sys


PREBUILTS_DIR = os.path.dirname(__file__)
PROJECTS = {'acloud', 'aidegen', 'atest'}
ARCHS = {'linux-x86', 'darwin-x86'}
SMOKE_TEST = 'smoke_tests'
EXIT_TEST_PASS = 0
EXIT_TEST_FAIL = 1
EXIT_INVALID_BINS = 2


def _get_prebuilt_bins():
    """Get asuite prebuilt binaries.

    Returns:
        A set of prebuilt binaries.
    """
    bins = {os.path.join(prj, arch, prj) for prj in PROJECTS for arch in ARCHS}
    # atest becomes an entrypoint which invokes atest-py2 and atest-py3 accordingly.
    bins.add('atest/linux-x86/atest-py2')
    bins.add('atest/linux-x86/atest-py3')
    return bins


def _get_prebuilt_dirs():
    """Get asuite prebuilt directories.

    Returns:
        A set of prebuilt paths of binaries.
    """
    return {os.path.dirname(bin) for bin in _get_prebuilt_bins()}


def _get_smoke_tests_bins():
    """Get asuite smoke test scripts.

    Returns:
        A dict of project and smoke test script paths.
    """
    return {prj: os.path.join(prj, SMOKE_TEST) for prj in PROJECTS}


def _is_executable(bin_path):
    """Check if the given file is executable.

    Args:
        bin_path: a string of a file path.

    Returns:
        True if it is executable, false otherwise.
    """
    return os.access(bin_path, os.X_OK)


def check_uploaded_bins(preupload_files):
    """This method validates the uploaded files.

    If the uploaded file is in prebuilt_bins, ensure:
        - it is executable.
        - only one at a time.
    If the uploaded file is a smoke_test script, ensure:
        - it is executable.
    If the uploaded file is placed in prebuilt_dirs, ensure:
        - it is not executable.
        (It is to ensure PATH is not contaminated.
         e.g. atest/linux-x86/atest-dev will override $OUT/bin/atest-dev, or
         atest/linux-x86/rm does fraud/harmful things.)

    Args:
        preupload_files: A list of preuploaded files.

    Returns:
        True is the above criteria are all fulfilled, otherwise None.
    """
    prebuilt_bins = _get_prebuilt_bins()
    prebuilt_dirs = _get_prebuilt_dirs()
    smoke_tests_bins = _get_smoke_tests_bins().values()
    # Store valid executables.
    target_bins = set()
    # Unexpected executable files which may cause issues(they are in $PATH).
    illegal_bins = set()
    # Store prebuilts or smoke test script that are inexecutable.
    insufficient_perm_bins = set()
    for f in preupload_files:
        # Ensure target_bins are executable.
        if f in prebuilt_bins:
            if _is_executable(f):
                target_bins.add(f)
            else:
                insufficient_perm_bins.add(f)
        # Ensure smoke_tests scripts are executable.
        elif f in smoke_tests_bins and not _is_executable(f):
            insufficient_perm_bins.add(f)
        # Avoid fraud commands in $PATH. e.g. atest/linux-x86/rm.
        # must not be executable.
        elif os.path.dirname(f) in prebuilt_dirs and _is_executable(f):
            illegal_bins.add(f)
    if len(target_bins) > 1:
        print('\nYou\'re uploading multiple binaries: %s'
              % ' '.join(target_bins))
        print('\nPlease upload one prebuilt at a time.')
        return False
    if insufficient_perm_bins:
        print('\nInsufficient permission found: %s'
              % ' '.join(insufficient_perm_bins))
        print('\nPlease run:\n\tchmod 0755 %s\nand try again.'
              % ' '.join(insufficient_perm_bins))
        return False
    if illegal_bins:
        illegal_dirs = {os.path.dirname(bin) for bin in illegal_bins}
        print('\nIt is forbidden to upload executable file: %s'
              % '\n - %s\n' % '\n - '.join(illegal_bins))
        print('Because they are in the project paths: %s'
              % '\n - %s\n' % '\n - '.join(illegal_dirs))
        print('Please remove the binaries or make the files non-executable.')
        return False
    return True


def run_smoke_tests_pass(files_to_check):
    """Run smoke tests.

    Args:
        files_to_check: A list of preuploaded files to check.

    Returns:
        True when test passed or no need to test.
        False when test failed.
    """
    for target in files_to_check:
        if target in _get_prebuilt_bins():
            project = target.split(os.path.sep)[0]
            test_file = _get_smoke_tests_bins().get(project)
            if os.path.exists(test_file):
                try:
                    subprocess.check_output(test_file, encoding='utf-8',
                                            stderr=subprocess.STDOUT)
                except subprocess.CalledProcessError as error:
                    print('Smoke tests failed at:\n\n%s' % error.output)
                    return False
                except OSError as oserror:
                    print('%s: Missing the header of the script.' % oserror)
                    print('Please define shebang like:\n')
                    print('#!/usr/bin/env bash\nor')
                    print('#!/usr/bin/env python3\n')
                    return False
    return True


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--skip-smoke-test', '-s', action="store_true",
                        help='Disable smoke testing.')
    parser.add_argument('preupload_files', nargs='*', help='Files to be uploaded.')
    args = parser.parse_args()
    files_to_check = args.preupload_files
    # Pre-process files_to_check(run directly by users.)
    if not files_to_check:
        # Only consider added(A), renamed(R) and modified(M) files.
        cmd = "git status --short | egrep ^[ARM] | awk '{print $NF}'"
        preupload_files = subprocess.check_output(cmd, shell=True,
                                                 encoding='utf-8').splitlines()
        if preupload_files:
            print('validating: %s' % preupload_files)
            files_to_check = preupload_files
    # Validating uploaded files and run smoke test script(run by repohook).
    if not check_uploaded_bins(files_to_check):
        sys.exit(EXIT_INVALID_BINS)
    if not args.skip_smoke_test and not run_smoke_tests_pass(files_to_check):
        sys.exit(EXIT_TEST_FAIL)
    sys.exit(EXIT_TEST_PASS)