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