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