Blame iptables-test.py

Packit Service d1fe03
#!/usr/bin/env python
Packit Service d1fe03
#
Packit Service d1fe03
# (C) 2012-2013 by Pablo Neira Ayuso <pablo@netfilter.org>
Packit Service d1fe03
#
Packit Service d1fe03
# This program is free software; you can redistribute it and/or modify
Packit Service d1fe03
# it under the terms of the GNU General Public License as published by
Packit Service d1fe03
# the Free Software Foundation; either version 2 of the License, or
Packit Service d1fe03
# (at your option) any later version.
Packit Service d1fe03
#
Packit Service d1fe03
# This software has been sponsored by Sophos Astaro <http://www.sophos.com>
Packit Service d1fe03
#
Packit Service d1fe03
Packit Service d1fe03
from __future__ import print_function
Packit Service d1fe03
import sys
Packit Service d1fe03
import os
Packit Service d1fe03
import subprocess
Packit Service d1fe03
import argparse
Packit Service d1fe03
Packit Service d1fe03
IPTABLES = "iptables"
Packit Service d1fe03
IP6TABLES = "ip6tables"
Packit Service d1fe03
ARPTABLES = "arptables"
Packit Service d1fe03
EBTABLES = "ebtables"
Packit Service d1fe03
Packit Service d1fe03
IPTABLES_SAVE = "iptables-save"
Packit Service d1fe03
IP6TABLES_SAVE = "ip6tables-save"
Packit Service d1fe03
ARPTABLES_SAVE = "arptables-save"
Packit Service d1fe03
EBTABLES_SAVE = "ebtables-save"
Packit Service d1fe03
#IPTABLES_SAVE = ['xtables-save','-4']
Packit Service d1fe03
#IP6TABLES_SAVE = ['xtables-save','-6']
Packit Service d1fe03
Packit Service d1fe03
EXTENSIONS_PATH = "extensions"
Packit Service d1fe03
LOGFILE="/tmp/iptables-test.log"
Packit Service d1fe03
log_file = None
Packit Service d1fe03
Packit Service d1fe03
Packit Service d1fe03
class Colors:
Packit Service d1fe03
    HEADER = '\033[95m'
Packit Service d1fe03
    BLUE = '\033[94m'
Packit Service d1fe03
    GREEN = '\033[92m'
Packit Service d1fe03
    YELLOW = '\033[93m'
Packit Service d1fe03
    RED = '\033[91m'
Packit Service d1fe03
    ENDC = '\033[0m'
Packit Service d1fe03
Packit Service d1fe03
Packit Service d1fe03
def print_error(reason, filename=None, lineno=None):
Packit Service d1fe03
    '''
Packit Service d1fe03
    Prints an error with nice colors, indicating file and line number.
Packit Service d1fe03
    '''
Packit Service d1fe03
    print(filename + ": " + Colors.RED + "ERROR" +
Packit Service d1fe03
        Colors.ENDC + ": line %d (%s)" % (lineno, reason))
Packit Service d1fe03
Packit Service d1fe03
Packit Service d1fe03
def delete_rule(iptables, rule, filename, lineno):
Packit Service d1fe03
    '''
Packit Service d1fe03
    Removes an iptables rule
Packit Service d1fe03
    '''
Packit Service d1fe03
    cmd = iptables + " -D " + rule
Packit Service d1fe03
    ret = execute_cmd(cmd, filename, lineno)
Packit Service d1fe03
    if ret == 1:
Packit Service d1fe03
        reason = "cannot delete: " + iptables + " -I " + rule
Packit Service d1fe03
        print_error(reason, filename, lineno)
Packit Service d1fe03
        return -1
Packit Service d1fe03
Packit Service d1fe03
    return 0
Packit Service d1fe03
Packit Service d1fe03
Packit Service d1fe03
def run_test(iptables, rule, rule_save, res, filename, lineno, netns):
Packit Service d1fe03
    '''
Packit Service d1fe03
    Executes an unit test. Returns the output of delete_rule().
Packit Service d1fe03
Packit Service d1fe03
    Parameters:
Packit Service d1fe03
    :param  iptables: string with the iptables command to execute
Packit Service d1fe03
    :param rule: string with iptables arguments for the rule to test
Packit Service d1fe03
    :param rule_save: string to find the rule in the output of iptables -save
Packit Service d1fe03
    :param res: expected result of the rule. Valid values: "OK", "FAIL"
Packit Service d1fe03
    :param filename: name of the file tested (used for print_error purposes)
Packit Service d1fe03
    :param lineno: line number being tested (used for print_error purposes)
Packit Service d1fe03
    '''
