scan_deps Namespace Reference

Classes
class	VarImpl

Functions
def	extract_deps (deps_file)
def	parse_readme ()
def	get_common_ancestor (dep, deps_list)
def	parse_args (args)
def	write_manifest (deps, manifest_file)
def	main (argv)

Variables
	SCRIPT_DIR = os.path.dirname(sys.argv[0])
	CHECKOUT_ROOT = os.path.realpath(os.path.join(SCRIPT_DIR, '..'))
string	CHROMIUM_README_FILE = 'third_party/accessibility/README.md'
int	CHROMIUM_README_COMMIT_LINE = 4
string	CHROMIUM = 'https://chromium.googlesource.com/chromium/src'
string	DEP_CLONE_DIR = CHECKOUT_ROOT + '/clone-test'
	DEPS = os.path.join(CHECKOUT_ROOT, 'DEPS')
string	UPSTREAM_PREFIX = 'upstream_'

Function Documentation

extract_deps()

def scan_deps.extract_deps

(

deps_file

)

Definition at line 54 of file scan_deps.py.

def extract_deps(deps_file):
  local_scope = {}
  var = VarImpl(local_scope)
  global_scope = {
      'Var': var.lookup,
      'deps_os': {},
  }
  # Read the content.
  with open(deps_file, 'r') as file:
    deps_content = file.read()
 
  # Eval the content.
  exec(deps_content, global_scope, local_scope)
 
  if not os.path.exists(DEP_CLONE_DIR):
    os.mkdir(DEP_CLONE_DIR)  # Clone deps with upstream into temporary dir.
 
  # Extract the deps and filter.
  deps = local_scope.get('deps', {})
  deps_list = local_scope.get('vars')
  filtered_osv_deps = []
  for _, dep in deps.items():
    # We currently do not support packages or cipd which are represented
    # as dictionaries.
    if not isinstance(dep, str):
      continue
 
    dep_split = dep.rsplit('@', 1)
    ancestor_result = get_common_ancestor([dep_split[0], dep_split[1]], deps_list)
    if ancestor_result:
      filtered_osv_deps.append({
          'package': {'name': ancestor_result[1], 'commit': ancestor_result[0]}
      })
 
  try:
    # Clean up cloned upstream dependency directory.
    shutil.rmtree(DEP_CLONE_DIR)  # Use shutil.rmtree since dir could be non-empty.
  except OSError as clone_dir_error:
    print('Error cleaning up clone directory: %s : %s' % (DEP_CLONE_DIR, clone_dir_error.strerror))
 
  osv_result = {
      'packageSource': {'path': deps_file, 'type': 'lockfile'}, 'packages': filtered_osv_deps
  }
  return osv_result
 
 

get_common_ancestor()

def scan_deps.get_common_ancestor	(		dep,
			deps_list
	)

Given an input of a mirrored dep,
compare to the mapping of deps to their upstream
in DEPS and find a common ancestor
commit SHA value.

This is done by first cloning the mirrored dep,
then a branch which tracks the upstream.
From there,  git merge-base operates using the HEAD
commit SHA of the upstream branch and the pinned
SHA value of the mirrored branch

Definition at line 121 of file scan_deps.py.

def get_common_ancestor(dep, deps_list):
  """
  Given an input of a mirrored dep,
  compare to the mapping of deps to their upstream
  in DEPS and find a common ancestor
  commit SHA value.
 
  This is done by first cloning the mirrored dep,
  then a branch which tracks the upstream.
  From there,  git merge-base operates using the HEAD
  commit SHA of the upstream branch and the pinned
  SHA value of the mirrored branch
  """
  # dep[0] contains the mirror repo.
  # dep[1] contains the mirror's pinned SHA.
  # upstream is the origin repo.
  dep_name = dep[0].split('/')[-1].split('.')[0]
  if UPSTREAM_PREFIX + dep_name not in deps_list:
    print('did not find dep: ' + dep_name)
    return None
  try:
    # Get the upstream URL from the mapping in DEPS file.
    upstream = deps_list.get(UPSTREAM_PREFIX + dep_name)
    temp_dep_dir = DEP_CLONE_DIR + '/' + dep_name
    # Clone dependency from mirror.
    subprocess.check_output(['git', 'clone', '--quiet', '--', dep[0], dep_name], cwd=DEP_CLONE_DIR)
 
    # Create branch that will track the upstream dep.
    print('attempting to add upstream remote from: {upstream}'.format(upstream=upstream))
    subprocess.check_output(['git', 'remote', 'add', 'upstream', upstream], cwd=temp_dep_dir)
    subprocess.check_output(['git', 'fetch', '--quiet', 'upstream'], cwd=temp_dep_dir)
    # Get name of the default branch for upstream (e.g. main/master/etc.).
    default_branch = subprocess.check_output(
        'git remote show upstream ' + "| sed -n \'/HEAD branch/s/.*: //p\'",
        cwd=temp_dep_dir,
        shell=True
    )
    default_branch = byte_str_decode(default_branch)
    default_branch = default_branch.strip()
 
    # Make upstream branch track the upstream dep.
    subprocess.check_output([
        'git', 'checkout', '--force', '-b', 'upstream', '--track', 'upstream/' + default_branch
    ],
                            cwd=temp_dep_dir)
    # Get the most recent commit from default branch of upstream.
    commit = subprocess.check_output(
        'git for-each-ref ' + "--format=\'%(objectname:short)\' refs/heads/upstream",
        cwd=temp_dep_dir,
        shell=True
    )
    commit = byte_str_decode(commit)
    commit = commit.strip()
 
    # Perform merge-base on most recent default branch commit and pinned mirror commit.
    ancestor_commit = subprocess.check_output(
        'git merge-base {commit} {depUrl}'.format(commit=commit, depUrl=dep[1]),
        cwd=temp_dep_dir,
        shell=True
    )
    ancestor_commit = byte_str_decode(ancestor_commit)
    ancestor_commit = ancestor_commit.strip()
    print('Ancestor commit: ' + ancestor_commit)
    return ancestor_commit, upstream
  except subprocess.CalledProcessError as error:
    print(
        "Subprocess command '{0}' failed with exit code: {1}.".format(
            error.cmd, str(error.returncode)
        )
    )
    if error.output:
      print("Subprocess error output: '{0}'".format(error.output))
  return None
 
 

