#!/usr/bin/env python3
import argparse
import os
import re
import sys
from collections import defaultdict
from elftools.elf.dynamic import DynamicSegment
from elftools.elf.elffile import ELFFile
def log(message):
sys.stderr.write(message + "\n")
def ignore_toolchains(component):
return not component.startswith("toolchain")
def read_file(path):
with open(path, "r") as input_file:
return [line.strip() for line in input_file]
def is_executable(path):
return os.access(path, os.X_OK)
def is_elf(path):
with open(path, "rb") as input_file:
return input_file.read(4) == b"\x7FELF"
def unique_or_none(list):
if len(list) == 1:
return list[0]
else:
return None
def get_dynamic(elf):
return unique_or_none([segment
for segment
in elf.iter_segments()
if type(segment) is DynamicSegment])
class Root:
def __init__(self, root_path):
self.file_map = dict()
self.reverse_file_map = defaultdict(list)
self.root_path = root_path
self.package_files_path = os.path.join(self.root_path, "share", "orchestra")
self.all_files = set(["lib"])
def load_file(self, path):
files = read_file(path)
path = os.path.relpath(path, self.package_files_path)
for file in files:
if file.startswith("./"):
file = file[2:]
self.reverse_file_map[file].append(path)
self.all_files.add(file)
self.file_map[path] = files
def load_package_files(self):
# Walk recursively all the file the text files
for directory, subdirectories, files in os.walk(self.package_files_path):
for file in files:
# Skip metadata files
if file.endswith(".json"):
continue
self.load_file(os.path.join(directory, file))
def report_duplicates(self):
header = False
for file, packages in self.reverse_file_map.items():
if len(packages) > 1:
if not header:
header = True
log("Files in multiple packages:")
log(" {}:".format(file))
for package in packages:
log(" {}".format(package))
return header
def collect_installed_files(self):
self.installed_files = set()
for directory, subdirectories, files in os.walk(self.root_path):
for subdirectory in subdirectories:
path = os.path.join(directory, subdirectory)
if os.path.islink(path):
self.installed_files.add(os.path.relpath(path, self.root_path))
for file in files:
path = os.path.join(directory, file)
self.installed_files.add(os.path.relpath(path, self.root_path))
def check_installed_files(self):
missing_files = self.all_files - self.installed_files
if missing_files:
log("The following files are listed as installed but are not"
+ " present in root:")
for missing_file in sorted(missing_files):
log(" {}".format(missing_file))
extra_files = self.installed_files - self.all_files
if extra_files:
log("The following files are present in root but do not belong to"
+ " any component:")
for extra_file in sorted(extra_files):
log(" {}".format(extra_file))
return len(missing_files) > 0 or len(extra_files) > 0
def is_for_host(self, path, elf):
if elf.header.e_machine != "EM_X86_64":
return False
return True
def prepare_file_list(self, files, prefix="", component_filter=lambda component: True):
result = ""
by_component = defaultdict(list)
for file in files:
components = ""
if file in self.reverse_file_map:
for component in self.reverse_file_map[file]:
by_component[component].append(file)
else:
by_component["(orphan)"].append(file)
for component, files in sorted(by_component.items()):
if not component_filter(component):
continue
result += "{}{}:\n".format(prefix, component)
for file in files:
result += "{} {}\n".format(prefix, file)
return result
def print_file_list(self, files, prefix=""):
log(self.prepare_file_list(files, prefix))
def verify_elfs(self):
missing_libraries = defaultdict(list)
libraries_in_root = defaultdict(list)
allowed_glibc_versions = set()
used_glibc_versions = dict()
invalid_runpaths = defaultdict(list)
for installed_file in sorted(self.installed_files):
path = os.path.join(self.root_path, installed_file)
if os.path.isfile(path) and is_executable(path) and is_elf(path):
with open(path, "rb") as elf_file:
elf = ELFFile(elf_file)
dynamic_segment = get_dynamic(elf)
if (self.is_for_host(installed_file, elf)
and dynamic_segment):
if "link-only" not in installed_file:
libraries_in_root[os.path.basename(installed_file)].append(installed_file)
# Get the string table
tag = unique_or_none([tag
for tag
in dynamic_segment.iter_tags()
if tag.entry.d_tag == "DT_STRTAB"])
string_table_address = tag.entry.d_val
string_table_offset = unique_or_none(list(elf.address_offsets(string_table_address)))
tag = unique_or_none([tag
for tag
in dynamic_segment.iter_tags()
if tag.entry.d_tag == "DT_STRSZ"])
string_table_size = tag.entry.d_val
elf_file.seek(string_table_offset)
string_table = elf_file.read(string_table_size)
glibc_versions = set([version.strip(b"\x00").decode("ascii")
for version in
re.findall(b"GLIBC_[0-9.]*\x00", string_table)])
if "link-only" in installed_file:
allowed_glibc_versions = allowed_glibc_versions.union(glibc_versions)
else:
used_glibc_versions[installed_file] = glibc_versions
runpaths = []
runpath_tag = unique_or_none([tag
for tag
in dynamic_segment.iter_tags()
if tag.entry.d_tag == "DT_RUNPATH"])
if runpath_tag:
runpath = string_table[runpath_tag.entry.d_val:].split(b"\x00")[0].decode("ascii")
runpath = runpath.replace("$ORIGIN", os.path.dirname(os.path.realpath(path)))
runpaths = runpath.split(":")
runpaths = map(os.path.realpath, runpaths)
runpaths = [os.path.relpath(runpath, self.root_path)
for runpath
in runpaths]
runpaths = list(set(runpaths))
for runpath in runpaths:
path = os.path.join(self.root_path, runpath)
if not (os.path.isdir(path) or os.path.islink(path)):
invalid_runpaths[runpath].append(installed_file)
# Collect DT_NEEDED
needed_string_offsets = [tag.entry.d_val
for tag
in dynamic_segment.iter_tags()
if tag.entry.d_tag == "DT_NEEDED"]
for needed_string_offset in needed_string_offsets:
lib_name = string_table[needed_string_offset:].split(b"\x00")[0].decode("ascii")
found = False
for runpath in runpaths:
candidate = os.path.relpath(os.path.join(self.root_path,
runpath,
lib_name),
self.root_path)
if candidate in self.all_files:
found = True
break
if not found:
missing_libraries[lib_name].append(installed_file)
if invalid_runpaths:
log("The following runpaths are invalid:")
for runpath, users in invalid_runpaths.items():
log(" {}".format(runpath))
self.print_file_list(users, " ")
system_libraries = []
for missing_library, users in missing_libraries.items():
if missing_library in libraries_in_root:
file_list = self.prepare_file_list(users, " ", ignore_toolchains)
if file_list:
log("{} is available in root".format(missing_library))
log(" These are the instances:")
self.print_file_list(libraries_in_root[missing_library], " ")
log(" These are the users:")
log(file_list)
else:
system_libraries.append((missing_library, users))
if system_libraries and False:
log("The following libraries are not provided in root:")
for system_library, users in system_libraries:
log(" {}:".format(system_library))
self.print_file_list(users, " ")
by_version = defaultdict(list)
for installed_file, versions in used_glibc_versions.items():
unallowed_versions = versions - allowed_glibc_versions
for unallowed_version in unallowed_versions:
by_version[unallowed_version].append(installed_file)
to_print = list()
for version, users in sorted(by_version.items()):
file_list = self.prepare_file_list(users, " ", ignore_toolchains)
if file_list:
to_print.append((version, users, file_list))
if to_print:
log("The following unallowed glibc versions are being used:")
for version, users, file_list in to_print:
log(" {}".format(version))
log(file_list)
return any(len(x) > 0 for x in [invalid_runpaths, to_print])
def main():
parser = argparse.ArgumentParser(description="Verify integrity of an orchestra root.")
parser.add_argument("root_path", metavar="ROOT_PATH", default=".", help="Path to Orchestra root")
args = parser.parse_args()
root_path = args.root_path
root = Root(root_path)
root.load_package_files()
duplicates_found = root.report_duplicates()
root.collect_installed_files()
orphans_found = root.check_installed_files()
errors_in_elfs = root.verify_elfs()
if duplicates_found or orphans_found or errors_in_elfs:
log("[!] Inconsistencies found in the root directory!")
return 1
else:
log("Root directory consistency checks passed!")
return 0
if __name__ == "__main__":
sys.exit(main())