Packit Service d1fe03
    ret = 0
Packit Service d1fe03
Packit Service d1fe03
    cmd = iptables + " -A " + rule
Packit Service d1fe03
    if netns:
Packit Service d1fe03
            cmd = "ip netns exec ____iptables-container-test " + EXECUTEABLE + " " + cmd
Packit Service d1fe03
Packit Service d1fe03
    ret = execute_cmd(cmd, filename, lineno)
Packit Service d1fe03
Packit Service d1fe03
    #
Packit Service d1fe03
    # report failed test
Packit Service d1fe03
    #
Packit Service d1fe03
    if ret:
Packit Service d1fe03
        if res == "OK":
Packit Service d1fe03
            reason = "cannot load: " + cmd
Packit Service d1fe03
            print_error(reason, filename, lineno)
Packit Service d1fe03
            return -1
Packit Service d1fe03
        else:
Packit Service d1fe03
            # do not report this error
Packit Service d1fe03
            return 0
Packit Service d1fe03
    else:
Packit Service d1fe03
        if res == "FAIL":
Packit Service d1fe03
            reason = "should fail: " + cmd
Packit Service d1fe03
            print_error(reason, filename, lineno)
Packit Service d1fe03
            delete_rule(iptables, rule, filename, lineno)
Packit Service d1fe03
            return -1
Packit Service d1fe03
Packit Service d1fe03
    matching = 0
Packit Service d1fe03
    splitted = iptables.split(" ")
Packit Service d1fe03
    if len(splitted) == 2:
Packit Service d1fe03
        if splitted[1] == '-4':
Packit Service d1fe03
            command = IPTABLES_SAVE
Packit Service d1fe03
        elif splitted[1] == '-6':
Packit Service d1fe03
            command = IP6TABLES_SAVE
Packit Service d1fe03
    elif len(splitted) == 1:
Packit Service d1fe03
        if splitted[0] == IPTABLES:
Packit Service d1fe03
            command = IPTABLES_SAVE
Packit Service d1fe03
        elif splitted[0] == IP6TABLES:
Packit Service d1fe03
            command = IP6TABLES_SAVE
Packit Service d1fe03
        elif splitted[0] == ARPTABLES:
Packit Service d1fe03
            command = ARPTABLES_SAVE
Packit Service d1fe03
        elif splitted[0] == EBTABLES:
Packit Service d1fe03
            command = EBTABLES_SAVE
Packit Service d1fe03
Packit Service ca2a4a
    command = EXECUTEABLE + " " + command
Packit Service d1fe03
Packit Service d1fe03
    if netns:
Packit Service d1fe03
            command = "ip netns exec ____iptables-container-test " + command
Packit Service d1fe03
Packit Service d1fe03
    args = splitted[1:]
