#!/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.
|
"""Tools to interact with BPF programs."""
|
|
import abc
|
import collections
|
import struct
|
|
# This comes from syscall(2). Most architectures only support passing 6 args to
|
# syscalls, but ARM supports passing 7.
|
MAX_SYSCALL_ARGUMENTS = 7
|
|
# The following fields were copied from <linux/bpf_common.h>:
|
|
# Instruction classes
|
BPF_LD = 0x00
|
BPF_LDX = 0x01
|
BPF_ST = 0x02
|
BPF_STX = 0x03
|
BPF_ALU = 0x04
|
BPF_JMP = 0x05
|
BPF_RET = 0x06
|
BPF_MISC = 0x07
|
|
# LD/LDX fields.
|
# Size
|
BPF_W = 0x00
|
BPF_H = 0x08
|
BPF_B = 0x10
|
# Mode
|
BPF_IMM = 0x00
|
BPF_ABS = 0x20
|
BPF_IND = 0x40
|
BPF_MEM = 0x60
|
BPF_LEN = 0x80
|
BPF_MSH = 0xa0
|
|
# JMP fields.
|
BPF_JA = 0x00
|
BPF_JEQ = 0x10
|
BPF_JGT = 0x20
|
BPF_JGE = 0x30
|
BPF_JSET = 0x40
|
|
# Source
|
BPF_K = 0x00
|
BPF_X = 0x08
|
|
BPF_MAXINSNS = 4096
|
|
# The following fields were copied from <linux/seccomp.h>:
|
|
SECCOMP_RET_KILL_PROCESS = 0x80000000
|
SECCOMP_RET_KILL_THREAD = 0x00000000
|
SECCOMP_RET_TRAP = 0x00030000
|
SECCOMP_RET_ERRNO = 0x00050000
|
SECCOMP_RET_TRACE = 0x7ff00000
|
SECCOMP_RET_LOG = 0x7ffc0000
|
SECCOMP_RET_ALLOW = 0x7fff0000
|
|
SECCOMP_RET_ACTION_FULL = 0xffff0000
|
SECCOMP_RET_DATA = 0x0000ffff
|
|
|
def arg_offset(arg_index, hi=False):
|
"""Return the BPF_LD|BPF_W|BPF_ABS addressing-friendly register offset."""
|
offsetof_args = 4 + 4 + 8
|
arg_width = 8
|
return offsetof_args + arg_width * arg_index + (arg_width // 2) * hi
|
|
|
def simulate(instructions, arch, syscall_number, *args):
|
"""Simulate a BPF program with the given arguments."""
|
args = ((args + (0, ) *
|
(MAX_SYSCALL_ARGUMENTS - len(args)))[:MAX_SYSCALL_ARGUMENTS])
|
input_memory = struct.pack('IIQ' + 'Q' * MAX_SYSCALL_ARGUMENTS,
|
syscall_number, arch, 0, *args)
|
|
register = 0
|
program_counter = 0
|
cost = 0
|
while program_counter < len(instructions):
|
ins = instructions[program_counter]
|
program_counter += 1
|
cost += 1
|
if ins.code == BPF_LD | BPF_W | BPF_ABS:
|
register = struct.unpack('I', input_memory[ins.k:ins.k + 4])[0]
|
elif ins.code == BPF_JMP | BPF_JA | BPF_K:
|
program_counter += ins.k
|
elif ins.code == BPF_JMP | BPF_JEQ | BPF_K:
|
if register == ins.k:
|
program_counter += ins.jt
|
else:
|
program_counter += ins.jf
|
elif ins.code == BPF_JMP | BPF_JGT | BPF_K:
|
if register > ins.k:
|
program_counter += ins.jt
|
else:
|
program_counter += ins.jf
|
elif ins.code == BPF_JMP | BPF_JGE | BPF_K:
|
if register >= ins.k:
|
program_counter += ins.jt
|
else:
|
program_counter += ins.jf
|
elif ins.code == BPF_JMP | BPF_JSET | BPF_K:
|
if register & ins.k != 0:
|
program_counter += ins.jt
|
else:
|
program_counter += ins.jf
|
elif ins.code == BPF_RET:
|
if ins.k == SECCOMP_RET_KILL_PROCESS:
|
return (cost, 'KILL_PROCESS')
|
if ins.k == SECCOMP_RET_KILL_THREAD:
|
return (cost, 'KILL_THREAD')
|
if ins.k == SECCOMP_RET_TRAP:
|
return (cost, 'TRAP')
|
if (ins.k & SECCOMP_RET_ACTION_FULL) == SECCOMP_RET_ERRNO:
|
return (cost, 'ERRNO', ins.k & SECCOMP_RET_DATA)
|
if ins.k == SECCOMP_RET_TRACE:
|
return (cost, 'TRACE')
|
if ins.k == SECCOMP_RET_LOG:
|
return (cost, 'LOG')
|
if ins.k == SECCOMP_RET_ALLOW:
|
return (cost, 'ALLOW')
|
raise Exception('unknown return %#x' % ins.k)
|
else:
|
raise Exception('unknown instruction %r' % (ins, ))
|
raise Exception('out-of-bounds')
|
|
|
class SockFilter(
|
collections.namedtuple('SockFilter', ['code', 'jt', 'jf', 'k'])):
|
"""A representation of struct sock_filter."""
|
|
__slots__ = ()
|
|
def encode(self):
|
"""Return an encoded version of the SockFilter."""
|
return struct.pack('HBBI', self.code, self.jt, self.jf, self.k)
|
|
|
class AbstractBlock(abc.ABC):
|
"""A class that implements the visitor pattern."""
|
|
def __init__(self):
|
super().__init__()
|
|
@abc.abstractmethod
|
def accept(self, visitor):
|
pass
|
|
|
class BasicBlock(AbstractBlock):
|
"""A concrete implementation of AbstractBlock that has been compiled."""
|
|
def __init__(self, instructions):
|
super().__init__()
|
self._instructions = instructions
|
|
def accept(self, visitor):
|
if visitor.visited(self):
|
return
|
visitor.visit(self)
|
|
@property
|
def instructions(self):
|
return self._instructions
|
|
@property
|
def opcodes(self):
|
return b''.join(i.encode() for i in self._instructions)
|
|
def __eq__(self, o):
|
if not isinstance(o, BasicBlock):
|
return False
|
return self._instructions == o._instructions
|
|
|
class KillProcess(BasicBlock):
|
"""A BasicBlock that unconditionally returns KILL_PROCESS."""
|
|
def __init__(self):
|
super().__init__(
|
[SockFilter(BPF_RET, 0x00, 0x00, SECCOMP_RET_KILL_PROCESS)])
|
|
|
class KillThread(BasicBlock):
|
"""A BasicBlock that unconditionally returns KILL_THREAD."""
|
|
def __init__(self):
|
super().__init__(
|
[SockFilter(BPF_RET, 0x00, 0x00, SECCOMP_RET_KILL_THREAD)])
|
|
|
class Trap(BasicBlock):
|
"""A BasicBlock that unconditionally returns TRAP."""
|
|
def __init__(self):
|
super().__init__([SockFilter(BPF_RET, 0x00, 0x00, SECCOMP_RET_TRAP)])
|
|
|
class Trace(BasicBlock):
|
"""A BasicBlock that unconditionally returns TRACE."""
|
|
def __init__(self):
|
super().__init__([SockFilter(BPF_RET, 0x00, 0x00, SECCOMP_RET_TRACE)])
|
|
|
class Log(BasicBlock):
|
"""A BasicBlock that unconditionally returns LOG."""
|
|
def __init__(self):
|
super().__init__([SockFilter(BPF_RET, 0x00, 0x00, SECCOMP_RET_LOG)])
|
|
|
class ReturnErrno(BasicBlock):
|
"""A BasicBlock that unconditionally returns the specified errno."""
|
|
def __init__(self, errno):
|
super().__init__([
|
SockFilter(BPF_RET, 0x00, 0x00,
|
SECCOMP_RET_ERRNO | (errno & SECCOMP_RET_DATA))
|
])
|
self.errno = errno
|
|
|
class Allow(BasicBlock):
|
"""A BasicBlock that unconditionally returns ALLOW."""
|
|
def __init__(self):
|
super().__init__([SockFilter(BPF_RET, 0x00, 0x00, SECCOMP_RET_ALLOW)])
|
|
|
class ValidateArch(AbstractBlock):
|
"""An AbstractBlock that validates the architecture."""
|
|
def __init__(self, next_block):
|
super().__init__()
|
self.next_block = next_block
|
|
def accept(self, visitor):
|
if visitor.visited(self):
|
return
|
self.next_block.accept(visitor)
|
visitor.visit(self)
|
|
|
class SyscallEntry(AbstractBlock):
|
"""An abstract block that represents a syscall comparison in a DAG."""
|
|
def __init__(self, syscall_number, jt, jf, *, op=BPF_JEQ):
|
super().__init__()
|
self.op = op
|
self.syscall_number = syscall_number
|
self.jt = jt
|
self.jf = jf
|
|
def __lt__(self, o):
|
# Defined because we want to compare tuples that contain SyscallEntries.
|
return False
|
|
def __gt__(self, o):
|
# Defined because we want to compare tuples that contain SyscallEntries.
|
return False
|
|
def accept(self, visitor):
|
if visitor.visited(self):
|
return
|
self.jt.accept(visitor)
|
self.jf.accept(visitor)
|
visitor.visit(self)
|
|
def __lt__(self, o):
|
# Defined because we want to compare tuples that contain SyscallEntries.
|
return False
|
|
def __gt__(self, o):
|
# Defined because we want to compare tuples that contain SyscallEntries.
|
return False
|
|
|
class WideAtom(AbstractBlock):
|
"""A BasicBlock that represents a 32-bit wide atom."""
|
|
def __init__(self, arg_offset, op, value, jt, jf):
|
super().__init__()
|
self.arg_offset = arg_offset
|
self.op = op
|
self.value = value
|
self.jt = jt
|
self.jf = jf
|
|
def accept(self, visitor):
|
if visitor.visited(self):
|
return
|
self.jt.accept(visitor)
|
self.jf.accept(visitor)
|
visitor.visit(self)
|
|
|
class Atom(AbstractBlock):
|
"""A BasicBlock that represents an atom (a simple comparison operation)."""
|
|
def __init__(self, arg_index, op, value, jt, jf):
|
super().__init__()
|
if op == '==':
|
op = BPF_JEQ
|
elif op == '!=':
|
op = BPF_JEQ
|
jt, jf = jf, jt
|
elif op == '>':
|
op = BPF_JGT
|
elif op == '<=':
|
op = BPF_JGT
|
jt, jf = jf, jt
|
elif op == '>=':
|
op = BPF_JGE
|
elif op == '<':
|
op = BPF_JGE
|
jt, jf = jf, jt
|
elif op == '&':
|
op = BPF_JSET
|
elif op == 'in':
|
op = BPF_JSET
|
# The mask is negated, so the comparison will be true when the
|
# argument includes a flag that wasn't listed in the original
|
# (non-negated) mask. This would be the failure case, so we switch
|
# |jt| and |jf|.
|
value = (~value) & ((1 << 64) - 1)
|
jt, jf = jf, jt
|
else:
|
raise Exception('Unknown operator %s' % op)
|
|
self.arg_index = arg_index
|
self.op = op
|
self.jt = jt
|
self.jf = jf
|
self.value = value
|
|
def accept(self, visitor):
|
if visitor.visited(self):
|
return
|
self.jt.accept(visitor)
|
self.jf.accept(visitor)
|
visitor.visit(self)
|
|
|
class AbstractVisitor(abc.ABC):
|
"""An abstract visitor."""
|
|
def __init__(self):
|
self._visited = set()
|
|
def visited(self, block):
|
if id(block) in self._visited:
|
return True
|
self._visited.add(id(block))
|
return False
|
|
def process(self, block):
|
block.accept(self)
|
return block
|
|
def visit(self, block):
|
if isinstance(block, KillProcess):
|
self.visitKillProcess(block)
|
elif isinstance(block, KillThread):
|
self.visitKillThread(block)
|
elif isinstance(block, Trap):
|
self.visitTrap(block)
|
elif isinstance(block, ReturnErrno):
|
self.visitReturnErrno(block)
|
elif isinstance(block, Trace):
|
self.visitTrace(block)
|
elif isinstance(block, Log):
|
self.visitLog(block)
|
elif isinstance(block, Allow):
|
self.visitAllow(block)
|
elif isinstance(block, BasicBlock):
|
self.visitBasicBlock(block)
|
elif isinstance(block, ValidateArch):
|
self.visitValidateArch(block)
|
elif isinstance(block, SyscallEntry):
|
self.visitSyscallEntry(block)
|
elif isinstance(block, WideAtom):
|
self.visitWideAtom(block)
|
elif isinstance(block, Atom):
|
self.visitAtom(block)
|
else:
|
raise Exception('Unknown block type: %r' % block)
|
|
@abc.abstractmethod
|
def visitKillProcess(self, block):
|
pass
|
|
@abc.abstractmethod
|
def visitKillThread(self, block):
|
pass
|
|
@abc.abstractmethod
|
def visitTrap(self, block):
|
pass
|
|
@abc.abstractmethod
|
def visitReturnErrno(self, block):
|
pass
|
|
@abc.abstractmethod
|
def visitTrace(self, block):
|
pass
|
|
@abc.abstractmethod
|
def visitLog(self, block):
|
pass
|
|
@abc.abstractmethod
|
def visitAllow(self, block):
|
pass
|
|
@abc.abstractmethod
|
def visitBasicBlock(self, block):
|
pass
|
|
@abc.abstractmethod
|
def visitValidateArch(self, block):
|
pass
|
|
@abc.abstractmethod
|
def visitSyscallEntry(self, block):
|
pass
|
|
@abc.abstractmethod
|
def visitWideAtom(self, block):
|
pass
|
|
@abc.abstractmethod
|
def visitAtom(self, block):
|
pass
|
|
|
class CopyingVisitor(AbstractVisitor):
|
"""A visitor that copies Blocks."""
|
|
def __init__(self):
|
super().__init__()
|
self._mapping = {}
|
|
def process(self, block):
|
self._mapping = {}
|
block.accept(self)
|
return self._mapping[id(block)]
|
|
def visitKillProcess(self, block):
|
assert id(block) not in self._mapping
|
self._mapping[id(block)] = KillProcess()
|
|
def visitKillThread(self, block):
|
assert id(block) not in self._mapping
|
self._mapping[id(block)] = KillThread()
|
|
def visitTrap(self, block):
|
assert id(block) not in self._mapping
|
self._mapping[id(block)] = Trap()
|
|
def visitReturnErrno(self, block):
|
assert id(block) not in self._mapping
|
self._mapping[id(block)] = ReturnErrno(block.errno)
|
|
def visitTrace(self, block):
|
assert id(block) not in self._mapping
|
self._mapping[id(block)] = Trace()
|
|
def visitLog(self, block):
|
assert id(block) not in self._mapping
|
self._mapping[id(block)] = Log()
|
|
def visitAllow(self, block):
|
assert id(block) not in self._mapping
|
self._mapping[id(block)] = Allow()
|
|
def visitBasicBlock(self, block):
|
assert id(block) not in self._mapping
|
self._mapping[id(block)] = BasicBlock(block.instructions)
|
|
def visitValidateArch(self, block):
|
assert id(block) not in self._mapping
|
self._mapping[id(block)] = ValidateArch(
|
block.arch, self._mapping[id(block.next_block)])
|
|
def visitSyscallEntry(self, block):
|
assert id(block) not in self._mapping
|
self._mapping[id(block)] = SyscallEntry(
|
block.syscall_number,
|
self._mapping[id(block.jt)],
|
self._mapping[id(block.jf)],
|
op=block.op)
|
|
def visitWideAtom(self, block):
|
assert id(block) not in self._mapping
|
self._mapping[id(block)] = WideAtom(
|
block.arg_offset, block.op, block.value, self._mapping[id(
|
block.jt)], self._mapping[id(block.jf)])
|
|
def visitAtom(self, block):
|
assert id(block) not in self._mapping
|
self._mapping[id(block)] = Atom(block.arg_index, block.op, block.value,
|
self._mapping[id(block.jt)],
|
self._mapping[id(block.jf)])
|
|
|
class LoweringVisitor(CopyingVisitor):
|
"""A visitor that lowers Atoms into WideAtoms."""
|
|
def __init__(self, *, arch):
|
super().__init__()
|
self._bits = arch.bits
|
|
def visitAtom(self, block):
|
assert id(block) not in self._mapping
|
|
lo = block.value & 0xFFFFFFFF
|
hi = (block.value >> 32) & 0xFFFFFFFF
|
|
lo_block = WideAtom(
|
arg_offset(block.arg_index, False), block.op, lo,
|
self._mapping[id(block.jt)], self._mapping[id(block.jf)])
|
|
if self._bits == 32:
|
self._mapping[id(block)] = lo_block
|
return
|
|
if block.op in (BPF_JGE, BPF_JGT):
|
# hi_1,lo_1 <op> hi_2,lo_2
|
#
|
# hi_1 > hi_2 || hi_1 == hi_2 && lo_1 <op> lo_2
|
if hi == 0:
|
# Special case: it's not needed to check whether |hi_1 == hi_2|,
|
# because it's true iff the JGT test fails.
|
self._mapping[id(block)] = WideAtom(
|
arg_offset(block.arg_index, True), BPF_JGT, hi,
|
self._mapping[id(block.jt)], lo_block)
|
return
|
hi_eq_block = WideAtom(
|
arg_offset(block.arg_index, True), BPF_JEQ, hi, lo_block,
|
self._mapping[id(block.jf)])
|
self._mapping[id(block)] = WideAtom(
|
arg_offset(block.arg_index, True), BPF_JGT, hi,
|
self._mapping[id(block.jt)], hi_eq_block)
|
return
|
if block.op == BPF_JSET:
|
# hi_1,lo_1 & hi_2,lo_2
|
#
|
# hi_1 & hi_2 || lo_1 & lo_2
|
if hi == 0:
|
# Special case: |hi_1 & hi_2| will never be True, so jump
|
# directly into the |lo_1 & lo_2| case.
|
self._mapping[id(block)] = lo_block
|
return
|
self._mapping[id(block)] = WideAtom(
|
arg_offset(block.arg_index, True), block.op, hi,
|
self._mapping[id(block.jt)], lo_block)
|
return
|
|
assert block.op == BPF_JEQ, block.op
|
|
# hi_1,lo_1 == hi_2,lo_2
|
#
|
# hi_1 == hi_2 && lo_1 == lo_2
|
self._mapping[id(block)] = WideAtom(
|
arg_offset(block.arg_index, True), block.op, hi, lo_block,
|
self._mapping[id(block.jf)])
|
|
|
class FlatteningVisitor:
|
"""A visitor that flattens a DAG of Block objects."""
|
|
def __init__(self, *, arch, kill_action):
|
self._visited = set()
|
self._kill_action = kill_action
|
self._instructions = []
|
self._arch = arch
|
self._offsets = {}
|
|
@property
|
def result(self):
|
return BasicBlock(self._instructions)
|
|
def _distance(self, block):
|
distance = self._offsets[id(block)] + len(self._instructions)
|
assert distance >= 0
|
return distance
|
|
def _emit_load_arg(self, offset):
|
return [SockFilter(BPF_LD | BPF_W | BPF_ABS, 0, 0, offset)]
|
|
def _emit_jmp(self, op, value, jt_distance, jf_distance):
|
if jt_distance < 0x100 and jf_distance < 0x100:
|
return [
|
SockFilter(BPF_JMP | op | BPF_K, jt_distance, jf_distance,
|
value),
|
]
|
if jt_distance + 1 < 0x100:
|
return [
|
SockFilter(BPF_JMP | op | BPF_K, jt_distance + 1, 0, value),
|
SockFilter(BPF_JMP | BPF_JA, 0, 0, jf_distance),
|
]
|
if jf_distance + 1 < 0x100:
|
return [
|
SockFilter(BPF_JMP | op | BPF_K, 0, jf_distance + 1, value),
|
SockFilter(BPF_JMP | BPF_JA, 0, 0, jt_distance),
|
]
|
return [
|
SockFilter(BPF_JMP | op | BPF_K, 0, 1, value),
|
SockFilter(BPF_JMP | BPF_JA, 0, 0, jt_distance + 1),
|
SockFilter(BPF_JMP | BPF_JA, 0, 0, jf_distance),
|
]
|
|
def visited(self, block):
|
if id(block) in self._visited:
|
return True
|
self._visited.add(id(block))
|
return False
|
|
def visit(self, block):
|
assert id(block) not in self._offsets
|
|
if isinstance(block, BasicBlock):
|
instructions = block.instructions
|
elif isinstance(block, ValidateArch):
|
instructions = [
|
SockFilter(BPF_LD | BPF_W | BPF_ABS, 0, 0, 4),
|
SockFilter(BPF_JMP | BPF_JEQ | BPF_K,
|
self._distance(block.next_block) + 1, 0,
|
self._arch.arch_nr),
|
] + self._kill_action.instructions + [
|
SockFilter(BPF_LD | BPF_W | BPF_ABS, 0, 0, 0),
|
]
|
elif isinstance(block, SyscallEntry):
|
instructions = self._emit_jmp(block.op, block.syscall_number,
|
self._distance(block.jt),
|
self._distance(block.jf))
|
elif isinstance(block, WideAtom):
|
instructions = (
|
self._emit_load_arg(block.arg_offset) + self._emit_jmp(
|
block.op, block.value, self._distance(block.jt),
|
self._distance(block.jf)))
|
else:
|
raise Exception('Unknown block type: %r' % block)
|
|
self._instructions = instructions + self._instructions
|
self._offsets[id(block)] = -len(self._instructions)
|
return
|
|
|
class ArgFilterForwardingVisitor:
|
"""A visitor that forwards visitation to all arg filters."""
|
|
def __init__(self, visitor):
|
self._visited = set()
|
self.visitor = visitor
|
|
def visited(self, block):
|
if id(block) in self._visited:
|
return True
|
self._visited.add(id(block))
|
return False
|
|
def visit(self, block):
|
# All arg filters are BasicBlocks.
|
if not isinstance(block, BasicBlock):
|
return
|
# But the ALLOW, KILL_PROCESS, TRAP, etc. actions are too and we don't
|
# want to visit them just yet.
|
if (isinstance(block, KillProcess) or isinstance(block, KillThread)
|
or isinstance(block, Trap) or isinstance(block, ReturnErrno)
|
or isinstance(block, Trace) or isinstance(block, Log)
|
or isinstance(block, Allow)):
|
return
|
block.accept(self.visitor)
|
