def __init__(self, line='<line not provided>', reader=None):

if reader:

BaseException.__init__(self,

'Error at file offset %d, parsing line: %s' %

(reader.tell(), line))

else:

"""

def __init__(self, proto='', desc='', ret=''):

self.proto = proto

self.desc = desc

self.ret = ret

def to_dict(self):

return {

'proto': self.proto,

'desc': self.desc,

'ret': self.ret

"""

def __init__(self, proto='', desc='', ret='', attrs=[]):

super().__init__(proto, desc, ret)

self.attrs = attrs

self.enum_val = None

def proto_break_down(self):

"""

arg_re = re.compile(r'((\w+ )*?(\w+|...))( (\**)(\w+))?$')

res = {}

proto_re = re.compile(r'(.+) (\**)(\w+)\(((([^,]+)(, )?){1,5})\)$')

capture = proto_re.match(self.proto)

res['ret_type'] = capture.group(1)

res['ret_star'] = capture.group(2)

res['name'] = capture.group(3)

res['args'] = []

args = capture.group(4).split(', ')

for a in args:

capture = arg_re.match(a)

res['args'].append({

'type' : capture.group(1),

'star' : capture.group(5),

'name' : capture.group(6)

})

return res

def to_dict(self):

d = super().to_dict()

d["attrs"] = self.attrs

d.update(self.proto_break_down())

"""

def __init__(self, filename):

self.reader = open(filename, 'r')

self.line = ''

self.helpers = []

self.commands = []

self.desc_unique_helpers = set()

self.define_unique_helpers = []

self.helper_enum_vals = {}

self.helper_enum_pos = {}

self.desc_syscalls = []

self.enum_syscalls = []

def parse_element(self):

proto = self.parse_symbol()

desc = self.parse_desc(proto)

ret = self.parse_ret(proto)

return APIElement(proto=proto, desc=desc, ret=ret)

def parse_helper(self):

proto = self.parse_proto()

desc = self.parse_desc(proto)

ret = self.parse_ret(proto)

attrs = self.parse_attrs(proto)

return Helper(proto=proto, desc=desc, ret=ret, attrs=attrs)

def parse_symbol(self):

p = re.compile(r' \* ?(BPF\w+)$')

capture = p.match(self.line)

if not capture:

raise NoSyscallCommandFound

end_re = re.compile(r' \* ?NOTES$')

end = end_re.match(self.line)

if end:

raise NoSyscallCommandFound

self.line = self.reader.readline()

return capture.group(1)

def parse_proto(self):

# Argument can be of shape:

# - "void"

# - "type name"

# - "type *name"

# - Same as above, with "const" and/or "struct" in front of type

# - "..." (undefined number of arguments, for bpf_trace_printk())

# There is at least one term ("void"), and at most five arguments.

p = re.compile(r' \* ?((.+) \**\w+\((((const )?(struct )?(\w+|\.\.\.)( \**\w+)?)(, )?){1,5}\))$')

capture = p.match(self.line)

if not capture:

raise NoHelperFound

self.line = self.reader.readline()

return capture.group(1)

def parse_desc(self, proto):

p = re.compile(r' \* ?(?:\t| {5,8})Description$')

capture = p.match(self.line)

if not capture:

raise Exception("No description section found for " + proto)

# Description can be several lines, some of them possibly empty, and it

# stops when another subsection title is met.

desc = ''

desc_present = False

while True:

self.line = self.reader.readline()

if self.line == ' *\n':

desc += '\n'

else:

p = re.compile(r' \* ?(?:\t| {5,8})(?:\t| {8})(.*)')

capture = p.match(self.line)

if capture:

desc_present = True

desc += capture.group(1) + '\n'

else:

break

if not desc_present:

raise Exception("No description found for " + proto)

return desc

def parse_ret(self, proto):

p = re.compile(r' \* ?(?:\t| {5,8})Return$')

capture = p.match(self.line)

if not capture:

raise Exception("No return section found for " + proto)

# Return value description can be several lines, some of them possibly

# empty, and it stops when another subsection title is met.

ret = ''

ret_present = False

