|  | #!/usr/bin/env python | 
|  |  | 
|  | import argparse | 
|  | import collections | 
|  | import logging | 
|  | import os | 
|  | import re | 
|  | import subprocess | 
|  | import textwrap | 
|  |  | 
|  | from gensyscalls import SysCallsTxtParser | 
|  |  | 
|  |  | 
|  | BPF_JGE = "BPF_JUMP(BPF_JMP|BPF_JGE|BPF_K, {0}, {1}, {2})" | 
|  | BPF_ALLOW = "BPF_STMT(BPF_RET|BPF_K, SECCOMP_RET_ALLOW)" | 
|  |  | 
|  |  | 
|  | class SyscallRange(object): | 
|  | def __init__(self, name, value): | 
|  | self.names = [name] | 
|  | self.begin = value | 
|  | self.end = self.begin + 1 | 
|  |  | 
|  | def __str__(self): | 
|  | return "(%s, %s, %s)" % (self.begin, self.end, self.names) | 
|  |  | 
|  | def add(self, name, value): | 
|  | if value != self.end: | 
|  | raise ValueError | 
|  | self.end += 1 | 
|  | self.names.append(name) | 
|  |  | 
|  |  | 
|  | def load_syscall_names_from_file(file_path, architecture): | 
|  | parser = SysCallsTxtParser() | 
|  | parser.parse_open_file(open(file_path)) | 
|  | return set([x["name"] for x in parser.syscalls if x.get(architecture)]) | 
|  |  | 
|  |  | 
|  | def merge_names(base_names, whitelist_names, blacklist_names): | 
|  | if bool(blacklist_names - base_names): | 
|  | raise RuntimeError("Blacklist item not in bionic - aborting " + str( | 
|  | blacklist_names - base_names)) | 
|  |  | 
|  | return (base_names - blacklist_names) | whitelist_names | 
|  |  | 
|  |  | 
|  | def parse_syscall_NRs(names_path): | 
|  | # The input is now the preprocessed source file. This will contain a lot | 
|  | # of junk from the preprocessor, but our lines will be in the format: | 
|  | # | 
|  | #    #define __(ARM_)?NR_${NAME} ${VALUE} | 
|  | # | 
|  | # Where ${VALUE} is a preprocessor expression. | 
|  |  | 
|  | constant_re = re.compile( | 
|  | r'^\s*#define\s+([A-Za-z_][A-Za-z0-9_]+)\s+(.+)\s*$') | 
|  | token_re = re.compile(r'\b[A-Za-z_][A-Za-z0-9_]+\b') | 
|  | constants = {} | 
|  | with open(names_path) as f: | 
|  | for line in f: | 
|  | m = constant_re.match(line) | 
|  | if not m: | 
|  | continue | 
|  | try: | 
|  | name = m.group(1) | 
|  | # eval() takes care of any arithmetic that may be done | 
|  | value = eval(token_re.sub(lambda x: str(constants[x.group(0)]), | 
|  | m.group(2))) | 
|  |  | 
|  | constants[name] = value | 
|  | except: | 
|  | logging.debug('Failed to parse %s', line) | 
|  | pass | 
|  |  | 
|  | syscalls = {} | 
|  | for name, value in constants.iteritems(): | 
|  | if not name.startswith("__NR_") and not name.startswith("__ARM_NR"): | 
|  | continue | 
|  | if name.startswith("__NR_"): | 
|  | # Remote the __NR_ prefix | 
|  | name = name[len("__NR_"):] | 
|  | syscalls[name] = value | 
|  |  | 
|  | return syscalls | 
|  |  | 
|  |  | 
|  | def convert_NRs_to_ranges(syscalls): | 
|  | # Sort the values so we convert to ranges and binary chop | 
|  | syscalls = sorted(syscalls, lambda x, y: cmp(x[1], y[1])) | 
|  |  | 
|  | # Turn into a list of ranges. Keep the names for the comments | 
|  | ranges = [] | 
|  | for name, value in syscalls: | 
|  | if not ranges: | 
|  | ranges.append(SyscallRange(name, value)) | 
|  | continue | 
|  |  | 
|  | last_range = ranges[-1] | 
|  | if last_range.end == value: | 
|  | last_range.add(name, value) | 
|  | else: | 
|  | ranges.append(SyscallRange(name, value)) | 
|  | return ranges | 
|  |  | 
|  |  | 
|  | # Converts the sorted ranges of allowed syscalls to a binary tree bpf | 
|  | # For a single range, output a simple jump to {fail} or {allow}. We can't set | 
|  | # the jump ranges yet, since we don't know the size of the filter, so use a | 
|  | # placeholder | 
|  | # For multiple ranges, split into two, convert the two halves and output a jump | 
|  | # to the correct half | 
|  | def convert_to_intermediate_bpf(ranges): | 
|  | if len(ranges) == 1: | 
|  | # We will replace {fail} and {allow} with appropriate range jumps later | 
|  | return [BPF_JGE.format(ranges[0].end, "{fail}", "{allow}") + | 
|  | ", //" + "|".join(ranges[0].names)] | 
|  | else: | 
|  | half = (len(ranges) + 1) / 2 | 
|  | first = convert_to_intermediate_bpf(ranges[:half]) | 
|  | second = convert_to_intermediate_bpf(ranges[half:]) | 
|  | jump = [BPF_JGE.format(ranges[half].begin, len(first), 0) + ","] | 
|  | return jump + first + second | 
|  |  | 
|  |  | 
|  | def convert_ranges_to_bpf(ranges): | 
|  | bpf = convert_to_intermediate_bpf(ranges) | 
|  |  | 
|  | # Now we know the size of the tree, we can substitute the {fail} and {allow} | 
|  | # placeholders | 
|  | for i, statement in enumerate(bpf): | 
|  | # Replace placeholder with | 
|  | # "distance to jump to fail, distance to jump to allow" | 
|  | # We will add a kill statement and an allow statement after the tree | 
|  | # With bpfs jmp 0 means the next statement, so the distance to the end is | 
|  | # len(bpf) - i - 1, which is where we will put the kill statement, and | 
|  | # then the statement after that is the allow statement | 
|  | if "{fail}" in statement and "{allow}" in statement: | 
|  | bpf[i] = statement.format(fail=str(len(bpf) - i), | 
|  | allow=str(len(bpf) - i - 1)) | 
|  |  | 
|  | # Add the allow calls at the end. If the syscall is not matched, we will | 
|  | # continue. This allows the user to choose to match further syscalls, and | 
|  | # also to choose the action when we want to block | 
|  | bpf.append(BPF_ALLOW + ",") | 
|  |  | 
|  | # Add check that we aren't off the bottom of the syscalls | 
|  | bpf.insert(0, BPF_JGE.format(ranges[0].begin, 0, str(len(bpf))) + ',') | 
|  | return bpf | 
|  |  | 
|  |  | 
|  | def convert_bpf_to_output(bpf, architecture, name_modifier): | 
|  | if name_modifier: | 
|  | name_modifier = name_modifier + "_" | 
|  | else: | 
|  | name_modifier = "" | 
|  | header = textwrap.dedent("""\ | 
|  | // File autogenerated by {self_path} - edit at your peril!! | 
|  |  | 
|  | #include <linux/filter.h> | 
|  | #include <errno.h> | 
|  |  | 
|  | #include "seccomp/seccomp_bpfs.h" | 
|  | const sock_filter {architecture}_{suffix}filter[] = {{ | 
|  | """).format(self_path=os.path.basename(__file__), architecture=architecture, | 
|  | suffix=name_modifier) | 
|  |  | 
|  | footer = textwrap.dedent("""\ | 
|  |  | 
|  | }}; | 
|  |  | 
|  | const size_t {architecture}_{suffix}filter_size = sizeof({architecture}_{suffix}filter) / sizeof(struct sock_filter); | 
|  | """).format(architecture=architecture,suffix=name_modifier) | 
|  | return header + "\n".join(bpf) + footer | 
|  |  | 
|  |  | 
|  | def construct_bpf(syscalls, architecture, name_modifier): | 
|  | ranges = convert_NRs_to_ranges(syscalls) | 
|  | bpf = convert_ranges_to_bpf(ranges) | 
|  | return convert_bpf_to_output(bpf, architecture, name_modifier) | 
|  |  | 
|  |  | 
|  | def gen_policy(name_modifier, out_dir, base_syscall_file, syscall_files, syscall_NRs): | 
|  | for arch in ('arm', 'arm64', 'mips', 'mips64', 'x86', 'x86_64'): | 
|  | base_names = load_syscall_names_from_file(base_syscall_file, arch) | 
|  | whitelist_names = set() | 
|  | blacklist_names = set() | 
|  | for f in syscall_files: | 
|  | if "blacklist" in f.lower(): | 
|  | blacklist_names |= load_syscall_names_from_file(f, arch) | 
|  | else: | 
|  | whitelist_names |= load_syscall_names_from_file(f, arch) | 
|  |  | 
|  | allowed_syscalls = [] | 
|  | for name in merge_names(base_names, whitelist_names, blacklist_names): | 
|  | try: | 
|  | allowed_syscalls.append((name, syscall_NRs[arch][name])) | 
|  | except: | 
|  | logging.exception("Failed to find %s in %s", name, arch) | 
|  | raise | 
|  | output = construct_bpf(allowed_syscalls, arch, name_modifier) | 
|  |  | 
|  | # And output policy | 
|  | existing = "" | 
|  | filename_modifier = "_" + name_modifier if name_modifier else "" | 
|  | output_path = os.path.join(out_dir, | 
|  | "{}{}_policy.cpp".format(arch, filename_modifier)) | 
|  | with open(output_path, "w") as output_file: | 
|  | output_file.write(output) | 
|  |  | 
|  |  | 
|  | def main(): | 
|  | parser = argparse.ArgumentParser( | 
|  | description="Generates a seccomp-bpf policy") | 
|  | parser.add_argument("--verbose", "-v", help="Enables verbose logging.") | 
|  | parser.add_argument("--name-modifier", | 
|  | help=("Specifies the name modifier for the policy. " | 
|  | "One of {app,global,system}.")) | 
|  | parser.add_argument("--out-dir", | 
|  | help="The output directory for the policy files") | 
|  | parser.add_argument("base_file", metavar="base-file", type=str, | 
|  | help="The path of the base syscall list (SYSCALLS.TXT).") | 
|  | parser.add_argument("files", metavar="FILE", type=str, nargs="+", | 
|  | help=("The path of the input files. In order to " | 
|  | "simplify the build rules, it can take any of the " | 
|  | "following files: \n" | 
|  | "* /blacklist.*\.txt$/ syscall blacklist.\n" | 
|  | "* /whitelist.*\.txt$/ syscall whitelist.\n" | 
|  | "* otherwise, syscall name-number mapping.\n")) | 
|  | args = parser.parse_args() | 
|  |  | 
|  | if args.verbose: | 
|  | logging.basicConfig(level=logging.DEBUG) | 
|  | else: | 
|  | logging.basicConfig(level=logging.INFO) | 
|  |  | 
|  | syscall_files = [] | 
|  | syscall_NRs = {} | 
|  | for filename in args.files: | 
|  | if filename.lower().endswith('.txt'): | 
|  | syscall_files.append(filename) | 
|  | else: | 
|  | m = re.search(r"libseccomp_gen_syscall_nrs_([^/]+)", filename) | 
|  | syscall_NRs[m.group(1)] = parse_syscall_NRs(filename) | 
|  |  | 
|  | gen_policy(name_modifier=args.name_modifier, out_dir=args.out_dir, | 
|  | syscall_NRs=syscall_NRs, base_syscall_file=args.base_file, | 
|  | syscall_files=args.files) | 
|  |  | 
|  |  | 
|  | if __name__ == "__main__": | 
|  | main() |