tools/compiler.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# Copyright (C) 2018 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.
"""A BPF compiler for the Minijail policy file."""

from __future__ import print_function

import enum

import bpf
import parser  # pylint: disable=wrong-import-order


class OptimizationStrategy(enum.Enum):
    """The available optimization strategies."""

    # Generate a linear chain of syscall number checks. Works best for policies
    # with very few syscalls.
    LINEAR = 'linear'

    # Generate a binary search tree for the syscalls. Works best for policies
    # with a lot of syscalls, where no one syscall dominates.
    BST = 'bst'

    def __str__(self):
        return self.value


class SyscallPolicyEntry:
    """The parsed version of a seccomp policy line."""

    def __init__(self, name, number, frequency):
        self.name = name
        self.number = number
        self.frequency = frequency
        self.accumulated = 0
        self.filter = None

    def __repr__(self):
        return ('SyscallPolicyEntry<name: %s, number: %d, '
                'frequency: %d, filter: %r>') % (self.name, self.number,
                                                 self.frequency,
                                                 self.filter.instructions
                                                 if self.filter else None)

    def simulate(self, arch, syscall_number, *args):
        """Simulate the policy with the given arguments."""
        if not self.filter:
            return (0, 'ALLOW')
        return bpf.simulate(self.filter.instructions, arch, syscall_number,
                            *args)


class SyscallPolicyRange:
    """A contiguous range of SyscallPolicyEntries that have the same action."""

    def __init__(self, *entries):
        self.numbers = (entries[0].number, entries[-1].number + 1)
        self.frequency = sum(e.frequency for e in entries)
        self.accumulated = 0
        self.filter = entries[0].filter

    def __repr__(self):
        return 'SyscallPolicyRange<numbers: %r, frequency: %d, filter: %r>' % (
            self.numbers, self.frequency, self.filter.instructions
            if self.filter else None)

    def simulate(self, arch, syscall_number, *args):
        """Simulate the policy with the given arguments."""
        if not self.filter:
            return (0, 'ALLOW')
        return self.filter.simulate(arch, syscall_number, *args)


def _convert_to_ranges(entries):
    entries = list(sorted(entries, key=lambda r: r.number))
    lower = 0
    while lower < len(entries):
        upper = lower + 1
        while upper < len(entries):
            if entries[upper - 1].filter != entries[upper].filter:
                break
            if entries[upper - 1].number + 1 != entries[upper].number:
                break
            upper += 1
        yield SyscallPolicyRange(*entries[lower:upper])
        lower = upper


def _compile_single_range(entry,
                          accept_action,
                          reject_action,
                          visitor,
                          lower_bound=0,
                          upper_bound=1e99):
    action = accept_action
    if entry.filter:
        entry.filter.accept(visitor)
        action = entry.filter
    if entry.numbers[1] - entry.numbers[0] == 1:
        # Single syscall.
        # Accept if |X == nr|.
        return bpf.SyscallEntry(
            entry.numbers[0], action, reject_action, op=bpf.BPF_JEQ)
    elif entry.numbers[0] == lower_bound:
        # Syscall range aligned with the lower bound.
        # Accept if |X < nr[1]|.
        return bpf.SyscallEntry(
            entry.numbers[1], reject_action, action, op=bpf.BPF_JGE)
    elif entry.numbers[1] == upper_bound:
        # Syscall range aligned with the upper bound.
        # Accept if |X >= nr[0]|.
        return bpf.SyscallEntry(
            entry.numbers[0], action, reject_action, op=bpf.BPF_JGE)
    # Syscall range in the middle.
    # Accept if |nr[0] <= X < nr[1]|.
    upper_entry = bpf.SyscallEntry(
        entry.numbers[1], reject_action, action, op=bpf.BPF_JGE)
    return bpf.SyscallEntry(
        entry.numbers[0], upper_entry, reject_action, op=bpf.BPF_JGE)


def _compile_entries_linear(entries, accept_action, reject_action, visitor):
    # Compiles the list of entries into a simple linear list of comparisons. In
    # order to make the generated code a bit more efficient, we sort the
    # entries by frequency, so that the most frequently-called syscalls appear
    # earlier in the chain.
    next_action = reject_action
    ranges = sorted(_convert_to_ranges(entries), key=lambda r: -r.frequency)
    for entry in ranges[::-1]:
        next_action = _compile_single_range(entry, accept_action, next_action,
                                            visitor)
    return next_action


