"""Test recent address cache."""
-import os
-import typing as T
-import shutil
-
from testlib import check
from testlib.log import log
from testlib.proc import Tinc, Script
from testlib.test import Test
+import os
+import typing as T
+import shutil
+
def init(ctx: Test) -> T.Tuple[Tinc, Tinc]:
"""Create test node."""
"""Check that legacy protocol works with different cryptographic algorithms."""
-import typing as T
-
from testlib.test import Test
from testlib.proc import Tinc
from testlib.log import log
from testlib import cmd, check
+import typing as T
+
def init(ctx: Test, digest: str, cipher: str) -> T.Tuple[Tinc, Tinc]:
"""Initialize new test nodes."""
"""Test binding to interfaces and addresses."""
+from testlib.const import EXIT_SKIP
+from testlib.log import log
+from testlib.test import Test
+from testlib import check, util
+
import json
import socket
import subprocess as subp
import sys
import typing as T
-from testlib import check, util
-from testlib.const import EXIT_SKIP
-from testlib.log import log
-from testlib.test import Test
-
util.require_command("ss", "-nlup")
util.require_command("ip", "--json", "addr")
"""Test binding to ports on localhost."""
-import socket
-import sys
-import typing as T
-
-from testlib import check, util
from testlib.const import EXIT_SKIP
from testlib.log import log
from testlib.proc import Script
from testlib.test import Test
+from testlib import check, util
+
+import socket
+import sys
+import typing as T
# Call to close opened port
Closer = T.Callable[[], None]
"""Test dump commands."""
-import subprocess as subp
-
from testlib import check
from testlib.log import log
from testlib.proc import Tinc
from testlib.test import Test
+import subprocess as subp
+
SUBNETS_FOO = ("10.0.0.0/16", "10.1.2.0/24")
SUBNETS_BAR = ("10.3.2.0/27", "fe80::/64")
SUBNETS_BROADCAST = len(
"""Test 'tinc fsck' command."""
-import os
-import sys
-import typing as T
-
from testlib import check
from testlib.const import RUN_ACCESS_CHECKS
from testlib.log import log
from testlib.proc import Tinc, Feature
from testlib.util import read_text, read_lines, write_lines, append_line, write_text
+import os
+import sys
+import typing as T
+
RUN_LEGACY_CHECKS = Feature.LEGACY_PROTOCOL in Tinc().features
RUN_EXECUTABILITY_CHECKS = os.name != "nt"
RUN_PERMISSION_CHECKS = RUN_EXECUTABILITY_CHECKS
"""Test import/export error conditions."""
-import os
-
from testlib import check, cmd, util
from testlib.log import log
from testlib.const import RUN_ACCESS_CHECKS
from testlib.proc import Tinc
from testlib.test import Test
+import os
+
SEPARATOR = f"#{'-' * 63}#"
MULTI_HOST = f"""
"""Test invite/join error conditions."""
-import os
-import shutil
-import socket
-
-from testlib import check, util
from testlib.log import log
from testlib.const import RUN_ACCESS_CHECKS
from testlib.proc import Tinc, Script
from testlib.test import Test
+from testlib import check, util
+
+import os
+import shutil
+import socket
FAKE_INVITE = "localhost:65535/pVOZMJGm3MqTvTu0UnhMGb2cfuqygiu79MdnERnGYdga5v8C"
"""Test key management commands."""
-import os
-import typing
-
-from testlib import check, util
from testlib.log import log
from testlib.const import RUN_ACCESS_CHECKS
from testlib.feature import Feature
from testlib.proc import Tinc
from testlib.test import Test
+from testlib import check, util
+
+import os
+import typing
TEST_DATA = b"foo bar baz"
"""Test miscellaneous commands."""
-import os
-import typing as T
-
from testlib import check, cmd
from testlib.log import log
from testlib.proc import Tinc, Script
from testlib.test import Test
+import os
+import typing as T
+
SUBNETS_BAR = ("10.20.30.40", "fe80::")
"""Test sign/verify commands."""
-import os
-import tempfile
-
-from testlib import util, cmd, check
from testlib.proc import Tinc
from testlib.test import Test
+from testlib import util, cmd, check
+
+import os
+import tempfile
PRIV_KEY = """
-----BEGIN ED25519 PRIVATE KEY-----
T8Bjg7dc7IjsCrZQC/20qLRsWPlrbthnjyDHQM0BMLoTeAHbLt0fxP5CbTy7Cifgg7P0K179GeahBFsnaIr4MA\n\
fake testing data\n\
hello there\n\
-""".encode("utf-8")
+""".encode(
+ "utf-8"
+)
raw_fd, raw_path = tempfile.mkstemp()
"""Test supported and unsupported commandline flags."""
+from testlib.log import log
+from testlib.proc import Tinc, Script
+from testlib.test import Test
+from testlib.feature import SANDBOX_LEVEL
+from testlib import check, util, path
+
import os
import shutil
import signal
import tempfile
import time
-from testlib import check, util, path
-from testlib.log import log
-from testlib.proc import Tinc, Script
-from testlib.test import Test
-from testlib.feature import SANDBOX_LEVEL
-
def init(ctx: Test) -> Tinc:
"""Initialize new test nodes."""
"""Test supported and unsupported compression levels."""
+from testlib import cmd, path, check, util
+from testlib.log import log
+from testlib.proc import Script, Tinc, Feature
+from testlib.test import Test
+from testlib.template import make_netns_config
+from testlib import external as ext
+
import os
import signal
import sys
import time
import typing as T
-from testlib import external as ext, cmd, path, check, util
-from testlib.log import log
-from testlib.proc import Script, Tinc, Feature
-from testlib.test import Test
-from testlib.template import make_netns_config
-
IP_FOO = "192.168.1.1"
IP_BAR = "192.168.1.2"
MASK = 24
"""Test device configuration variables."""
-import os
-import platform
-import typing as T
-
from testlib import check
from testlib.feature import Feature
from testlib.log import log
from testlib.proc import Script
from testlib.test import Test
+import os
+import platform
+import typing as T
+
system = platform.system()
"""Test FD device support."""
+from testlib import check
+from testlib.log import log
+from testlib.test import Test
+from testlib.proc import Script
+
import array
import socket
import tempfile
import threading
import time
-from testlib import check
-from testlib.log import log
-from testlib.test import Test
-from testlib.proc import Script
-
JUNK_FRAME = b"\xff" * 80
"""Test multicast device."""
+from testlib.log import log
+from testlib.proc import Tinc, Script
+from testlib.test import Test
+from testlib import check
+
import enum
import os
import select
import struct
import time
-from testlib import check
-from testlib.log import log
-from testlib.proc import Tinc, Script
-from testlib.test import Test
-
MCAST_ADDR = "224.15.98.12"
PORT = 38245
try:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as server:
+ server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
server.bind((MCAST_ADDR, PORT))
req = struct.pack("=4sl", socket.inet_aton(MCAST_ADDR), socket.INADDR_ANY)
log.info("multicast blocked")
else:
log.info("multicast not supported")
- test_no_mcast_support(foo)
+ # test_no_mcast_support(foo)
with Test("test DeviceType = multicast") as context:
"""Test raw socket device support."""
-import sys
-import subprocess as subp
-import socket
-
-from testlib import check, util
from testlib.log import log
from testlib.const import EXIT_SKIP
from testlib.proc import Script
from testlib.test import Test
from testlib.external import veth_add, move_dev, ping
+from testlib import check, util
+
+import sys
+import subprocess as subp
+import socket
util.require_root()
util.require_command("ip", "link")
try:
s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.IPPROTO_IP)
- print("Success!")
except PermissionError:
log.info("This test requires raw socket privileges")
sys.exit(77)
"""Test TAP device support."""
-import typing as T
-
-from testlib import check, util, cmd
from testlib.log import log
from testlib.proc import Script, Tinc
from testlib.test import Test
from testlib.external import netns_add, netns_exec, ping
+from testlib import check, util, cmd
+
+import typing as T
util.require_root()
util.require_command("ip", "netns", "list")
"""Basic sanity checks on compiled executables."""
-from subprocess import run, PIPE
-
from testlib import path, check
from testlib.log import log
+from subprocess import run, PIPE
+
for exe in (
path.TINC_PATH,
path.TINCD_PATH,
"""Test tinc peer invitations."""
-import time
-import subprocess as subp
-
-from testlib import check, util
from testlib.proc import Tinc
from testlib.log import log
from testlib.test import Test
+from testlib import check, util
+
+import time
+import subprocess as subp
def run_port0_test(ctx: Test) -> None:
"""Test inviting tinc nodes through tinc-up script."""
-import os
-import typing as T
-
-from testlib import check, util
from testlib.log import log
from testlib.proc import Tinc, Script
from testlib.test import Test
+from testlib import check, util
+
+import os
+import typing as T
IFCONFIG = "93.184.216.34/24"
ROUTES_IPV6 = ("2606:2800:220:1::/64", "2606:2800:220:1:248:1893:25c8:1946")
"""Test legacy protocol support (tinc 1.0)."""
-import typing as T
-
from testlib import check, cmd
from testlib.log import log
from testlib.proc import Tinc, Script
from testlib.test import Test
+import typing as T
+
TIMEOUT = 2
"""Create two network namespaces and run ping between them."""
-import typing as T
-
-from testlib import external as ext, util, template, cmd
from testlib.log import log
from testlib.proc import Tinc, Script
from testlib.test import Test
from testlib.external import ping
+from testlib import external as ext, util, template, cmd
+
+import typing as T
util.require_root()
util.require_command("ip", "netns", "list")
"""Test that tincd works through proxies."""
+from threading import Thread
+from socketserver import ThreadingMixIn, TCPServer, StreamRequestHandler
+from testlib import check, cmd, path, util
+from testlib.proc import Tinc, Script
+from testlib.test import Test
+from testlib.util import random_string
+from testlib.log import log
+from testlib.feature import HAVE_SANDBOX
+
import os
import re
import time
import socket
import struct
-from threading import Thread
-from socketserver import ThreadingMixIn, TCPServer, StreamRequestHandler
-from testlib import check, cmd, path, util
-from testlib.proc import Tinc, Script
-from testlib.test import Test
-from testlib.util import random_string
-from testlib.log import log
-from testlib.feature import HAVE_SANDBOX
-
USERNAME = random_string(8)
PASSWORD = random_string(8)
"""Test that tincd works through proxies."""
-import os
-import time
-
from testlib import check, cmd, path, util
from testlib.proc import Tinc, Script
from testlib.test import Test
from testlib.log import log
from testlib.feature import HAVE_SANDBOX
+import os
+import time
+
def init(ctx: Test, level: str) -> Tinc:
"""Create a new tinc node."""
"""Test that all tincd scripts execute in correct order and contain expected env vars."""
-import os
-import typing as T
-
from testlib import check, path
from testlib.log import log
from testlib.proc import Tinc, Script, ScriptType, TincScript
from testlib.test import Test
from testlib.util import random_string
+import os
+import typing as T
+
SUBNET_SERVER = ("10.0.0.1", "fec0::/64")
SUBNET_CLIENT = ("10.0.0.2", "fec0::/64#5")
NETNAMES = {
"""Test tinc protocol security."""
-import asyncio
-import typing as T
-
from testlib import check
from testlib.log import log
from testlib.proc import Tinc, Script
from testlib.test import Test
from testlib.feature import SANDBOX_LEVEL
+import asyncio
+import typing as T
+
TIMEOUT = 2
)
-loop = asyncio.get_event_loop()
+loop = asyncio.new_event_loop()
with Test("security") as context:
loop.run_until_complete(run_tests(context))
"""Test splicing connection between tinc peers."""
-import os
-import subprocess as subp
-import typing as T
-
from testlib import check, cmd, path
from testlib.log import log
from testlib.proc import Tinc, Script
from testlib.test import Test
from testlib.feature import SANDBOX_LEVEL
+import os
+import subprocess as subp
+import typing as T
+
def init(ctx: Test, *options: str) -> T.Tuple[Tinc, Tinc]:
"""Initialize new test nodes."""
"""Test basic SPTPS features."""
+from testlib.log import log
+from testlib.proc import Tinc, Script
+from testlib.test import Test
+from testlib import check, util, path
+
import os
import subprocess as subp
import re
-from testlib import path, util, check
-from testlib.log import log
-
port_re = re.compile(r"Listening on (\d+)\.\.\.")
"""Test systemd integration."""
-import os
-import socket
-import tempfile
-import time
-
from testlib import check, path
from testlib.log import log
from testlib.feature import Feature
from testlib.proc import Tinc, Script
from testlib.test import Test
+import os
+import socket
+import tempfile
+import time
+
def tincd_start_socket(foo: Tinc, pass_pid: bool) -> int:
"""Start tincd as systemd socket activation does it."""
"""Classes for working with compiled instances of tinc and tincd binaries."""
+from . import check, path
+from .log import log
+from .script import TincScript, Script, ScriptType
+from .template import make_script, make_cmd_wrap
+from .util import random_string, random_port
+
import os
import random
import tempfile
from enum import Enum
from platform import system
-from . import check, path
-from .log import log
-from .script import TincScript, Script, ScriptType
-from .template import make_script, make_cmd_wrap
-from .util import random_string, random_port
-
# Does the OS support all addresses in 127.0.0.0/8 without additional configuration?
_FULL_LOCALHOST_SUBNET = system() in ("Linux", "Windows")
--- /dev/null
+"""Wrappers for more complicated tinc/tincd commands."""
+
+import typing as T
+
+from . import check
+from .log import log
+from .proc import Tinc
+
+ExchangeIO = T.Tuple[
+ T.Tuple[str, str],
+ T.Tuple[str, str],
+ T.Tuple[str, str],
+]
+
+
+def connect(node0: Tinc, node1: Tinc) -> ExchangeIO:
+ """Exchange configuration between nodes and start
+ them in such an order that `Port 0` works on both sides.
+ """
+ node0.add_script(node1.script_up)
+ node0.start()
+ result = exchange(node0, node1)
+ node1.add_script(node0.script_up)
+ node1.cmd("add", "ConnectTo", node0.name)
+ node1.start()
+ node0[node1.script_up].wait()
+ node1[node0.script_up].wait()
+ return result
+
+
+def exchange(node0: Tinc, node1: Tinc, export_all: bool = False) -> ExchangeIO:
+ """Run `export(-all) | exchange | import` between the passed nodes.
+ `export-all` is used if export_all is set to True.
+ """
+ export_cmd = "export-all" if export_all else "export"
+ log.debug("%s between %s and %s", export_cmd, node0.name, node1.name)
+
+ exp_out, exp_err = node0.cmd(export_cmd)
+ log.debug(
+ 'exchange: %s %s returned ("%s", "%s")', export_cmd, node0, exp_out, exp_err
+ )
+ check.is_in("Name =", exp_out)
+
+ xch_out, xch_err = node1.cmd("exchange", stdin=exp_out)
+ log.debug('exchange: exchange %s returned ("%s", "%s")', node1, xch_out, xch_err)
+ check.is_in("Name =", xch_out)
+ check.is_in("Imported ", xch_err)
+
+ imp_out, imp_err = node0.cmd("import", stdin=xch_out)
+ log.debug('exchange: import %s returned ("%s", "%s")', node0, imp_out, imp_err)
+ check.is_in("Imported ", imp_err)
+
+ return (
+ (exp_out, exp_err),
+ (xch_out, xch_err),
+ (imp_out, imp_err),
+ )
+
+
+def get(tinc: Tinc, var: str) -> str:
+ """Get the value of the variable, stripped of whitespace."""
+ assert var
+ stdout, _ = tinc.cmd("get", var)
+ return stdout.strip()
--- /dev/null
+"""Global logger for using in test and tincd scripts."""
+
+import logging
+import os
+import sys
+import typing as T
+from types import TracebackType
+
+from .path import TEST_WD, TEST_NAME
+
+logging.basicConfig(level=logging.DEBUG)
+
+_fmt = logging.Formatter(
+ "%(asctime)s %(name)s %(filename)s:%(lineno)d %(levelname)s %(message)s"
+)
+
+# Where to put log files for this test and nodes started by it
+_log_dir = os.path.join(TEST_WD, "logs")
+
+
+def new_logger(name: str) -> logging.Logger:
+ """Create a new named logger with common logging format.
+ Log entries will go into a separate logfile named 'name.log'.
+ """
+ os.makedirs(_log_dir, exist_ok=True)
+
+ logger = logging.getLogger(name)
+ logger.setLevel(logging.DEBUG)
+
+ file = logging.FileHandler(os.path.join(_log_dir, name + ".log"))
+ file.setFormatter(_fmt)
+ logger.addHandler(file)
+
+ return logger
+
+
+# Main logger used by most tests
+log = new_logger(TEST_NAME)
+
+
+def _exc_hook(
+ ex_type: T.Type[BaseException],
+ base: BaseException,
+ tb_type: T.Optional[TracebackType],
+) -> None:
+ """Logging handler for uncaught exceptions."""
+ log.error("Uncaught exception", exc_info=(ex_type, base, tb_type))
+
+
+sys.excepthook = _exc_hook
--- /dev/null
+"""Miscellaneous utility functions."""
+
+from . import check
+from .log import log
+from .const import EXIT_SKIP
+
+import os
+import sys
+import subprocess as subp
+import random
+import string
+import socket
+import typing as T
+import tempfile
+from pathlib import Path
+
+_ALPHA_NUMERIC = string.ascii_lowercase + string.digits
+
+
+def random_port() -> int:
+ """Return an unused TCP port in the unprivileged range.
+ Note that this function releases the port before returning, and it can be
+ overtaken by something else before you use it.
+ """
+ while True:
+ port = random.randint(1024, 65535)
+ try:
+ with socket.socket() as sock:
+ sock.bind(("0.0.0.0", port))
+ sock.listen()
+ return port
+ except OSError as ex:
+ log.debug("could not bind to random port %d", port, exc_info=ex)
+
+
+def temp_file(content: str) -> str:
+ """Create a temporary file and write text content into it."""
+ fd, path = tempfile.mkstemp()
+ with os.fdopen(fd, "w", encoding="utf-8") as f:
+ f.write(content)
+ return path
+
+
+def remove_file(path: T.Union[str, Path]) -> bool:
+ """Try to remove file without failing if it does not exist."""
+ try:
+ os.remove(path)
+ return True
+ except FileNotFoundError:
+ return False
+
+
+def random_string(k: int) -> str:
+ """Generate a random alphanumeric string of length k."""
+ return "".join(random.choices(_ALPHA_NUMERIC, k=k))
+
+
+def find_line(filename: str, prefix: str) -> str:
+ """Find a line with the prefix in a text file.
+ Check that only one line matches.
+ """
+ with open(filename, "r", encoding="utf-8") as f:
+ keylines = [line for line in f.readlines() if line.startswith(prefix)]
+ check.equals(1, len(keylines))
+ return keylines[0].rstrip()
+
+
+def require_root() -> None:
+ """Check that test is running with root privileges.
+ Exit with code 77 otherwise.
+ """
+ euid = os.geteuid()
+ if euid:
+ log.info("this test requires root (but running under UID %d)", euid)
+ sys.exit(EXIT_SKIP)
+
+
+def require_command(*args: str) -> None:
+ """Check that command args runs with exit code 0.
+ Exit with code 77 otherwise.
+ """
+ try:
+ if subp.run(args, check=False).returncode == 0:
+ return
+ except FileNotFoundError:
+ pass
+ log.info('this test requires command "%s" to work', " ".join(args))
+ sys.exit(EXIT_SKIP)
+
+
+def require_path(path: str) -> None:
+ """Check that path exists in your file system.
+ Exit with code 77 otherwise.
+ """
+ if not os.path.exists(path):
+ log.warning("this test requires path %s to be present", path)
+ sys.exit(EXIT_SKIP)
+
+
+# Thin wrappers around `with open(...) as f: f.do_something()`
+# Don't do much, besides saving quite a bit of space because of how frequently they're needed.
+
+
+def read_text(path: str) -> str:
+ """Return the text contents of a file."""
+ with open(path, encoding="utf-8") as f:
+ return f.read()
+
+
+def write_text(path: str, text: str) -> str:
+ """Write text to a file, replacing its content. Return the text added."""
+ with open(path, "w", encoding="utf-8") as f:
+ f.write(text)
+ return text
+
+
+def read_lines(path: str) -> T.List[str]:
+ """Read file as a list of lines."""
+ with open(path, encoding="utf-8") as f:
+ return f.read().splitlines()
+
+
+def write_lines(path: str, lines: T.List[str]) -> T.List[str]:
+ """Write text lines to a file, replacing it content. Return the line added."""
+ with open(path, "w", encoding="utf-8") as f:
+ f.write(os.linesep.join(lines))
+ f.write(os.linesep)
+ return lines
+
+
+def append_line(path: str, line: str) -> str:
+ """Append a line to the end of the file. Return the line added."""
+ line = f"{os.linesep}{line}{os.linesep}"
+ with open(path, "a", encoding="utf-8") as f:
+ f.write(line)
+ return line
"""Miscellaneous utility functions."""
+from . import check
+from .log import log
+from .const import EXIT_SKIP
+
import os
import sys
import subprocess as subp
import tempfile
from pathlib import Path
-from . import check
-from .log import log
-from .const import EXIT_SKIP
-
_ALPHA_NUMERIC = string.ascii_lowercase + string.digits
"""Test tinc and tincd configuration variables."""
-import typing as T
-from pathlib import Path
-
from testlib import check, cmd
from testlib.log import log
from testlib.proc import Tinc
from testlib.test import Test
+import typing as T
+from pathlib import Path
+
bad_subnets = (
"1.1.1",
"1:2:3:4:5:",
projects / tinc / commitdiff