while True:

self.line = self.reader.readline()

if self.line == ' *\n':

ret += '\n'

else:

p = re.compile(r' \* ?(?:\t| {5,8})(?:\t| {8})(.*)')

capture = p.match(self.line)

if capture:

ret_present = True

ret += capture.group(1) + '\n'

else:

break

if not ret_present:

raise Exception("No return found for " + proto)

return ret

def parse_attrs(self, proto):

p = re.compile(r' \* ?(?:\t| {5,8})Attributes$')

capture = p.match(self.line)

if not capture:

return []

# Expect a single line with mnemonics for attributes separated by spaces

self.line = self.reader.readline()

p = re.compile(r' \* ?(?:\t| {5,8})(?:\t| {8})(.*)')

capture = p.match(self.line)

if not capture:

raise Exception("Incomplete 'Attributes' section for " + proto)

attrs = capture.group(1).split(' ')

for attr in attrs:

if attr not in ATTRS:

raise Exception("Unexpected attribute '" + attr + "' specified for " + proto)

self.line = self.reader.readline()

if self.line != ' *\n':

raise Exception("Expecting empty line after 'Attributes' section for " + proto)

# Prepare a line for next self.parse_* to consume

self.line = self.reader.readline()

return attrs

def seek_to(self, target, help_message, discard_lines = 1):

self.reader.seek(0)

offset = self.reader.read().find(target)

if offset == -1:

raise Exception(help_message)

self.reader.seek(offset)

self.reader.readline()

for _ in range(discard_lines):

self.reader.readline()

self.line = self.reader.readline()

def parse_desc_syscall(self):

self.seek_to('* DOC: eBPF Syscall Commands',

'Could not find start of eBPF syscall descriptions list')

while True:

try:

command = self.parse_element()

self.commands.append(command)

self.desc_syscalls.append(command.proto)

except NoSyscallCommandFound:

break

def parse_enum_syscall(self):

self.seek_to('enum bpf_cmd {',

'Could not find start of bpf_cmd enum', 0)

# Searches for either one or more BPF\w+ enums

bpf_p = re.compile(r'\s*(BPF\w+)+')

# Searches for an enum entry assigned to another entry,

# for e.g. BPF_PROG_RUN = BPF_PROG_TEST_RUN, which is

# not documented hence should be skipped in check to

# determine if the right number of syscalls are documented

assign_p = re.compile(r'\s*(BPF\w+)\s*=\s*(BPF\w+)')

bpf_cmd_str = ''

while True:

capture = assign_p.match(self.line)

if capture:

# Skip line if an enum entry is assigned to another entry

self.line = self.reader.readline()

continue

capture = bpf_p.match(self.line)

if capture:

bpf_cmd_str += self.line

else:

break

self.line = self.reader.readline()

# Find the number of occurences of BPF\w+

self.enum_syscalls = re.findall(r'(BPF\w+)+', bpf_cmd_str)

def parse_desc_helpers(self):

self.seek_to(helpersDocStart,

'Could not find start of eBPF helper descriptions list')

while True:

try:

helper = self.parse_helper()

self.helpers.append(helper)

proto = helper.proto_break_down()

self.desc_unique_helpers.add(proto['name'])

except NoHelperFound:

break

def parse_define_helpers(self):

# Parse FN(...) in #define ___BPF_FUNC_MAPPER to compare later with the

# number of unique function names present in description and use the

# correct enumeration value.

# Note: seek_to(..) discards the first line below the target search text,

# resulting in FN(unspec, 0, ##ctx) being skipped and not added to

# self.define_unique_helpers.

self.seek_to('#define ___BPF_FUNC_MAPPER(FN, ctx...)',

'Could not find start of eBPF helper definition list')

# Searches for one FN(\w+) define or a backslash for newline

p = re.compile(r'\s*FN\((\w+), (\d+), ##ctx\)|\\\\')

fn_defines_str = ''

i = 0

while True:

capture = p.match(self.line)

if capture:

fn_defines_str += self.line

helper_name = capture.expand(r'bpf_\1')

self.helper_enum_vals[helper_name] = int(capture.group(2))

