www.tinc-vpn.org Git - tinc/blob - test/integration/splice.py

   1 #!/usr/bin/env python3
   2
   3 """Test splicing connection between tinc peers."""
   4
   5 import os
   6 import subprocess as subp
   7 import typing as T
   8
   9 from testlib import check, cmd, path
  10 from testlib.log import log
  11 from testlib.proc import Tinc, Script
  12 from testlib.test import Test
  13 from testlib.feature import SANDBOX_LEVEL
  14
  15
  16 def init(ctx: Test, *options: str) -> T.Tuple[Tinc, Tinc]:
  17     """Initialize new test nodes."""
  18     custom = os.linesep.join(options)
  19     log.info('init two nodes with options "%s"', custom)
  20
  21     foo, bar = ctx.node(), ctx.node()
  22
  23     stdin = f"""
  24         init {foo}
  25         set Port 0
  26         set DeviceType dummy
  27         set Address localhost
  28         set AutoConnect no
  29         set Subnet 10.96.96.1
  30         set Sandbox {SANDBOX_LEVEL}
  31         {custom}
  32     """
  33     foo.cmd(stdin=stdin)
  34
  35     stdin = f"""
  36         init {bar}
  37         set Port 0
  38         set Address localhost
  39         set DeviceType dummy
  40         set AutoConnect no
  41         set Subnet 10.96.96.2
  42         set Sandbox {SANDBOX_LEVEL}
  43         {custom}
  44     """
  45     bar.cmd(stdin=stdin)
  46
  47     foo.add_script(Script.SUBNET_UP)
  48     bar.add_script(Script.SUBNET_UP)
  49
  50     foo.start()
  51     bar.start()
  52
  53     log.info("exchange host configs")
  54     cmd.exchange(foo, bar)
  55
  56     return foo, bar
  57
  58
  59 def splice(foo: Tinc, bar: Tinc, protocol: str) -> subp.Popen:
  60     """Start splice between nodes."""
  61     args = [
  62         path.SPLICE_PATH,
  63         foo.name,
  64         "localhost",
  65         str(foo.port),
  66         bar.name,
  67         "localhost",
  68         str(bar.port),
  69         protocol,
  70     ]
  71     log.info("splice with args %s", args)
  72     return subp.Popen(args)
  73
  74
  75 def test_splice(ctx: Test, protocol: str, *options: str) -> None:
  76     """Splice connection and check that it fails."""
  77     log.info("no splicing allowed (%s)", protocol)
  78     foo, bar = init(ctx, *options)
  79
  80     log.info("waiting for subnets to come up")
  81     foo[Script.SUBNET_UP].wait()
  82     bar[Script.SUBNET_UP].wait()
  83
  84     splice_proc = splice(foo, bar, protocol)
  85     try:
  86         check.nodes(foo, 1)
  87         check.nodes(bar, 1)
  88     finally:
  89         splice_proc.kill()
  90
  91
  92 with Test("sptps") as context:
  93     test_splice(context, "17.7")
  94
  95 with Test("legacy") as context:
  96     test_splice(context, "17.0", "set ExperimentalProtocol no")