| #!/usr/bin/env python3 |
| # encoding: utf-8 |
| |
| import os |
| import sys |
| import shlex |
| import argparse |
| from subprocess import Popen, PIPE |
| |
| def run_proc(args, shell = False, input = None): |
| """A simple wrapper around Popen, returning (rc, stdout, stderr)""" |
| process = Popen(args, text = True, shell = shell, |
| stdin = PIPE, stdout = PIPE, stderr = PIPE) |
| output, error = process.communicate(input) |
| return (process.returncode, output, error) |
| |
| keywords = ("iptables-translate", "ip6tables-translate", "ebtables-translate") |
| xtables_nft_multi = 'xtables-nft-multi' |
| |
| if sys.stdout.isatty(): |
| colors = {"magenta": "\033[95m", "green": "\033[92m", "yellow": "\033[93m", |
| "red": "\033[91m", "end": "\033[0m"} |
| else: |
| colors = {"magenta": "", "green": "", "yellow": "", "red": "", "end": ""} |
| |
| |
| def magenta(string): |
| return colors["magenta"] + string + colors["end"] |
| |
| |
| def red(string): |
| return colors["red"] + string + colors["end"] |
| |
| |
| def yellow(string): |
| return colors["yellow"] + string + colors["end"] |
| |
| |
| def green(string): |
| return colors["green"] + string + colors["end"] |
| |
| |
| def test_one_xlate(name, sourceline, expected, result): |
| rc, output, error = run_proc([xtables_nft_multi] + shlex.split(sourceline)) |
| if rc != 0: |
| result.append(name + ": " + red("Error: ") + "iptables-translate failure") |
| result.append(error) |
| return False |
| |
| translation = output.rstrip(" \n") |
| if translation != expected: |
| result.append(name + ": " + red("Fail")) |
| result.append(magenta("src: ") + sourceline.rstrip(" \n")) |
| result.append(magenta("exp: ") + expected) |
| result.append(magenta("res: ") + translation + "\n") |
| return False |
| |
| return True |
| |
| def test_one_replay(name, sourceline, expected, result): |
| global args |
| |
| searchline = None |
| if sourceline.find(';') >= 0: |
| sourceline, searchline = sourceline.split(';') |
| |
| srcwords = shlex.split(sourceline) |
| |
| srccmd = srcwords[0] |
| ipt = srccmd.split('-')[0] |
| table_idx = -1 |
| chain_idx = -1 |
| table_name = "filter" |
| chain_name = None |
| for idx in range(1, len(srcwords)): |
| if srcwords[idx] in ["-A", "-I", "--append", "--insert"]: |
| chain_idx = idx |
| chain_name = srcwords[idx + 1] |
| elif srcwords[idx] in ["-t", "--table"]: |
| table_idx = idx |
| table_name = srcwords[idx + 1] |
| |
| if not chain_name: |
| return True # nothing to do? |
| |
| if searchline is None: |
| # adjust sourceline as required |
| checkcmd = srcwords[:] |
| checkcmd[0] = ipt |
| checkcmd[chain_idx] = "--check" |
| else: |
| checkcmd = [ipt, "-t", table_name] |
| checkcmd += ["--check", chain_name, searchline] |
| |
| fam = "" |
| if srccmd.startswith("ip6"): |
| fam = "ip6 " |
| elif srccmd.startswith("ebt"): |
| fam = "bridge " |
| |
| expected = [ l.removeprefix("nft ").strip(" '") for l in expected.split("\n") ] |
| nft_input = [ |
| "flush ruleset", |
| "add table " + fam + table_name, |
| "add chain " + fam + table_name + " " + chain_name, |
| ] + expected |
| |
| rc, output, error = run_proc([args.nft, "-f", "-"], shell = False, input = "\n".join(nft_input)) |
| if rc != 0: |
| result.append(name + ": " + red("Replay Fail")) |
| result.append(args.nft + " call failed: " + error.rstrip('\n')) |
| for line in nft_input: |
| result.append(magenta("input: ") + line) |
| return False |
| |
| rc, output, error = run_proc([xtables_nft_multi] + checkcmd) |
| if rc != 0: |
| result.append(name + ": " + red("Check Fail")) |
| result.append(magenta("check: ") + " ".join(checkcmd)) |
| result.append(magenta("error: ") + error) |
| rc, output, error = run_proc([xtables_nft_multi, ipt + "-save"]) |
| for l in output.split("\n"): |
| result.append(magenta("ipt: ") + l) |
| rc, output, error = run_proc([args.nft, "list", "ruleset"]) |
| for l in output.split("\n"): |
| result.append(magenta("nft: ") + l) |
| return False |
| |
| return True |
| |
| |
| def run_test(name, payload): |
| global xtables_nft_multi |
| global args |
| |
| test_passed = True |
| tests = passed = failed = errors = 0 |
| result = [] |
| |
| line = payload.readline() |
| while line: |
| if not line.startswith(keywords): |
| line = payload.readline() |
| continue |
| |
| sourceline = replayline = line.rstrip("\n") |
| if line.find(';') >= 0: |
| sourceline = line.split(';')[0] |
| |
| expected = payload.readline().rstrip(" \n") |
| next_expected = payload.readline() |
| if next_expected.startswith("nft"): |
| expected += "\n" + next_expected.rstrip(" \n") |
| line = payload.readline() |
| else: |
| line = next_expected |
| |
| tests += 1 |
| if test_one_xlate(name, sourceline, expected, result): |
| passed += 1 |
| else: |
| errors += 1 |
| test_passed = False |
| continue |
| |
| if args.replay: |
| tests += 1 |
| if test_one_replay(name, replayline, expected, result): |
| passed += 1 |
| else: |
| errors += 1 |
| test_passed = False |
| |
| rc, output, error = run_proc([args.nft, "flush", "ruleset"]) |
| if rc != 0: |
| result.append(name + ": " + red("Fail")) |
| result.append("nft flush ruleset call failed: " + error) |
| |
| if (passed == tests): |
| print(name + ": " + green("OK")) |
| if not test_passed: |
| print("\n".join(result), file=sys.stderr) |
| return tests, passed, failed, errors |
| |
| |
| def load_test_files(): |
| test_files = total_tests = total_passed = total_error = total_failed = 0 |
| tests = sorted(os.listdir("extensions")) |
| for test in ['extensions/' + f for f in tests if f.endswith(".txlate")]: |
| with open(test, "r") as payload: |
| tests, passed, failed, errors = run_test(test, payload) |
| test_files += 1 |
| total_tests += tests |
| total_passed += passed |
| total_failed += failed |
| total_error += errors |
| return (test_files, total_tests, total_passed, total_failed, total_error) |
| |
| |
| def spawn_netns(): |
| # prefer unshare module |
| try: |
| import unshare |
| unshare.unshare(unshare.CLONE_NEWNET) |
| return True |
| except: |
| pass |
| |
| # sledgehammer style: |
| # - call ourselves prefixed by 'unshare -n' if found |
| # - pass extra --no-netns parameter to avoid another recursion |
| try: |
| import shutil |
| |
| unshare = shutil.which("unshare") |
| if unshare is None: |
| return False |
| |
| sys.argv.append("--no-netns") |
| os.execv(unshare, [unshare, "-n", sys.executable] + sys.argv) |
| except: |
| pass |
| |
| return False |
| |
| |
| def main(): |
| global xtables_nft_multi |
| |
| if args.replay: |
| if os.getuid() != 0: |
| print("Replay test requires root, sorry", file=sys.stderr) |
| return |
| if not args.no_netns and not spawn_netns(): |
| print("Cannot run in own namespace, connectivity might break", |
| file=sys.stderr) |
| |
| if not args.host: |
| os.putenv("XTABLES_LIBDIR", os.path.abspath("extensions")) |
| xtables_nft_multi = os.path.abspath(os.path.curdir) \ |
| + '/iptables/' + xtables_nft_multi |
| |
| files = tests = passed = failed = errors = 0 |
| for test in args.test: |
| if not test.endswith(".txlate"): |
| test += ".txlate" |
| try: |
| with open(test, "r") as payload: |
| t, p, f, e = run_test(test, payload) |
| files += 1 |
| tests += t |
| passed += p |
| failed += f |
| errors += e |
| except IOError: |
| print(red("Error: ") + "test file does not exist", file=sys.stderr) |
| return 99 |
| |
| if files == 0: |
| files, tests, passed, failed, errors = load_test_files() |
| |
| if files > 1: |
| file_word = "files" |
| else: |
| file_word = "file" |
| print("%d test %s, %d tests, %d tests passed, %d tests failed, %d errors" |
| % (files, file_word, tests, passed, failed, errors)) |
| return passed - tests |
| |
| |
| parser = argparse.ArgumentParser() |
| parser.add_argument('-H', '--host', action='store_true', |
| help='Run tests against installed binaries') |
| parser.add_argument('-R', '--replay', action='store_true', |
| help='Replay tests to check iptables-nft parser') |
| parser.add_argument('-n', '--nft', type=str, default='nft', |
| help='Replay using given nft binary (default: \'%(default)s\')') |
| parser.add_argument('--no-netns', action='store_true', |
| help='Do not run testsuite in own network namespace') |
| parser.add_argument("test", nargs="*", help="run only the specified test file(s)") |
| args = parser.parse_args() |
| sys.exit(main()) |