def _compile_entries_bst(entries, accept_action, reject_action, visitor):
    # Instead of generating a linear list of comparisons, this method generates
    # a binary search tree.
    #
    # Even though we are going to perform a binary search over the syscall
    # number, we would still like to rotate some of the internal nodes of the
    # binary search tree so that more frequently-used syscalls can be accessed
    # more cheaply (i.e. fewer internal nodes need to be traversed to reach
    # them).
    #
    # The overall idea then is to, at any step, instead of naively partitioning
    # the list of syscalls by the midpoint of the interval, we choose a
    # midpoint that minimizes the difference of the sum of all frequencies
    # between the left and right subtrees. For that, we need to sort the
    # entries by syscall number and keep track of the accumulated frequency of
    # all entries prior to the current one so that we can compue the midpoint
    # efficiently.
    #
    # TODO(lhchavez): There is one further possible optimization, which is to
    # hoist any syscalls that are more frequent than all other syscalls in the
    # BST combined into a linear chain before entering the BST.
    ranges = list(_convert_to_ranges(entries))

    accumulated = 0
    for entry in ranges:
        accumulated += entry.frequency
        entry.accumulated = accumulated

    # Recursively create the internal nodes.
    def _generate_syscall_bst(ranges, lower_bound=0, upper_bound=2**64 - 1):
        assert ranges
        if len(ranges) == 1:
            # This is a single syscall entry range, but the interval we are
            # currently considering contains other syscalls that we want to
            # reject. So instead of an internal node, create one or more leaf
            # nodes that check the range.
            assert lower_bound < upper_bound
            return _compile_single_range(ranges[0], accept_action,
                                         reject_action, visitor, lower_bound,
                                         upper_bound)

        # Find the midpoint that minimizes the difference between accumulated
        # costs in the left and right subtrees.
        previous_accumulated = ranges[0].accumulated - ranges[0].frequency
        last_accumulated = ranges[-1].accumulated - previous_accumulated
        best = (1e99, -1)
        for i, entry in enumerate(ranges):
            if not i:
                continue
            left_accumulated = entry.accumulated - previous_accumulated
            right_accumulated = last_accumulated - left_accumulated
            best = min(best, (abs(left_accumulated - right_accumulated), i))
        midpoint = best[1]
        assert midpoint >= 1, best

        cutoff_bound = ranges[midpoint].numbers[0]

        # Now we build the right and left subtrees independently. If any of the
        # subtrees consist of a single entry _and_ the bounds are tight around
        # that entry (that is, the bounds contain _only_ the syscall we are
        # going to consider), we can avoid emitting a leaf node and instead
        # have the comparison jump directly into the action that would be taken
        # by the entry.
        if (cutoff_bound, upper_bound) == ranges[midpoint].numbers:
            if ranges[midpoint].filter:
                ranges[midpoint].filter.accept(visitor)
                right_subtree = ranges[midpoint].filter
            else:
                right_subtree = accept_action
        else:
            right_subtree = _generate_syscall_bst(ranges[midpoint:],
                                                  cutoff_bound, upper_bound)

        if (lower_bound, cutoff_bound) == ranges[midpoint - 1].numbers:
            if ranges[midpoint - 1].filter:
                ranges[midpoint - 1].filter.accept(visitor)
                left_subtree = ranges[midpoint - 1].filter
            else:
                left_subtree = accept_action
        else:
            left_subtree = _generate_syscall_bst(ranges[:midpoint],
                                                 lower_bound, cutoff_bound)

        # Finally, now that both subtrees have been generated, we can create
        # the internal node of the binary search tree.
        return bpf.SyscallEntry(
            cutoff_bound, right_subtree, left_subtree, op=bpf.BPF_JGE)

    return _generate_syscall_bst(ranges)


class PolicyCompiler:
    """A parser for the Minijail seccomp policy file format."""

    def __init__(self, arch):
        self._arch = arch

    def compile_file(self,
                     policy_filename,
                     *,
                     optimization_strategy,
                     kill_action,
                     include_depth_limit=10):
        """Return a compiled BPF program from the provided policy file."""
        policy_parser = parser.PolicyParser(
            self._arch,
            kill_action=kill_action,
            include_depth_limit=include_depth_limit)
        parsed_policy = policy_parser.parse_file(policy_filename)
        entries = [
            self.compile_filter_statement(
                filter_statement, kill_action=kill_action)
            for filter_statement in parsed_policy.filter_statements
        ]

        visitor = bpf.FlatteningVisitor(
            arch=self._arch, kill_action=kill_action)
        accept_action = bpf.Allow()
        reject_action = parsed_policy.default_action
        if entries:
            if optimization_strategy == OptimizationStrategy.BST:
                next_action = _compile_entries_bst(entries, accept_action,
                                                   reject_action, visitor)
            else:
                next_action = _compile_entries_linear(entries, accept_action,
                                                      reject_action, visitor)
            reject_action.accept(visitor)
            accept_action.accept(visitor)
            bpf.ValidateArch(next_action).accept(visitor)
        else:
            reject_action.accept(visitor)
            bpf.ValidateArch(reject_action).accept(visitor)
        return visitor.result

    def compile_filter_statement(self, filter_statement, *, kill_action):
        """Compile one parser.FilterStatement into BPF."""
        policy_entry = SyscallPolicyEntry(filter_statement.syscall.name,
                                          filter_statement.syscall.number,
                                          filter_statement.frequency)
        # In each step of the way, the false action is the one that is taken if
        # the immediate boolean condition does not match. This means that the
        # false action taken here is the one that applies if the whole
        # expression fails to match.
        false_action = filter_statement.filters[-1].action
        if false_action == bpf.Allow():
            return policy_entry
        # We then traverse the list of filters backwards since we want
        # the root of the DAG to be the very first boolean operation in
        # the filter chain.
        for filt in filter_statement.filters[:-1][::-1]:
            for disjunction in filt.expression:
                # This is the jump target of the very last comparison in the
                # conjunction. Given that any conjunction that succeeds should
                # make the whole expression succeed, make the very last
                # comparison jump to the accept action if it succeeds.
                true_action = filt.action
                for atom in disjunction:
                    block = bpf.Atom(atom.argument_index, atom.op, atom.value,
                                     true_action, false_action)
                    true_action = block
                false_action = true_action
        policy_filter = false_action

        # Lower all Atoms into WideAtoms.
        lowering_visitor = bpf.LoweringVisitor(arch=self._arch)
        policy_filter = lowering_visitor.process(policy_filter)

        # Flatten the IR DAG into a single BasicBlock.
        flattening_visitor = bpf.FlatteningVisitor(
            arch=self._arch, kill_action=kill_action)
        policy_filter.accept(flattening_visitor)
        policy_entry.filter = flattening_visitor.result
        return policy_entry