main()

def scan_deps.main

(

argv

)

Definition at line 225 of file scan_deps.py.

def main(argv):
  args = parse_args(argv)
  deps = extract_deps(args.deps)
  readme_deps = parse_readme()
  write_manifest([deps, readme_deps], args.output)
  return 0
 
 

parse_args()

def scan_deps.parse_args

(

args

)

Definition at line 196 of file scan_deps.py.

def parse_args(args):
  args = args[1:]
  parser = argparse.ArgumentParser(description='A script to find common ancestor commit SHAs')
 
  parser.add_argument(
      '--deps',
      '-d',
      type=str,
      help='Input DEPS file to extract.',
      default=os.path.join(CHECKOUT_ROOT, 'DEPS')
  )
  parser.add_argument(
      '--output',
      '-o',
      type=str,
      help='Output osv-scanner compatible deps file.',
      default=os.path.join(CHECKOUT_ROOT, 'osv-lockfile.json')
  )
 
  return parser.parse_args(args)
 
 

parse_readme()

def scan_deps.parse_readme

(

)

Opens the Flutter Accessibility Library README and uses the commit hash
found in the README to check for viulnerabilities.
The commit hash in this README will always be in the same format

Definition at line 100 of file scan_deps.py.

def parse_readme():
  """
  Opens the Flutter Accessibility Library README and uses the commit hash
  found in the README to check for viulnerabilities.
  The commit hash in this README will always be in the same format
  """
  file_path = os.path.join(CHECKOUT_ROOT, CHROMIUM_README_FILE)
  with open(file_path) as file:
    # Read the content of the file opened.
    content = file.readlines()
    commit_line = content[CHROMIUM_README_COMMIT_LINE]
    commit = re.search(r'(?<=\[).*(?=\])', commit_line)
 
    osv_result = {
        'packageSource': {'path': file_path, 'type': 'lockfile'},
        'packages': [{'package': {'name': CHROMIUM, 'commit': commit.group()}}]
    }
 
    return osv_result
 
 

write_manifest()

def scan_deps.write_manifest	(		deps,
			manifest_file
	)

Definition at line 218 of file scan_deps.py.

def write_manifest(deps, manifest_file):
  output = {'results': deps}
  print(json.dumps(output, indent=2))
  with open(manifest_file, 'w') as manifest:
    json.dump(output, manifest, indent=2)
 
 

Variable Documentation

CHECKOUT_ROOT

scan_deps.CHECKOUT_ROOT = os.path.realpath(os.path.join(SCRIPT_DIR, '..'))

Definition at line 25 of file scan_deps.py.

CHROMIUM

string scan_deps.CHROMIUM = 'https://chromium.googlesource.com/chromium/src'

Definition at line 28 of file scan_deps.py.

CHROMIUM_README_COMMIT_LINE

int scan_deps.CHROMIUM_README_COMMIT_LINE = 4

Definition at line 27 of file scan_deps.py.

CHROMIUM_README_FILE

string scan_deps.CHROMIUM_README_FILE = 'third_party/accessibility/README.md'

Definition at line 26 of file scan_deps.py.

DEP_CLONE_DIR

string scan_deps.DEP_CLONE_DIR = CHECKOUT_ROOT + '/clone-test'

Definition at line 29 of file scan_deps.py.

DEPS

scan_deps.DEPS = os.path.join(CHECKOUT_ROOT, 'DEPS')

Definition at line 30 of file scan_deps.py.

SCRIPT_DIR

scan_deps.SCRIPT_DIR = os.path.dirname(sys.argv[0])

Definition at line 24 of file scan_deps.py.

UPSTREAM_PREFIX

string scan_deps.UPSTREAM_PREFIX = 'upstream_'

Definition at line 31 of file scan_deps.py.