self.helper_enum_pos[helper_name] = i

i += 1

else:

break

self.line = self.reader.readline()

# Find the number of occurences of FN(\w+)

self.define_unique_helpers = re.findall(r'FN\(\w+, \d+, ##ctx\)', fn_defines_str)

def validate_helpers(self):

last_helper = ''

seen_helpers = set()

seen_enum_vals = set()

i = 0

for helper in self.helpers:

proto = helper.proto_break_down()

name = proto['name']

try:

enum_val = self.helper_enum_vals[name]

enum_pos = self.helper_enum_pos[name]

except KeyError:

raise Exception("Helper %s is missing from enum bpf_func_id" % name)

if name in seen_helpers:

if last_helper != name:

raise Exception("Helper %s has multiple descriptions which are not grouped together" % name)

continue

# Enforce current practice of having the descriptions ordered

# by enum value.

if enum_pos != i:

raise Exception("Helper %s (ID %d) comment order (#%d) must be aligned with its position (#%d) in enum bpf_func_id" % (name, enum_val, i + 1, enum_pos + 1))

if enum_val in seen_enum_vals:

raise Exception("Helper %s has duplicated value %d" % (name, enum_val))

seen_helpers.add(name)

last_helper = name

seen_enum_vals.add(enum_val)

helper.enum_val = enum_val

i += 1

def run(self):

self.parse_desc_syscall()

self.parse_enum_syscall()

self.parse_desc_helpers()

self.parse_define_helpers()

self.validate_helpers()

"""

def __init__(self, parser):

self.parser = parser

self.elements = []

def print_header(self):

pass

def print_footer(self):

pass

def print_one(self, helper):

pass

def print_all(self):

self.print_header()

for elem in self.elements:

self.print_one(elem)

self.print_footer()

def elem_number_check(self, desc_unique_elem, define_unique_elem, type, instance):

"""

nr_desc_unique_elem = len(desc_unique_elem)

nr_define_unique_elem = len(define_unique_elem)

if nr_desc_unique_elem != nr_define_unique_elem:

exception_msg = '''

''' % (type, nr_desc_unique_elem, type, instance, nr_define_unique_elem)

if nr_desc_unique_elem < nr_define_unique_elem:

# Function description is parsed until no helper is found (which can be due to

# misformatting). Hence, only print the first missing/misformatted helper/enum.

exception_msg += '''

''' % (define_unique_elem[nr_desc_unique_elem])

"""

def __init__(self, parser):

self.parser = parser

def print_license(self):

license = '''\

print(license)

def print_elem(self, elem):

if (elem.desc):

print('\tDescription')

# Do not strip all newline characters: formatted code at the end of

# a section must be followed by a blank line.

for line in re.sub('\n$', '', elem.desc, count=1).split('\n'):

print('{}{}'.format('\t\t' if line else '', line))

if (elem.ret):

print('\tReturn')

for line in elem.ret.rstrip().split('\n'):

print('{}{}'.format('\t\t' if line else '', line))

print('')

def get_kernel_version(self):

try:

version = subprocess.run(['git', 'describe'], cwd=linuxRoot,

capture_output=True, check=True)

version = version.stdout.decode().rstrip()

except:

try:

version = subprocess.run(['make', '-s', '--no-print-directory', 'kernelversion'],

cwd=linuxRoot, capture_output=True, check=True)

version = version.stdout.decode().rstrip()

except:

return 'Linux'

return 'Linux {version}'.format(version=version)

def get_last_doc_update(self, delimiter):

try:

cmd = ['git', 'log', '-1', '--pretty=format:%cs', '--no-patch',

'-L',

'/{}/,/\\*\\//:include/uapi/linux/bpf.h'.format(delimiter)]

date = subprocess.run(cmd, cwd=linuxRoot,

capture_output=True, check=True)

return date.stdout.decode().rstrip()

except:

"""

def __init__(self, parser):

self.elements = parser.helpers

self.elem_number_check(parser.desc_unique_helpers, parser.define_unique_helpers, 'helper', '___BPF_FUNC_MAPPER')

