| #!/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. |
| """Unittests for the compiler module.""" |
| |
| from __future__ import print_function |
| |
| import os |
| import random |
| import shutil |
| import tempfile |
| import unittest |
| |
| import arch |
| import bpf |
| import compiler |
| import parser # pylint: disable=wrong-import-order |
| |
| ARCH_64 = arch.Arch.load_from_json( |
| os.path.join( |
| os.path.dirname(os.path.abspath(__file__)), 'testdata/arch_64.json')) |
| |
| |
| class CompileFilterStatementTests(unittest.TestCase): |
| """Tests for PolicyCompiler.compile_filter_statement.""" |
| |
| def setUp(self): |
| self.arch = ARCH_64 |
| self.compiler = compiler.PolicyCompiler(self.arch) |
| |
| def _compile(self, line): |
| with tempfile.NamedTemporaryFile(mode='w') as policy_file: |
| policy_file.write(line) |
| policy_file.flush() |
| policy_parser = parser.PolicyParser( |
| self.arch, kill_action=bpf.KillProcess()) |
| parsed_policy = policy_parser.parse_file(policy_file.name) |
| assert len(parsed_policy.filter_statements) == 1 |
| return self.compiler.compile_filter_statement( |
| parsed_policy.filter_statements[0], |
| kill_action=bpf.KillProcess()) |
| |
| def test_allow(self): |
| """Accept lines where the syscall is accepted unconditionally.""" |
| block = self._compile('read: allow') |
| self.assertEqual(block.filter, None) |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 0)[1], 'ALLOW') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 1)[1], 'ALLOW') |
| |
| def test_arg0_eq_generated_code(self): |
| """Accept lines with an argument filter with ==.""" |
| block = self._compile('read: arg0 == 0x100') |
| # It might be a bit brittle to check the generated code in each test |
| # case instead of just the behavior, but there should be at least one |
| # test where this happens. |
| self.assertEqual( |
| block.filter.instructions, |
| [ |
| bpf.SockFilter(bpf.BPF_LD | bpf.BPF_W | bpf.BPF_ABS, 0, 0, |
| bpf.arg_offset(0, True)), |
| # Jump to KILL_PROCESS if the high word does not match. |
| bpf.SockFilter(bpf.BPF_JMP | bpf.BPF_JEQ | bpf.BPF_K, 0, 2, 0), |
| bpf.SockFilter(bpf.BPF_LD | bpf.BPF_W | bpf.BPF_ABS, 0, 0, |
| bpf.arg_offset(0, False)), |
| # Jump to KILL_PROCESS if the low word does not match. |
| bpf.SockFilter(bpf.BPF_JMP | bpf.BPF_JEQ | bpf.BPF_K, 1, 0, |
| 0x100), |
| bpf.SockFilter(bpf.BPF_RET, 0, 0, |
| bpf.SECCOMP_RET_KILL_PROCESS), |
| bpf.SockFilter(bpf.BPF_RET, 0, 0, bpf.SECCOMP_RET_ALLOW), |
| ]) |
| |
| def test_arg0_comparison_operators(self): |
| """Accept lines with an argument filter with comparison operators.""" |
| biases = (-1, 0, 1) |
| # For each operator, store the expectations of simulating the program |
| # against the constant plus each entry from the |biases| array. |
| cases = ( |
| ('==', ('KILL_PROCESS', 'ALLOW', 'KILL_PROCESS')), |
| ('!=', ('ALLOW', 'KILL_PROCESS', 'ALLOW')), |
| ('<', ('ALLOW', 'KILL_PROCESS', 'KILL_PROCESS')), |
| ('<=', ('ALLOW', 'ALLOW', 'KILL_PROCESS')), |
| ('>', ('KILL_PROCESS', 'KILL_PROCESS', 'ALLOW')), |
| ('>=', ('KILL_PROCESS', 'ALLOW', 'ALLOW')), |
| ) |
| for operator, expectations in cases: |
| block = self._compile('read: arg0 %s 0x100' % operator) |
| |
| # Check the filter's behavior. |
| for bias, expectation in zip(biases, expectations): |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, |
| self.arch.syscalls['read'], |
| 0x100 + bias)[1], expectation) |
| |
| def test_arg0_mask_operator(self): |
| """Accept lines with an argument filter with &.""" |
| block = self._compile('read: arg0 & 0x3') |
| |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 0)[1], 'KILL_PROCESS') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 1)[1], 'ALLOW') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 2)[1], 'ALLOW') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 3)[1], 'ALLOW') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 4)[1], 'KILL_PROCESS') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 5)[1], 'ALLOW') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 6)[1], 'ALLOW') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 7)[1], 'ALLOW') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 8)[1], 'KILL_PROCESS') |
| |
| def test_arg0_in_operator(self): |
| """Accept lines with an argument filter with in.""" |
| block = self._compile('read: arg0 in 0x3') |
| |
| # The 'in' operator only ensures that no bits outside the mask are set, |
| # which means that 0 is always allowed. |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 0)[1], 'ALLOW') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 1)[1], 'ALLOW') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 2)[1], 'ALLOW') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 3)[1], 'ALLOW') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 4)[1], 'KILL_PROCESS') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 5)[1], 'KILL_PROCESS') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 6)[1], 'KILL_PROCESS') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 7)[1], 'KILL_PROCESS') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 8)[1], 'KILL_PROCESS') |
| |
| def test_arg0_short_gt_ge_comparisons(self): |
| """Ensure that the short comparison optimization kicks in.""" |
| if self.arch.bits == 32: |
| return |
| short_constant_str = '0xdeadbeef' |
| short_constant = int(short_constant_str, base=0) |
| long_constant_str = '0xbadc0ffee0ddf00d' |
| long_constant = int(long_constant_str, base=0) |
| biases = (-1, 0, 1) |
| # For each operator, store the expectations of simulating the program |
| # against the constant plus each entry from the |biases| array. |
| cases = ( |
| ('<', ('ALLOW', 'KILL_PROCESS', 'KILL_PROCESS')), |
| ('<=', ('ALLOW', 'ALLOW', 'KILL_PROCESS')), |
| ('>', ('KILL_PROCESS', 'KILL_PROCESS', 'ALLOW')), |
| ('>=', ('KILL_PROCESS', 'ALLOW', 'ALLOW')), |
| ) |
| for operator, expectations in cases: |
| short_block = self._compile( |
| 'read: arg0 %s %s' % (operator, short_constant_str)) |
| long_block = self._compile( |
| 'read: arg0 %s %s' % (operator, long_constant_str)) |
| |
| # Check that the emitted code is shorter when the high word of the |
| # constant is zero. |
| self.assertLess( |
| len(short_block.filter.instructions), |
| len(long_block.filter.instructions)) |
| |
| # Check the filter's behavior. |
| for bias, expectation in zip(biases, expectations): |
| self.assertEqual( |
| long_block.simulate(self.arch.arch_nr, |
| self.arch.syscalls['read'], |
| long_constant + bias)[1], expectation) |
| self.assertEqual( |
| short_block.simulate( |
| self.arch.arch_nr, self.arch.syscalls['read'], |
| short_constant + bias)[1], expectation) |
| |
| def test_and_or(self): |
| """Accept lines with a complex expression in DNF.""" |
| block = self._compile('read: arg0 == 0 && arg1 == 0 || arg0 == 1') |
| |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], 0, |
| 0)[1], 'ALLOW') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], 0, |
| 1)[1], 'KILL_PROCESS') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], 1, |
| 0)[1], 'ALLOW') |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], 1, |
| 1)[1], 'ALLOW') |
| |
| def test_trap(self): |
| """Accept lines that trap unconditionally.""" |
| block = self._compile('read: trap') |
| |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 0)[1], 'TRAP') |
| |
| def test_ret_errno(self): |
| """Accept lines that return errno.""" |
| block = self._compile('read : arg0 == 0 || arg0 == 1 ; return 1') |
| |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 0)[1:], ('ERRNO', 1)) |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 1)[1:], ('ERRNO', 1)) |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 2)[1], 'KILL_PROCESS') |
| |
| def test_ret_errno_unconditionally(self): |
| """Accept lines that return errno unconditionally.""" |
| block = self._compile('read: return 1') |
| |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 0)[1:], ('ERRNO', 1)) |
| |
| def test_trace(self): |
| """Accept lines that trace unconditionally.""" |
| block = self._compile('read: trace') |
| |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 0)[1], 'TRACE') |
| |
| def test_log(self): |
| """Accept lines that log unconditionally.""" |
| block = self._compile('read: log') |
| |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], |
| 0)[1], 'LOG') |
| |
| def test_mmap_write_xor_exec(self): |
| """Accept the idiomatic filter for mmap.""" |
| block = self._compile( |
| 'read : arg0 in ~PROT_WRITE || arg0 in ~PROT_EXEC') |
| |
| prot_exec_and_write = 6 |
| for prot in range(0, 0xf): |
| if (prot & prot_exec_and_write) == prot_exec_and_write: |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, |
| self.arch.syscalls['read'], prot)[1], |
| 'KILL_PROCESS') |
| else: |
| self.assertEqual( |
| block.simulate(self.arch.arch_nr, |
| self.arch.syscalls['read'], prot)[1], |
| 'ALLOW') |
| |
| |
| class CompileFileTests(unittest.TestCase): |
| """Tests for PolicyCompiler.compile_file.""" |
| |
| def setUp(self): |
| self.arch = ARCH_64 |
| self.compiler = compiler.PolicyCompiler(self.arch) |
| self.tempdir = tempfile.mkdtemp() |
| |
| def tearDown(self): |
| shutil.rmtree(self.tempdir) |
| |
| def _write_file(self, filename, contents): |
| """Helper to write out a file for testing.""" |
| path = os.path.join(self.tempdir, filename) |
| with open(path, 'w') as outf: |
| outf.write(contents) |
| return path |
| |
| def test_compile(self): |
| """Ensure compilation works with all strategies.""" |
| self._write_file( |
| 'test.frequency', """ |
| read: 1 |
| close: 10 |
| """) |
| path = self._write_file( |
| 'test.policy', """ |
| @frequency ./test.frequency |
| read: 1 |
| close: 1 |
| """) |
| |
| program = self.compiler.compile_file( |
| path, |
| optimization_strategy=compiler.OptimizationStrategy.LINEAR, |
| kill_action=bpf.KillProcess()) |
| self.assertGreater( |
| bpf.simulate(program.instructions, self.arch.arch_nr, |
| self.arch.syscalls['read'], 0)[0], |
| bpf.simulate(program.instructions, self.arch.arch_nr, |
| self.arch.syscalls['close'], 0)[0], |
| ) |
| |
| def test_compile_bst(self): |
| """Ensure compilation with BST is cheaper than the linear model.""" |
| self._write_file( |
| 'test.frequency', """ |
| read: 1 |
| close: 10 |
| """) |
| path = self._write_file( |
| 'test.policy', """ |
| @frequency ./test.frequency |
| read: 1 |
| close: 1 |
| """) |
| |
| for strategy in list(compiler.OptimizationStrategy): |
| program = self.compiler.compile_file( |
| path, |
| optimization_strategy=strategy, |
| kill_action=bpf.KillProcess()) |
| self.assertGreater( |
| bpf.simulate(program.instructions, self.arch.arch_nr, |
| self.arch.syscalls['read'], 0)[0], |
| bpf.simulate(program.instructions, self.arch.arch_nr, |
| self.arch.syscalls['close'], 0)[0], |
| ) |
| self.assertEqual( |
| bpf.simulate(program.instructions, self.arch.arch_nr, |
| self.arch.syscalls['read'], 0)[1], 'ALLOW') |
| self.assertEqual( |
| bpf.simulate(program.instructions, self.arch.arch_nr, |
| self.arch.syscalls['close'], 0)[1], 'ALLOW') |
| |
| def test_compile_empty_file(self): |
| """Accept empty files.""" |
| path = self._write_file( |
| 'test.policy', """ |
| @default kill-thread |
| """) |
| |
| for strategy in list(compiler.OptimizationStrategy): |
| program = self.compiler.compile_file( |
| path, |
| optimization_strategy=strategy, |
| kill_action=bpf.KillProcess()) |
| self.assertEqual( |
| bpf.simulate(program.instructions, self.arch.arch_nr, |
| self.arch.syscalls['read'], 0)[1], 'KILL_THREAD') |
| |
| def test_compile_simulate(self): |
| """Ensure policy reflects script by testing some random scripts.""" |
| iterations = 5 |
| for i in range(iterations): |
| num_entries = 64 * (i + 1) // iterations |
| syscalls = dict( |
| zip( |
| random.sample(self.arch.syscalls.keys(), num_entries), |
| (random.randint(1, 1024) for _ in range(num_entries)), |
| )) |
| |
| frequency_contents = '\n'.join( |
| '%s: %d' % s for s in syscalls.items()) |
| policy_contents = '@frequency ./test.frequency\n' + '\n'.join( |
| '%s: 1' % s[0] for s in syscalls.items()) |
| |
| self._write_file('test.frequency', frequency_contents) |
| path = self._write_file('test.policy', policy_contents) |
| |
| for strategy in list(compiler.OptimizationStrategy): |
| program = self.compiler.compile_file( |
| path, |
| optimization_strategy=strategy, |
| kill_action=bpf.KillProcess()) |
| for name, number in self.arch.syscalls.items(): |
| expected_result = ('ALLOW' |
| if name in syscalls else 'KILL_PROCESS') |
| self.assertEqual( |
| bpf.simulate(program.instructions, self.arch.arch_nr, |
| number, 0)[1], expected_result, |
| ('syscall name: %s, syscall number: %d, ' |
| 'strategy: %s, policy:\n%s') % |
| (name, number, strategy, policy_contents)) |
| |
| @unittest.skipIf(not int(os.getenv('SLOW_TESTS', '0')), 'slow') |
| def test_compile_huge_policy(self): |
| """Ensure jumps while compiling a huge policy are still valid.""" |
| # Given that the BST strategy is O(n^3), don't choose a crazy large |
| # value, but it still needs to be around 128 so that we exercise the |
| # codegen paths that depend on the length of the jump. |
| # |
| # Immediate jump offsets in BPF comparison instructions are limited to |
| # 256 instructions, so given that every syscall filter consists of a |
| # load and jump instructions, with 128 syscalls there will be at least |
| # one jump that's further than 256 instructions. |
| num_entries = 128 |
| syscalls = dict(random.sample(self.arch.syscalls.items(), num_entries)) |
| # Here we force every single filter to be distinct. Otherwise the |
| # codegen layer will coalesce filters that compile to the same |
| # instructions. |
| policy_contents = '\n'.join( |
| '%s: arg0 == %d' % s for s in syscalls.items()) |
| |
| path = self._write_file('test.policy', policy_contents) |
| |
| program = self.compiler.compile_file( |
| path, |
| optimization_strategy=compiler.OptimizationStrategy.BST, |
| kill_action=bpf.KillProcess()) |
| for name, number in self.arch.syscalls.items(): |
| expected_result = ('ALLOW' |
| if name in syscalls else 'KILL_PROCESS') |
| self.assertEqual( |
| bpf.simulate(program.instructions, self.arch.arch_nr, |
| self.arch.syscalls[name], number)[1], |
| expected_result) |
| self.assertEqual( |
| bpf.simulate(program.instructions, self.arch.arch_nr, |
| self.arch.syscalls[name], number + 1)[1], |
| 'KILL_PROCESS') |
| |
| def test_compile_huge_filter(self): |
| """Ensure jumps while compiling a huge policy are still valid.""" |
| # This is intended to force cases where the AST visitation would result |
| # in a combinatorial explosion of calls to Block.accept(). An optimized |
| # implementation should be O(n). |
| num_entries = 128 |
| syscalls = {} |
| # Here we force every single filter to be distinct. Otherwise the |
| # codegen layer will coalesce filters that compile to the same |
| # instructions. |
| policy_contents = [] |
| for name in random.sample(self.arch.syscalls.keys(), num_entries): |
| values = random.sample(range(1024), num_entries) |
| syscalls[name] = values |
| policy_contents.append( |
| '%s: %s' % (name, ' || '.join('arg0 == %d' % value |
| for value in values))) |
| |
| path = self._write_file('test.policy', '\n'.join(policy_contents)) |
| |
| program = self.compiler.compile_file( |
| path, |
| optimization_strategy=compiler.OptimizationStrategy.LINEAR, |
| kill_action=bpf.KillProcess()) |
| for name, values in syscalls.items(): |
| self.assertEqual( |
| bpf.simulate(program.instructions, |
| self.arch.arch_nr, self.arch.syscalls[name], |
| random.choice(values))[1], 'ALLOW') |
| self.assertEqual( |
| bpf.simulate(program.instructions, self.arch.arch_nr, |
| self.arch.syscalls[name], 1025)[1], |
| 'KILL_PROCESS') |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
