1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3#
4# Copyright (C) 2018 The Android Open Source Project
5#
6# Licensed under the Apache License, Version 2.0 (the "License");
7# you may not use this file except in compliance with the License.
8# You may obtain a copy of the License at
9#
10#      http://www.apache.org/licenses/LICENSE-2.0
11#
12# Unless required by applicable law or agreed to in writing, software
13# distributed under the License is distributed on an "AS IS" BASIS,
14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15# See the License for the specific language governing permissions and
16# limitations under the License.
17"""Unittests for the compiler module."""
18
19from __future__ import print_function
20
21import os
22import random
23import shutil
24import tempfile
25import unittest
26
27import arch
28import bpf
29import compiler
30import parser  # pylint: disable=wrong-import-order
31
32ARCH_64 = arch.Arch.load_from_json(
33    os.path.join(
34        os.path.dirname(os.path.abspath(__file__)), 'testdata/arch_64.json'))
35
36
37class CompileFilterStatementTests(unittest.TestCase):
38    """Tests for PolicyCompiler.compile_filter_statement."""
39
40    def setUp(self):
41        self.arch = ARCH_64
42        self.compiler = compiler.PolicyCompiler(self.arch)
43
44    def _compile(self, line):
45        with tempfile.NamedTemporaryFile(mode='w') as policy_file:
46            policy_file.write(line)
47            policy_file.flush()
48            policy_parser = parser.PolicyParser(
49                self.arch, kill_action=bpf.KillProcess())
50            parsed_policy = policy_parser.parse_file(policy_file.name)
51            assert len(parsed_policy.filter_statements) == 1
52            return self.compiler.compile_filter_statement(
53                parsed_policy.filter_statements[0],
54                kill_action=bpf.KillProcess())
55
56    def test_allow(self):
57        """Accept lines where the syscall is accepted unconditionally."""
58        block = self._compile('read: allow')
59        self.assertEqual(block.filter, None)
60        self.assertEqual(
61            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
62                           0)[1], 'ALLOW')
63        self.assertEqual(
64            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
65                           1)[1], 'ALLOW')
66
67    def test_arg0_eq_generated_code(self):
68        """Accept lines with an argument filter with ==."""
69        block = self._compile('read: arg0 == 0x100')
70        # It might be a bit brittle to check the generated code in each test
71        # case instead of just the behavior, but there should be at least one
72        # test where this happens.
73        self.assertEqual(
74            block.filter.instructions,
75            [
76                bpf.SockFilter(bpf.BPF_LD | bpf.BPF_W | bpf.BPF_ABS, 0, 0,
77                               bpf.arg_offset(0, True)),
78                # Jump to KILL_PROCESS if the high word does not match.
79                bpf.SockFilter(bpf.BPF_JMP | bpf.BPF_JEQ | bpf.BPF_K, 0, 2, 0),
80                bpf.SockFilter(bpf.BPF_LD | bpf.BPF_W | bpf.BPF_ABS, 0, 0,
81                               bpf.arg_offset(0, False)),
82                # Jump to KILL_PROCESS if the low word does not match.
83                bpf.SockFilter(bpf.BPF_JMP | bpf.BPF_JEQ | bpf.BPF_K, 1, 0,
84                               0x100),
85                bpf.SockFilter(bpf.BPF_RET, 0, 0,
86                               bpf.SECCOMP_RET_KILL_PROCESS),
87                bpf.SockFilter(bpf.BPF_RET, 0, 0, bpf.SECCOMP_RET_ALLOW),
88            ])
89
90    def test_arg0_comparison_operators(self):
91        """Accept lines with an argument filter with comparison operators."""
92        biases = (-1, 0, 1)
93        # For each operator, store the expectations of simulating the program
94        # against the constant plus each entry from the |biases| array.
95        cases = (
96            ('==', ('KILL_PROCESS', 'ALLOW', 'KILL_PROCESS')),
97            ('!=', ('ALLOW', 'KILL_PROCESS', 'ALLOW')),
98            ('<', ('ALLOW', 'KILL_PROCESS', 'KILL_PROCESS')),
99            ('<=', ('ALLOW', 'ALLOW', 'KILL_PROCESS')),
100            ('>', ('KILL_PROCESS', 'KILL_PROCESS', 'ALLOW')),
101            ('>=', ('KILL_PROCESS', 'ALLOW', 'ALLOW')),
102        )
103        for operator, expectations in cases:
104            block = self._compile('read: arg0 %s 0x100' % operator)
105
106            # Check the filter's behavior.
107            for bias, expectation in zip(biases, expectations):
108                self.assertEqual(
109                    block.simulate(self.arch.arch_nr,
110                                   self.arch.syscalls['read'],
111                                   0x100 + bias)[1], expectation)
112
113    def test_arg0_mask_operator(self):
114        """Accept lines with an argument filter with &."""
115        block = self._compile('read: arg0 & 0x3')
116
117        self.assertEqual(
118            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
119                           0)[1], 'KILL_PROCESS')
120        self.assertEqual(
121            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
122                           1)[1], 'ALLOW')
123        self.assertEqual(
124            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
125                           2)[1], 'ALLOW')
126        self.assertEqual(
127            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
128                           3)[1], 'ALLOW')
129        self.assertEqual(
130            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
131                           4)[1], 'KILL_PROCESS')
132        self.assertEqual(
133            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
134                           5)[1], 'ALLOW')
135        self.assertEqual(
136            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
137                           6)[1], 'ALLOW')
138        self.assertEqual(
139            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
140                           7)[1], 'ALLOW')
141        self.assertEqual(
142            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
143                           8)[1], 'KILL_PROCESS')
144
145    def test_arg0_in_operator(self):
146        """Accept lines with an argument filter with in."""
147        block = self._compile('read: arg0 in 0x3')
148
149        # The 'in' operator only ensures that no bits outside the mask are set,
150        # which means that 0 is always allowed.
151        self.assertEqual(
152            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
153                           0)[1], 'ALLOW')
154        self.assertEqual(
155            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
156                           1)[1], 'ALLOW')
157        self.assertEqual(
158            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
159                           2)[1], 'ALLOW')
160        self.assertEqual(
161            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
162                           3)[1], 'ALLOW')
163        self.assertEqual(
164            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
165                           4)[1], 'KILL_PROCESS')
166        self.assertEqual(
167            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
168                           5)[1], 'KILL_PROCESS')
169        self.assertEqual(
170            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
171                           6)[1], 'KILL_PROCESS')
172        self.assertEqual(
173            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
174                           7)[1], 'KILL_PROCESS')
175        self.assertEqual(
176            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
177                           8)[1], 'KILL_PROCESS')
178
179    def test_arg0_short_gt_ge_comparisons(self):
180        """Ensure that the short comparison optimization kicks in."""
181        if self.arch.bits == 32:
182            return
183        short_constant_str = '0xdeadbeef'
184        short_constant = int(short_constant_str, base=0)
185        long_constant_str = '0xbadc0ffee0ddf00d'
186        long_constant = int(long_constant_str, base=0)
187        biases = (-1, 0, 1)
188        # For each operator, store the expectations of simulating the program
189        # against the constant plus each entry from the |biases| array.
190        cases = (
191            ('<', ('ALLOW', 'KILL_PROCESS', 'KILL_PROCESS')),
192            ('<=', ('ALLOW', 'ALLOW', 'KILL_PROCESS')),
193            ('>', ('KILL_PROCESS', 'KILL_PROCESS', 'ALLOW')),
194            ('>=', ('KILL_PROCESS', 'ALLOW', 'ALLOW')),
195        )
196        for operator, expectations in cases:
197            short_block = self._compile(
198                'read: arg0 %s %s' % (operator, short_constant_str))
199            long_block = self._compile(
200                'read: arg0 %s %s' % (operator, long_constant_str))
201
202            # Check that the emitted code is shorter when the high word of the
203            # constant is zero.
204            self.assertLess(
205                len(short_block.filter.instructions),
206                len(long_block.filter.instructions))
207
208            # Check the filter's behavior.
209            for bias, expectation in zip(biases, expectations):
210                self.assertEqual(
211                    long_block.simulate(self.arch.arch_nr,
212                                        self.arch.syscalls['read'],
213                                        long_constant + bias)[1], expectation)
214                self.assertEqual(
215                    short_block.simulate(
216                        self.arch.arch_nr, self.arch.syscalls['read'],
217                        short_constant + bias)[1], expectation)
218
219    def test_and_or(self):
220        """Accept lines with a complex expression in DNF."""
221        block = self._compile('read: arg0 == 0 && arg1 == 0 || arg0 == 1')
222
223        self.assertEqual(
224            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], 0,
225                           0)[1], 'ALLOW')
226        self.assertEqual(
227            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], 0,
228                           1)[1], 'KILL_PROCESS')
229        self.assertEqual(
230            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], 1,
231                           0)[1], 'ALLOW')
232        self.assertEqual(
233            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'], 1,
234                           1)[1], 'ALLOW')
235
236    def test_trap(self):
237        """Accept lines that trap unconditionally."""
238        block = self._compile('read: trap')
239
240        self.assertEqual(
241            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
242                           0)[1], 'TRAP')
243
244    def test_ret_errno(self):
245        """Accept lines that return errno."""
246        block = self._compile('read : arg0 == 0 || arg0 == 1 ; return 1')
247
248        self.assertEqual(
249            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
250                           0)[1:], ('ERRNO', 1))
251        self.assertEqual(
252            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
253                           1)[1:], ('ERRNO', 1))
254        self.assertEqual(
255            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
256                           2)[1], 'KILL_PROCESS')
257
258    def test_ret_errno_unconditionally(self):
259        """Accept lines that return errno unconditionally."""
260        block = self._compile('read: return 1')
261
262        self.assertEqual(
263            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
264                           0)[1:], ('ERRNO', 1))
265
266    def test_trace(self):
267        """Accept lines that trace unconditionally."""
268        block = self._compile('read: trace')
269
270        self.assertEqual(
271            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
272                           0)[1], 'TRACE')
273
274    def test_log(self):
275        """Accept lines that log unconditionally."""
276        block = self._compile('read: log')
277
278        self.assertEqual(
279            block.simulate(self.arch.arch_nr, self.arch.syscalls['read'],
280                           0)[1], 'LOG')
281
282    def test_mmap_write_xor_exec(self):
283        """Accept the idiomatic filter for mmap."""
284        block = self._compile(
285            'read : arg0 in ~PROT_WRITE || arg0 in ~PROT_EXEC')
286
287        prot_exec_and_write = 6
288        for prot in range(0, 0xf):
289            if (prot & prot_exec_and_write) == prot_exec_and_write:
290                self.assertEqual(
291                    block.simulate(self.arch.arch_nr,
292                                   self.arch.syscalls['read'], prot)[1],
293                    'KILL_PROCESS')
294            else:
295                self.assertEqual(
296                    block.simulate(self.arch.arch_nr,
297                                   self.arch.syscalls['read'], prot)[1],
298                    'ALLOW')
299
300
301class CompileFileTests(unittest.TestCase):
302    """Tests for PolicyCompiler.compile_file."""
303
304    def setUp(self):
305        self.arch = ARCH_64
306        self.compiler = compiler.PolicyCompiler(self.arch)
307        self.tempdir = tempfile.mkdtemp()
308
309    def tearDown(self):
310        shutil.rmtree(self.tempdir)
311
312    def _write_file(self, filename, contents):
313        """Helper to write out a file for testing."""
314        path = os.path.join(self.tempdir, filename)
315        with open(path, 'w') as outf:
316            outf.write(contents)
317        return path
318
319    def test_compile(self):
320        """Ensure compilation works with all strategies."""
321        self._write_file(
322            'test.frequency', """
323            read: 1
324            close: 10
325        """)
326        path = self._write_file(
327            'test.policy', """
328            @frequency ./test.frequency
329            read: 1
330            close: 1
331        """)
332
333        program = self.compiler.compile_file(
334            path,
335            optimization_strategy=compiler.OptimizationStrategy.LINEAR,
336            kill_action=bpf.KillProcess())
337        self.assertGreater(
338            bpf.simulate(program.instructions, self.arch.arch_nr,
339                         self.arch.syscalls['read'], 0)[0],
340            bpf.simulate(program.instructions, self.arch.arch_nr,
341                         self.arch.syscalls['close'], 0)[0],
342        )
343
344    def test_compile_bst(self):
345        """Ensure compilation with BST is cheaper than the linear model."""
346        self._write_file(
347            'test.frequency', """
348            read: 1
349            close: 10
350        """)
351        path = self._write_file(
352            'test.policy', """
353            @frequency ./test.frequency
354            read: 1
355            close: 1
356        """)
357
358        for strategy in list(compiler.OptimizationStrategy):
359            program = self.compiler.compile_file(
360                path,
361                optimization_strategy=strategy,
362                kill_action=bpf.KillProcess())
363            self.assertGreater(
364                bpf.simulate(program.instructions, self.arch.arch_nr,
365                             self.arch.syscalls['read'], 0)[0],
366                bpf.simulate(program.instructions, self.arch.arch_nr,
367                             self.arch.syscalls['close'], 0)[0],
368            )
369            self.assertEqual(
370                bpf.simulate(program.instructions, self.arch.arch_nr,
371                             self.arch.syscalls['read'], 0)[1], 'ALLOW')
372            self.assertEqual(
373                bpf.simulate(program.instructions, self.arch.arch_nr,
374                             self.arch.syscalls['close'], 0)[1], 'ALLOW')
375
376    def test_compile_empty_file(self):
377        """Accept empty files."""
378        path = self._write_file(
379            'test.policy', """
380            @default kill-thread
381        """)
382
383        for strategy in list(compiler.OptimizationStrategy):
384            program = self.compiler.compile_file(
385                path,
386                optimization_strategy=strategy,
387                kill_action=bpf.KillProcess())
388            self.assertEqual(
389                bpf.simulate(program.instructions, self.arch.arch_nr,
390                             self.arch.syscalls['read'], 0)[1], 'KILL_THREAD')
391
392    def test_compile_simulate(self):
393        """Ensure policy reflects script by testing some random scripts."""
394        iterations = 5
395        for i in range(iterations):
396            num_entries = 64 * (i + 1) // iterations
397            syscalls = dict(
398                zip(
399                    random.sample(self.arch.syscalls.keys(), num_entries),
400                    (random.randint(1, 1024) for _ in range(num_entries)),
401                ))
402
403            frequency_contents = '\n'.join(
404                '%s: %d' % s for s in syscalls.items())
405            policy_contents = '@frequency ./test.frequency\n' + '\n'.join(
406                '%s: 1' % s[0] for s in syscalls.items())
407
408            self._write_file('test.frequency', frequency_contents)
409            path = self._write_file('test.policy', policy_contents)
410
411            for strategy in list(compiler.OptimizationStrategy):
412                program = self.compiler.compile_file(
413                    path,
414                    optimization_strategy=strategy,
415                    kill_action=bpf.KillProcess())
416                for name, number in self.arch.syscalls.items():
417                    expected_result = ('ALLOW'
418                                       if name in syscalls else 'KILL_PROCESS')
419                    self.assertEqual(
420                        bpf.simulate(program.instructions, self.arch.arch_nr,
421                                     number, 0)[1], expected_result,
422                        ('syscall name: %s, syscall number: %d, '
423                         'strategy: %s, policy:\n%s') %
424                        (name, number, strategy, policy_contents))
425
426    @unittest.skipIf(not int(os.getenv('SLOW_TESTS', '0')), 'slow')
427    def test_compile_huge_policy(self):
428        """Ensure jumps while compiling a huge policy are still valid."""
429        # Given that the BST strategy is O(n^3), don't choose a crazy large
430        # value, but it still needs to be around 128 so that we exercise the
431        # codegen paths that depend on the length of the jump.
432        #
433        # Immediate jump offsets in BPF comparison instructions are limited to
434        # 256 instructions, so given that every syscall filter consists of a
435        # load and jump instructions, with 128 syscalls there will be at least
436        # one jump that's further than 256 instructions.
437        num_entries = 128
438        syscalls = dict(random.sample(self.arch.syscalls.items(), num_entries))
439        # Here we force every single filter to be distinct. Otherwise the
440        # codegen layer will coalesce filters that compile to the same
441        # instructions.
442        policy_contents = '\n'.join(
443            '%s: arg0 == %d' % s for s in syscalls.items())
444
445        path = self._write_file('test.policy', policy_contents)
446
447        program = self.compiler.compile_file(
448            path,
449            optimization_strategy=compiler.OptimizationStrategy.BST,
450            kill_action=bpf.KillProcess())
451        for name, number in self.arch.syscalls.items():
452            expected_result = ('ALLOW'
453                               if name in syscalls else 'KILL_PROCESS')
454            self.assertEqual(
455                bpf.simulate(program.instructions, self.arch.arch_nr,
456                             self.arch.syscalls[name], number)[1],
457                expected_result)
458            self.assertEqual(
459                bpf.simulate(program.instructions, self.arch.arch_nr,
460                             self.arch.syscalls[name], number + 1)[1],
461                'KILL_PROCESS')
462
463    def test_compile_huge_filter(self):
464        """Ensure jumps while compiling a huge policy are still valid."""
465        # This is intended to force cases where the AST visitation would result
466        # in a combinatorial explosion of calls to Block.accept(). An optimized
467        # implementation should be O(n).
468        num_entries = 128
469        syscalls = {}
470        # Here we force every single filter to be distinct. Otherwise the
471        # codegen layer will coalesce filters that compile to the same
472        # instructions.
473        policy_contents = []
474        for name in random.sample(self.arch.syscalls.keys(), num_entries):
475            values = random.sample(range(1024), num_entries)
476            syscalls[name] = values
477            policy_contents.append(
478                '%s: %s' % (name, ' || '.join('arg0 == %d' % value
479                                              for value in values)))
480
481        path = self._write_file('test.policy', '\n'.join(policy_contents))
482
483        program = self.compiler.compile_file(
484            path,
485            optimization_strategy=compiler.OptimizationStrategy.LINEAR,
486            kill_action=bpf.KillProcess())
487        for name, values in syscalls.items():
488            self.assertEqual(
489                bpf.simulate(program.instructions,
490                             self.arch.arch_nr, self.arch.syscalls[name],
491                             random.choice(values))[1], 'ALLOW')
492            self.assertEqual(
493                bpf.simulate(program.instructions, self.arch.arch_nr,
494                             self.arch.syscalls[name], 1025)[1],
495                'KILL_PROCESS')
496
497
498if __name__ == '__main__':
499    unittest.main()
500