def print_header(self):

header = '''\

kernelVersion = self.get_kernel_version()

lastUpdate = self.get_last_doc_update(helpersDocStart)

PrinterRST.print_license(self)

print(header.format(version=kernelVersion,

date_field = ':Date: ' if lastUpdate else '',

date=lastUpdate))

def print_footer(self):

footer = '''

print(footer)

def print_proto(self, helper):

"""

proto = helper.proto_break_down()

print('**%s %s%s(' % (proto['ret_type'],

proto['ret_star'].replace('*', '\\*'),

proto['name']),

end='')

comma = ''

for a in proto['args']:

one_arg = '{}{}'.format(comma, a['type'])

if a['name']:

if a['star']:

one_arg += ' {}**\\ '.format(a['star'].replace('*', '\\*'))

else:

one_arg += '** '

one_arg += '*{}*\\ **'.format(a['name'])

comma = ', '

print(one_arg, end='')

print(')**')

def print_one(self, helper):

self.print_proto(helper)

"""

def __init__(self, parser):

self.elements = parser.commands

self.elem_number_check(parser.desc_syscalls, parser.enum_syscalls, 'syscall', 'bpf_cmd')

def print_header(self):

header = '''\

PrinterRST.print_license(self)

print(header)

def print_one(self, command):

print('**%s**' % (command.proto))

"""

def __init__(self, parser):

self.elements = parser.helpers

self.elem_number_check(parser.desc_unique_helpers, parser.define_unique_helpers, 'helper', '___BPF_FUNC_MAPPER')

type_fwds = [

'struct bpf_fib_lookup',

'struct bpf_sk_lookup',

'struct bpf_perf_event_data',

'struct bpf_perf_event_value',

'struct bpf_pidns_info',

'struct bpf_redir_neigh',

'struct bpf_sock',

'struct bpf_sock_addr',

'struct bpf_sock_ops',

'struct bpf_sock_tuple',

'struct bpf_spin_lock',

'struct bpf_sysctl',

'struct bpf_tcp_sock',

'struct bpf_tunnel_key',

'struct bpf_xfrm_state',

'struct linux_binprm',

'struct pt_regs',

'struct sk_reuseport_md',

'struct sockaddr',

'struct tcphdr',

'struct seq_file',

'struct tcp6_sock',

'struct tcp_sock',

'struct tcp_timewait_sock',

'struct tcp_request_sock',

'struct udp6_sock',

'struct unix_sock',

'struct task_struct',

'struct cgroup',

'struct __sk_buff',

'struct sk_msg_md',

'struct xdp_md',

'struct path',

'struct btf_ptr',

'struct inode',

'struct socket',

'struct file',

'struct bpf_timer',

'struct mptcp_sock',

'struct bpf_dynptr',

'struct iphdr',

'struct ipv6hdr',

]

known_types = {

'...',

'void',

'const void',

'char',

'const char',

'int',

'long',

'unsigned long',

'__be16',

'__be32',

'__wsum',

'struct bpf_fib_lookup',

'struct bpf_perf_event_data',

'struct bpf_perf_event_value',

'struct bpf_pidns_info',

'struct bpf_redir_neigh',

'struct bpf_sk_lookup',

'struct bpf_sock',

'struct bpf_sock_addr',

'struct bpf_sock_ops',

'struct bpf_sock_tuple',

'struct bpf_spin_lock',

'struct bpf_sysctl',

'struct bpf_tcp_sock',

'struct bpf_tunnel_key',

'struct bpf_xfrm_state',

'struct linux_binprm',

'struct pt_regs',

'struct sk_reuseport_md',

'struct sockaddr',

'struct tcphdr',

'struct seq_file',

'struct tcp6_sock',

'struct tcp_sock',

'struct tcp_timewait_sock',

'struct tcp_request_sock',

'struct udp6_sock',

'struct unix_sock',

'struct task_struct',

'struct cgroup',

'struct path',

'struct btf_ptr',

'struct inode',

'struct socket',

'struct file',

'struct bpf_timer',

'struct mptcp_sock',

'struct bpf_dynptr',

'const struct bpf_dynptr',

'struct iphdr',

'struct ipv6hdr',

}

mapped_types = {

'u8': '__u8',

'u16': '__u16',

'u32': '__u32',

'u64': '__u64',

's8': '__s8',

's16': '__s16',

's32': '__s32',

's64': '__s64',

'size_t': 'unsigned long',

'struct bpf_map': 'void',

'struct sk_buff': 'struct __sk_buff',

'const struct sk_buff': 'const struct __sk_buff',

'struct sk_msg_buff': 'struct sk_msg_md',

'struct xdp_buff': 'struct xdp_md',

}

# Helpers overloaded for different context types.

overloaded_helpers = [

'bpf_get_socket_cookie',

'bpf_sk_assign',

]

def print_header(self):

header = '''\