Packit Service d1fe03
    proc = subprocess.Popen(command, shell=True,
Packit Service d1fe03
                            stdin=subprocess.PIPE,
Packit Service d1fe03
                            stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Packit Service d1fe03
    out, err = proc.communicate()
Packit Service d1fe03
Packit Service d1fe03
    #
Packit Service d1fe03
    # check for segfaults
Packit Service d1fe03
    #
Packit Service d1fe03
    if proc.returncode == -11:
Packit Service d1fe03
        reason = "iptables-save segfaults: " + cmd
Packit Service d1fe03
        print_error(reason, filename, lineno)
Packit Service d1fe03
        delete_rule(iptables, rule, filename, lineno)
Packit Service d1fe03
        return -1
Packit Service d1fe03
Packit Service d1fe03
    # find the rule
Packit Service d1fe03
    matching = out.find(rule_save.encode('utf-8'))
Packit Service d1fe03
    if matching < 0:
Packit Service d1fe03
        reason = "cannot find: " + iptables + " -I " + rule
Packit Service d1fe03
        print_error(reason, filename, lineno)
Packit Service d1fe03
        delete_rule(iptables, rule, filename, lineno)
Packit Service d1fe03
        return -1
Packit Service d1fe03
Packit Service d1fe03
    # Test "ip netns del NETNS" path with rules in place
Packit Service d1fe03
    if netns:
Packit Service d1fe03
        return 0
Packit Service d1fe03
Packit Service d1fe03
    return delete_rule(iptables, rule, filename, lineno)
Packit Service d1fe03
Packit Service d1fe03
def execute_cmd(cmd, filename, lineno):
Packit Service d1fe03
    '''
Packit Service d1fe03
    Executes a command, checking for segfaults and returning the command exit
Packit Service d1fe03
    code.
Packit Service d1fe03
Packit Service d1fe03
    :param cmd: string with the command to be executed
Packit Service d1fe03
    :param filename: name of the file tested (used for print_error purposes)
Packit Service d1fe03
    :param lineno: line number being tested (used for print_error purposes)
Packit Service d1fe03
    '''
Packit Service d1fe03
    global log_file
Packit Service d1fe03
    if cmd.startswith('iptables ') or cmd.startswith('ip6tables ') or cmd.startswith('ebtables ') or cmd.startswith('arptables '):
Packit Service ca2a4a
        cmd = EXECUTEABLE + " " + cmd
Packit Service d1fe03
Packit Service d1fe03
    print("command: {}".format(cmd), file=log_file)
Packit Service d1fe03
    ret = subprocess.call(cmd, shell=True, universal_newlines=True,
Packit Service d1fe03
        stderr=subprocess.STDOUT, stdout=log_file)
Packit Service d1fe03
    log_file.flush()
Packit Service d1fe03
Packit Service d1fe03
    # generic check for segfaults
Packit Service d1fe03
    if ret  == -11:
Packit Service d1fe03
        reason = "command segfaults: " + cmd
Packit Service d1fe03
        print_error(reason, filename, lineno)
Packit Service d1fe03
    return ret
Packit Service d1fe03
Packit Service d1fe03
Packit Service d1fe03
def run_test_file(filename, netns):
Packit Service d1fe03
    '''
Packit Service d1fe03
    Runs a test file
Packit Service d1fe03
Packit Service d1fe03
    :param filename: name of the file with the test rules
Packit Service d1fe03
    '''
Packit Service d1fe03
    #
Packit Service d1fe03
    # if this is not a test file, skip.
Packit Service d1fe03
    #
Packit Service d1fe03
    if not filename.endswith(".t"):
Packit Service d1fe03
        return 0, 0
Packit Service d1fe03
Packit Service d1fe03
    if "libipt_" in filename:
Packit Service d1fe03
        iptables = IPTABLES
Packit Service d1fe03
    elif "libip6t_" in filename:
Packit Service d1fe03
        iptables = IP6TABLES
Packit Service d1fe03
    elif "libxt_"  in filename:
Packit Service d1fe03
        iptables = IPTABLES
Packit Service d1fe03
    elif "libarpt_" in filename:
Packit Service d1fe03
        # only supported with nf_tables backend
Packit Service d1fe03
        if EXECUTEABLE != "xtables-nft-multi":
Packit Service d1fe03
           return 0, 0
Packit Service d1fe03
        iptables = ARPTABLES
Packit Service d1fe03
    elif "libebt_" in filename:
Packit Service d1fe03
        # only supported with nf_tables backend
Packit Service d1fe03
        if EXECUTEABLE != "xtables-nft-multi":
Packit Service d1fe03
           return 0, 0
Packit Service d1fe03
        iptables = EBTABLES
Packit Service d1fe03
    else:
Packit Service d1fe03
        # default to iptables if not known prefix
Packit Service d1fe03
        iptables = IPTABLES
Packit Service d1fe03
Packit Service d1fe03
    f = open(filename)
Packit Service d1fe03
Packit Service d1fe03
    tests = 0
Packit Service d1fe03
    passed = 0
Packit Service d1fe03
    table = ""
Packit Service d1fe03
    total_test_passed = True
Packit Service d1fe03
Packit Service d1fe03
    if netns:
Packit Service d1fe03
        execute_cmd("ip netns add ____iptables-container-test", filename, 0)
Packit Service d1fe03
Packit Service d1fe03
    for lineno, line in enumerate(f):
Packit Service d1fe03
        if line[0] == "#":
Packit Service d1fe03
            continue
Packit Service d1fe03
Packit Service d1fe03
        if line[0] == ":":
Packit Service d1fe03
            chain_array = line.rstrip()[1:].split(",")
Packit Service d1fe03
            continue
Packit Service d1fe03
Packit Service d1fe03
        # external non-iptables invocation, executed as is.
Packit Service d1fe03
        if line[0] == "@":
Packit Service d1fe03
            external_cmd = line.rstrip()[1:]
Packit Service d1fe03
            if netns:
Packit Service d1fe03
                external_cmd = "ip netns exec ____iptables-container-test " + external_cmd
Packit Service d1fe03
            execute_cmd(external_cmd, filename, lineno)
Packit Service d1fe03
            continue
Packit Service d1fe03
Packit Service d1fe03
        # external iptables invocation, executed as is.
Packit Service d1fe03
        if line[0] == "%":
Packit Service d1fe03
            external_cmd = line.rstrip()[1:]
Packit Service d1fe03
            if netns:
Packit Service d1fe03
                external_cmd = "ip netns exec ____iptables-container-test " + EXECUTEABLE + " " + external_cmd
Packit Service d1fe03
            execute_cmd(external_cmd, filename, lineno)
Packit Service d1fe03
            continue
Packit Service d1fe03
Packit Service d1fe03
        if line[0] == "*":
Packit Service d1fe03
            table = line.rstrip()[1:]
Packit Service d1fe03
            continue
Packit Service d1fe03
Packit Service d1fe03
        if len(chain_array) == 0:
Packit Service d1fe03
            print("broken test, missing chain, leaving")
Packit Service d1fe03
            sys.exit()
Packit Service d1fe03
Packit Service d1fe03
        test_passed = True
Packit Service d1fe03
        tests += 1
Packit Service d1fe03
Packit Service d1fe03
        for chain in chain_array:
Packit Service d1fe03
            item = line.split(";")
Packit Service d1fe03
            if table == "":
Packit Service d1fe03
                rule = chain + " " + item[0]
Packit Service d1fe03
            else:
Packit Service d1fe03
                rule = chain + " -t " + table + " " + item[0]
Packit Service d1fe03
Packit Service d1fe03
            if item[1] == "=":
Packit Service d1fe03
                rule_save = chain + " " + item[0]
Packit Service d1fe03
            else:
Packit Service d1fe03
                rule_save = chain + " " + item[1]
Packit Service d1fe03
Packit Service d1fe03
            res = item[2].rstrip()
Packit Service d1fe03
            ret = run_test(iptables, rule, rule_save,
Packit Service d1fe03
                           res, filename, lineno + 1, netns)
Packit Service d1fe03
Packit Service d1fe03
            if ret < 0:
Packit Service d1fe03
                test_passed = False
Packit Service d1fe03
                total_test_passed = False
Packit Service d1fe03
                break
Packit Service d1fe03
Packit Service d1fe03
        if test_passed:
Packit Service d1fe03
            passed += 1
Packit Service d1fe03
Packit Service d1fe03
    if netns:
Packit Service d1fe03
        execute_cmd("ip netns del ____iptables-container-test", filename, 0)
Packit Service d1fe03
    if total_test_passed:
Packit Service d1fe03
        print(filename + ": " + Colors.GREEN + "OK" + Colors.ENDC)
Packit Service d1fe03
Packit Service d1fe03
    f.close()
Packit Service d1fe03
    return tests, passed
Packit Service d1fe03
Packit Service d1fe03
Packit Service d1fe03
def show_missing():
Packit Service d1fe03
    '''
Packit Service d1fe03
    Show the list of missing test files
Packit Service d1fe03
    '''
Packit Service d1fe03
    file_list = os.listdir(EXTENSIONS_PATH)
Packit Service d1fe03
    testfiles = [i for i in file_list if i.endswith('.t')]
Packit Service d1fe03
    libfiles = [i for i in file_list
Packit Service d1fe03
                if i.startswith('lib') and i.endswith('.c')]
Packit Service d1fe03
Packit Service d1fe03
    def test_name(x):
Packit Service d1fe03
        return x[0:-2] + '.t'
Packit Service d1fe03
    missing = [test_name(i) for i in libfiles
Packit Service d1fe03
               if not test_name(i) in testfiles]
Packit Service d1fe03
Packit Service d1fe03
    print('\n'.join(missing))
Packit Service d1fe03
Packit Service d1fe03
Packit Service d1fe03
#
Packit Service d1fe03
# main
Packit Service d1fe03
#
Packit Service d1fe03
def main():
Packit Service d1fe03
    parser = argparse.ArgumentParser(description='Run iptables tests')
Packit Service d1fe03
    parser.add_argument('filename', nargs='?',
Packit Service d1fe03
                        metavar='path/to/file.t',
Packit Service d1fe03
                        help='Run only this test')
Packit Service d1fe03
    parser.add_argument('-H', '--host', action='store_true',
Packit Service d1fe03
                        help='Run tests against installed binaries')
Packit Service d1fe03
    parser.add_argument('-l', '--legacy', action='store_true',
Packit Service d1fe03
                        help='Test iptables-legacy')
Packit Service d1fe03
    parser.add_argument('-m', '--missing', action='store_true',
Packit Service d1fe03
                        help='Check for missing tests')
Packit Service d1fe03
    parser.add_argument('-n', '--nftables', action='store_true',
Packit Service d1fe03
                        help='Test iptables-over-nftables')
Packit Service d1fe03
    parser.add_argument('-N', '--netns', action='store_true',
Packit Service d1fe03
                        help='Test netnamespace path')
Packit Service d1fe03
    args = parser.parse_args()
Packit Service d1fe03
Packit Service d1fe03
    #
Packit Service d1fe03
    # show list of missing test files
Packit Service d1fe03
    #
Packit Service d1fe03
    if args.missing:
Packit Service d1fe03
        show_missing()
Packit Service d1fe03
        return
Packit Service d1fe03
Packit Service d1fe03
    global EXECUTEABLE
Packit Service d1fe03
    EXECUTEABLE = "xtables-legacy-multi"
Packit Service d1fe03
    if args.nftables:
Packit Service d1fe03
        EXECUTEABLE = "xtables-nft-multi"
Packit Service d1fe03
Packit Service d1fe03
    if os.getuid() != 0:
Packit Service d1fe03
        print("You need to be root to run this, sorry")
Packit Service d1fe03
        return
Packit Service d1fe03
Packit Service d1fe03
    if not args.host:
Packit Service d1fe03
        os.putenv("XTABLES_LIBDIR", os.path.abspath(EXTENSIONS_PATH))
Packit Service d1fe03
        os.putenv("PATH", "%s/iptables:%s" % (os.path.abspath(os.path.curdir),
Packit Service d1fe03
                                              os.getenv("PATH")))
Packit Service d1fe03
Packit Service d1fe03
    test_files = 0
Packit Service d1fe03
    tests = 0
Packit Service d1fe03
    passed = 0
Packit Service d1fe03
Packit Service d1fe03
    # setup global var log file
Packit Service d1fe03
    global log_file
Packit Service d1fe03
    try:
Packit Service d1fe03
        log_file = open(LOGFILE, 'w')
Packit Service d1fe03
    except IOError:
Packit Service d1fe03
        print("Couldn't open log file %s" % LOGFILE)
Packit Service d1fe03
        return
Packit Service d1fe03
Packit Service d1fe03
    if args.filename:
Packit Service d1fe03
        file_list = [args.filename]
Packit Service d1fe03
    else:
Packit Service d1fe03
        file_list = [os.path.join(EXTENSIONS_PATH, i)
Packit Service d1fe03
                     for i in os.listdir(EXTENSIONS_PATH)
Packit Service d1fe03
                     if i.endswith('.t')]
Packit Service d1fe03
        file_list.sort()
Packit Service d1fe03
Packit Service d1fe03
    for filename in file_list:
Packit Service d1fe03
        file_tests, file_passed = run_test_file(filename, args.netns)
Packit Service d1fe03
        if file_tests:
Packit Service d1fe03
            tests += file_tests
Packit Service d1fe03
            passed += file_passed
Packit Service d1fe03
            test_files += 1
Packit Service d1fe03
Packit Service d1fe03
    print("%d test files, %d unit tests, %d passed" % (test_files, tests, passed))
Packit Service d1fe03
Packit Service d1fe03
Packit Service d1fe03
if __name__ == '__main__':
Packit Service d1fe03
    main()