src/nepi/resources/linux/ns3/fdudptunnel.py

   1 #
   2 #    NEPI, a framework to manage network experiments
   3 #    Copyright (C) 2013 INRIA
   4 #
   5 #    This program is free software: you can redistribute it and/or modify
   6 #    it under the terms of the GNU General Public License as published by
   7 #    the Free Software Foundation, either version 3 of the License, or
   8 #    (at your option) any later version.
   9 #
  10 #    This program is distributed in the hope that it will be useful,
  11 #    but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 #    GNU General Public License for more details.
  14 #
  15 #    You should have received a copy of the GNU General Public License
  16 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 #
  18 # Author: Alina Quereilhac <alina.quereilhac@inria.fr>
  19
  20 from nepi.execution.attribute import Attribute, Flags, Types
  21 from nepi.execution.resource import clsinit_copy, ResourceState
  22 from nepi.resources.linux.udptunnel import LinuxUdpTunnel
  23 from nepi.util.sshfuncs import ProcStatus
  24 from nepi.util.timefuncs import tnow, tdiffsec
  25
  26 import base64
  27 import os
  28 import socket
  29 import time
  30
  31 @clsinit_copy
  32 class LinuxNs3FdUdpTunnel(LinuxUdpTunnel):
  33     _rtype = "linux::ns3::FdUdpTunnel"
  34     _help = "Constructs a tunnel between two Ns-3 FdNetdevices " \
  35             "located in remote Linux nodes using a UDP connection "
  36     _platform = "linux::ns3"
  37
  38     @classmethod
  39     def _register_attributes(cls):
  40         cipher = Attribute("cipher",
  41                "Cipher to encript communication. "
  42                 "One of PLAIN, AES, Blowfish, DES, DES3. ",
  43                 default = None,
  44                 allowed = ["PLAIN", "AES", "Blowfish", "DES", "DES3"],
  45                 type = Types.Enumerate,
  46                 flags = Flags.Design)
  47
  48         cipher_key = Attribute("cipherKey",
  49                 "Specify a symmetric encryption key with which to protect "
  50                 "packets across the tunnel. python-crypto must be installed "
  51                 "on the system." ,
  52                 flags = Flags.Design)
  53
  54         txqueuelen = Attribute("txQueueLen",
  55                 "Specifies the interface's transmission queue length. "
  56                 "Defaults to 1000. ",
  57                 type = Types.Integer,
  58                 flags = Flags.Design)
  59
  60         bwlimit = Attribute("bwLimit",
  61                 "Specifies the interface's emulated bandwidth in bytes "
  62                 "per second.",
  63                 type = Types.Integer,
  64                 flags = Flags.Design)
  65
  66         cls._register_attribute(cipher)
  67         cls._register_attribute(cipher_key)
  68         cls._register_attribute(txqueuelen)
  69         cls._register_attribute(bwlimit)
  70
  71     def __init__(self, ec, guid):
  72         super(LinuxUdpTunnel, self).__init__(ec, guid)
  73         self._home = "fd-udp-tunnel-%s" % self.guid
  74         self._pids = dict()
  75         self._fd1 = None
  76         self._fd1node = None
  77         self._fd2 = None
  78         self._fd2node = None
  79         self._pi = False
  80
  81     def log_message(self, msg):
  82         self.get_endpoints()
  83         return " guid %d - %s - %s - %s " % (self.guid,
  84                 self.node1.get("hostname"),
  85                 self.node2.get("hostname"),
  86                 msg)
  87
  88     def get_endpoints(self):
  89         """ Returns the list of RM that are endpoints to the tunnel
  90         """
  91         if not self._fd2 or not self._fd1:
  92             from nepi.resources.ns3.ns3fdnetdevice import NS3BaseFdNetDevice
  93             devices = self.get_connected(NS3BaseFdNetDevice.get_rtype())
  94             if not devices or len(devices) != 2:
  95                 msg = "linux::ns3::TunTapFdLink must be connected to exactly one FdNetDevice"
  96                 self.error(msg)
  97                 raise RuntimeError, msg
  98
  99             self._fd1 = devices[0]
 100             self._fd2 = devices[1]
 101
 102             simu = self._fd1.simulation
 103             from nepi.resources.linux.node import LinuxNode
 104             nodes = simu.get_connected(LinuxNode.get_rtype())
 105             self._fd1node = nodes[0]
 106
 107             simu = self._fd2.simulation
 108             from nepi.resources.linux.node import LinuxNode
 109             nodes = simu.get_connected(LinuxNode.get_rtype())
 110             self._fd2node = nodes[0]
 111
 112             if self._fd1node.get("hostname") == \
 113                     self._fd2node.get("hostname"):
 114                 msg = "linux::ns3::FdUdpTunnel requires endpoints on different hosts"
 115                 self.error(msg)
 116                 raise RuntimeError, msg
 117
 118         return [self._fd1, self._fd2]
 119
 120     @property
 121     def pi(self):
 122         return self._pi
 123
 124     @property
 125     def endpoint1(self):
 126         return self._fd1
 127
 128     @property
 129     def endpoint2(self):
 130         return self._fd2
 131
 132     @property
 133     def node1(self):
 134         return self._fd1node
 135
 136     @property
 137     def node2(self):
 138         return self._fd2node
 139
 140     def endpoint_node(self, endpoint):
 141         node = None
 142         if endpoint == self.endpoint1:
 143             node = self.node1
 144         else:
 145             node = self.node2
 146
 147         return node
 148
 149     def app_home(self, endpoint):
 150         node = self.endpoint_node(endpoint)
 151         return os.path.join(node.exp_home, self._home)
 152
 153     def run_home(self, endpoint):
 154         return os.path.join(self.app_home(endpoint), self.ec.run_id)
 155
 156     def upload_sources(self, endpoint):
 157         scripts = []
 158
 159         # vif-passfd python script
 160         linux_passfd = os.path.join(os.path.dirname(__file__),
 161                 "..",
 162                 "scripts",
 163                 "fd-udp-connect.py")
 164
 165         scripts.append(linux_passfd)
 166
 167         # tunnel creation python script
 168         tunchannel = os.path.join(os.path.dirname(__file__),
 169                 "..",
 170                 "scripts",
 171                 "tunchannel.py")
 172
 173         scripts.append(tunchannel)
 174
 175         # Upload scripts
 176         scripts = ";".join(scripts)
 177
 178         node = self.endpoint_node(endpoint)
 179         node.upload(scripts,
 180                 os.path.join(node.src_dir),
 181                 overwrite = False)
 182
 183     def endpoint_mkdir(self, endpoint):
 184         node = self.endpoint_node(endpoint)
 185         run_home = self.run_home(endpoint)
 186         node.mkdir(run_home)
 187
 188     def initiate_connection(self, endpoint, remote_endpoint):
 189         cipher = self.get("cipher")
 190         cipher_key = self.get("cipherKey")
 191         bwlimit = self.get("bwLimit")
 192         txqueuelen = self.get("txQueueLen")
 193
 194         # Upload the tunnel creating script
 195         self.upload_sources(endpoint)
 196
 197         # Request an address to send the file descriptor to the ns-3 simulation
 198         address = endpoint.recv_fd()
 199
 200         # execute the tunnel creation script
 201         node = self.endpoint_node(remote_endpoint)
 202         port = self.initiate(endpoint, remote_endpoint, address, cipher,
 203                 cipher_key, bwlimit, txqueuelen)
 204
 205         return port
 206
 207     def establish_connection(self, endpoint, remote_endpoint, port):
 208         self.establish(endpoint, remote_endpoint, port)
 209
 210     def verify_connection(self, endpoint, remote_endpoint):
 211         self.verify(endpoint)
 212
 213     def terminate_connection(self, endpoint, remote_endpoint):
 214         # Nothing to do
 215         return
 216
 217     def check_state_connection(self):
 218         # Make sure the process is still running in background
 219         # No execution errors occurred. Make sure the background
 220         # process with the recorded pid is still running.
 221
 222         node1 = self.endpoint_node(self.endpoint1)
 223         node2 = self.endpoint_node(self.endpoint2)
 224         run_home1 = self.run_home(self.endpoint1)
 225         run_home2 = self.run_home(self.endpoint1)
 226         (pid1, ppid1) = self._pids[endpoint1]
 227         (pid2, ppid2) = self._pids[endpoint2]
 228
 229         status1 = node1.status(pid1, ppid1)
 230         status2 = node2.status(pid2, ppid2)
 231
 232         if status1 == ProcStatus.FINISHED and \
 233                 status2 == ProcStatus.FINISHED:
 234
 235             # check if execution errors occurred
 236             (out1, err1), proc1 = node1.check_errors(run_home1)
 237             (out2, err2), proc2 = node2.check_errors(run_home2)
 238
 239             if err1 or err2:
 240                 msg = "Error occurred in tunnel"
 241                 self.error(msg, err1, err2)
 242                 self.fail()
 243             else:
 244                 self.set_stopped()
 245
 246     def wait_local_port(self, endpoint):
 247         """ Waits until the local_port file for the endpoint is generated,
 248         and returns the port number
 249
 250         """
 251         return self.wait_file(endpoint, "local_port")
 252
 253     def wait_result(self, endpoint):
 254         """ Waits until the return code file for the endpoint is generated
 255
 256         """
 257         return self.wait_file(endpoint, "ret_file")
 258
 259     def wait_file(self, endpoint, filename):
 260         """ Waits until file on endpoint is generated """
 261         result = None
 262         delay = 1.0
 263
 264         node = self.endpoint_node(endpoint)
 265         run_home = self.run_home(endpoint)
 266
 267         for i in xrange(20):
 268             (out, err), proc = node.check_output(run_home, filename)
 269
 270             if out:
 271                 result = out.strip()
 272                 break
 273             else:
 274                 time.sleep(delay)
 275                 delay = delay * 1.5
 276         else:
 277             msg = "Couldn't retrieve %s" % filename
 278             self.error(msg, out, err)
 279             raise RuntimeError, msg
 280
 281         return result
 282
 283     def initiate(self, endpoint, remote_endpoint, address, cipher, cipher_key,
 284             bwlimit, txqueuelen):
 285
 286         command = self._initiate_command(endpoint, remote_endpoint,
 287                 address, cipher, cipher_key, bwlimit, txqueuelen)
 288
 289         node = self.endpoint_node(endpoint)
 290         run_home = self.run_home(endpoint)
 291         app_home = self.app_home(endpoint)
 292
 293         # upload command to connect.sh script
 294         shfile = os.path.join(app_home, "fd-udp-connect.sh")
 295         node.upload_command(command,
 296                 shfile = shfile,
 297                 overwrite = False)
 298
 299         # invoke connect script
 300         cmd = "bash %s" % shfile
 301         (out, err), proc = node.run(cmd, run_home)
 302
 303         # check if execution errors occurred
 304         msg = "Failed to connect endpoints "
 305
 306         if proc.poll():
 307             self.error(msg, out, err)
 308             raise RuntimeError, msg
 309
 310         # Wait for pid file to be generated
 311         pid, ppid = node.wait_pid(run_home)
 312
 313         self._pids[endpoint] = (pid, ppid)
 314
 315         # Check for error information on the remote machine
 316         (out, err), proc = node.check_errors(run_home)
 317         # Out is what was written in the stderr file
 318         if err:
 319             msg = " Failed to start command '%s' " % command
 320             self.error(msg, out, err)
 321             raise RuntimeError, msg
 322
 323         port = self.wait_local_port(endpoint)
 324
 325         return port
 326
 327     def _initiate_command(self, endpoint, remote_endpoint, address,
 328             cipher, cipher_key, bwlimit, txqueuelen):
 329         local_node = self.endpoint_node(endpoint)
 330         local_run_home = self.run_home(endpoint)
 331         local_app_home = self.app_home(endpoint)
 332         remote_node = self.endpoint_node(remote_endpoint)
 333
 334         local_ip = local_node.get("ip")
 335         remote_ip = remote_node.get("ip")
 336
 337         local_port_file = os.path.join(local_run_home, "local_port")
 338         remote_port_file = os.path.join(local_run_home,  "remote_port")
 339         ret_file = os.path.join(local_run_home, "ret_file")
 340
 341         address = base64.b64encode(address)
 342
 343         command = [""]
 344         command.append("PYTHONPATH=$PYTHONPATH:${SRC}")
 345         command.append("python ${SRC}/fd-udp-connect.py")
 346         command.append("-a %s" % address)
 347         command.append("-p %s" % local_port_file)
 348         command.append("-P %s" % remote_port_file)
 349         command.append("-o %s" % local_ip)
 350         command.append("-O %s" % remote_ip)
 351         command.append("-R %s" % ret_file)
 352         command.append("-t %s" % "IFF_TAP")
 353         if self.pi:
 354             command.append("-n")
 355         if cipher:
 356             command.append("-c %s" % cipher)
 357         if cipher_key:
 358             command.append("-k %s " % cipher_key)
 359         if txqueuelen:
 360             command.append("-q %s " % txqueuelen)
 361         if bwlimit:
 362             command.append("-b %s " % bwlimit)
 363
 364         command = " ".join(command)
 365         command = self.replace_paths(command, node=local_node,
 366                 app_home=local_app_home, run_home=local_run_home)
 367
 368         return command
 369
 370     def establish(self, endpoint, remote_endpoint, port):
 371         node = self.endpoint_node(endpoint)
 372         run_home = self.run_home(endpoint)
 373
 374         # upload remote port number to file
 375         remote_port = "%s\n" % port
 376         node.upload(remote_port,
 377                 os.path.join(run_home, "remote_port"),
 378                 text = True,
 379                 overwrite = False)
 380
 381     def verify(self, endpoint):
 382         self.wait_result(endpoint)
 383