print(header)

for fwd in self.type_fwds:

print('%s;' % fwd)

print('')

used_attrs = set()

for helper in self.elements:

for attr in helper.attrs:

used_attrs.add(attr)

for attr in sorted(used_attrs):

print('#ifndef %s' % attr)

print('#if __has_attribute(%s)' % ATTRS[attr])

print('#define %s __attribute__((%s))' % (attr, ATTRS[attr]))

print('#else')

print('#define %s' % attr)

print('#endif')

print('#endif')

if used_attrs:

print('')

def print_footer(self):

footer = ''

print(footer)

def map_type(self, t):

if t in self.known_types:

return t

if t in self.mapped_types:

return self.mapped_types[t]

print("Unrecognized type '%s', please add it to known types!" % t,

file=sys.stderr)

sys.exit(1)

seen_helpers = set()

def print_one(self, helper):

proto = helper.proto_break_down()

if proto['name'] in self.seen_helpers:

return

self.seen_helpers.add(proto['name'])

print('/*')

print(" * %s" % proto['name'])

print(" *")

if (helper.desc):

# Do not strip all newline characters: formatted code at the end of

# a section must be followed by a blank line.

for line in re.sub('\n$', '', helper.desc, count=1).split('\n'):

print(' *{}{}'.format(' \t' if line else '', line))

if (helper.ret):

print(' *')

print(' * Returns')

for line in helper.ret.rstrip().split('\n'):

print(' *{}{}'.format(' \t' if line else '', line))

print(' */')

print('static ', end='')

if helper.attrs:

print('%s ' % (" ".join(helper.attrs)), end='')

print('%s %s(* const %s)(' % (self.map_type(proto['ret_type']),

proto['ret_star'], proto['name']), end='')

comma = ''

for i, a in enumerate(proto['args']):

t = a['type']

n = a['name']

if proto['name'] in self.overloaded_helpers and i == 0:

t = 'void'

n = 'ctx'

one_arg = '{}{}'.format(comma, self.map_type(t))

if n:

if a['star']:

one_arg += ' {}'.format(a['star'])

else:

one_arg += ' '

one_arg += '{}'.format(n)

comma = ', '

print(one_arg, end='')

print(') = (void *) %d;' % helper.enum_val)

"""

def __init__(self, parser):

self.elements = parser.helpers

self.elem_number_check(

parser.desc_unique_helpers,

parser.define_unique_helpers,

"helper",

"___BPF_FUNC_MAPPER",

)

def print_all(self):

helper_dicts = [helper.to_dict() for helper in self.elements]

out_dict = {'helpers': helper_dicts}

"""

def __init__(self, parser):

self.elements = parser.commands

self.elem_number_check(parser.desc_syscalls, parser.enum_syscalls, 'syscall', 'bpf_cmd')

def print_all(self):

syscall_dicts = [syscall.to_dict() for syscall in self.elements]

out_dict = {'syscall': syscall_dicts}

'helpers': {

'rst': PrinterHelpersRST,

'json': PrinterHelpersJSON,

'header': PrinterHelpersHeader,

},

'syscall': {

'rst': PrinterSyscallRST,

'json': PrinterSyscallJSON

},

args = argParser.parse_args()

# Parse file.

headerParser = HeaderParser(args.filename)

headerParser.run()