1 # -*- coding: utf-8 -*-
3 from constants import TESTBED_ID
4 import nepi.util.ipaddr2 as ipaddr2
5 import nepi.util.server as server
16 class NodeIface(object):
17 def __init__(self, api=None):
25 # These get initialized at configuration time
31 self._interface_id = None
33 # These get initialized when the iface is connected to its node
36 # These get initialized when the iface is connected to the internet
37 self.has_internet = False
40 return "%s<ip:%s/%s up mac:%s>" % (
41 self.__class__.__name__,
42 self.address, self.netmask,
48 def add_address(self, address, netprefix, broadcast):
49 raise RuntimeError, "Cannot add explicit addresses to public interface"
51 def pick_iface(self, siblings):
53 Picks an interface using the PLCAPI to query information about the node.
55 Needs an assigned node.
58 siblings: other NodeIface elements attached to the same node
61 if self.node is None or self.node._node_id is None:
62 raise RuntimeError, "Cannot pick interface without an assigned node"
64 # HACK: SFA doesnt give the node_id!!
65 if not isinstance(self.node._node_id, int):
66 node_data = self._api.GetNodes(filters={'hostname':self.node.hostname}, fields=('node_id',))[0]
67 node_id = node_data['node_id']
69 node_id = self.node._node_id
71 avail = self._api.GetInterfaces(
73 is_primary=self.primary,
74 fields=('interface_id','mac','netmask','ip') )
76 used = set([sibling._interface_id for sibling in siblings
77 if sibling._interface_id is not None])
79 for candidate in avail:
80 candidate_id = candidate['interface_id']
81 if candidate_id not in used:
83 self._interface_id = candidate_id
84 self.address = candidate['ip']
85 self.lladdr = candidate['mac']
86 self.netprefix = candidate['netmask']
87 self.netmask = ipaddr2.ipv4_dot2mask(self.netprefix) if self.netprefix else None
90 raise RuntimeError, "Cannot configure interface: cannot find suitable interface in PlanetLab node"
93 if not self.has_internet:
94 raise RuntimeError, "All external interface devices must be connected to the Internet"
97 class _CrossIface(object):
98 def __init__(self, proto, addr, port, cipher):
99 self.tun_proto = proto
102 self.tun_cipher = cipher
106 self.netprefix = None
109 # Cannot access cross peers
110 self.peer_proto_impl = None
114 self.__class__.__name__,
123 class TunIface(object):
124 _PROTO_MAP = tunproto.TUN_PROTO_MAP
127 def __init__(self, api=None):
129 api = plcapi.PLCAPI()
134 self.netprefix = None
140 self.txqueuelen = 1000
141 self.pointopoint = None
142 self.multicast = False
148 # These get initialized when the iface is connected to its node
151 # These get initialized when the iface is connected to any filter
152 self.filter_module = None
153 self.multicast_forwarder = None
155 # These get initialized when the iface is configured
156 self.external_iface = None
158 # These get initialized when the iface is configured
159 # They're part of the TUN standard attribute set
162 self.tun_cipher = "AES"
164 # These get initialized when the iface is connected to its peer
165 self.peer_iface = None
166 self.peer_proto = None
167 self.peer_addr = None
168 self.peer_port = None
169 self.peer_proto_impl = None
170 self._delay_recover = False
172 # same as peer proto, but for execute-time standard attribute lookups
173 self.tun_proto = None
176 # Generate an initial random cryptographic key to use for tunnelling
177 # Upon connection, both endpoints will agree on a common one based on
179 self.tun_key = ( ''.join(map(chr, [
182 for r in (random.SystemRandom(),) ])
183 ).encode("base64").strip() )
187 return "%s<ip:%s/%s %s%s%s>" % (
188 self.__class__.__name__,
189 self.address, self.netprefix,
190 " up" if self.up else " down",
191 " snat" if self.snat else "",
192 (" p2p %s" % (self.pointopoint,)) if self.pointopoint else "",
199 if self.peer_proto_impl:
200 return self.peer_proto_impl.if_name
203 if self.peer_proto_impl:
204 return self.peer_proto_impl.if_up()
207 if self.peer_proto_impl:
208 return self.peer_proto_impl.if_down()
210 def routes_here(self, route):
212 Returns True if the route should be attached to this interface
213 (ie, it references a gateway in this interface's network segment)
215 if self.address and self.netprefix:
216 addr, prefix = self.address, self.netprefix
217 pointopoint = self.pointopoint
218 if not pointopoint and self.peer_iface:
219 pointopoint = self.peer_iface.address
224 dest, destprefix, nexthop, metric, device = route
226 myNet = ipaddr.IPv4Network("%s/%d" % (addr, prefix))
227 gwIp = ipaddr.IPv4Network(nexthop)
230 peerIp = ipaddr.IPv4Network(pointopoint)
239 def add_address(self, address, netprefix, broadcast):
240 if (self.address or self.netprefix or self.netmask) is not None:
241 raise RuntimeError, "Cannot add more than one address to %s interfaces" % (self._KIND,)
243 raise ValueError, "%s interfaces cannot broadcast in PlanetLab (%s)" % (self._KIND,broadcast)
245 self.address = address
246 self.netprefix = netprefix
247 self.netmask = ipaddr2.ipv4_mask2dot(netprefix)
251 raise RuntimeError, "Unconnected %s iface - missing node" % (self._KIND,)
252 if self.peer_iface and self.peer_proto not in self._PROTO_MAP:
253 raise RuntimeError, "Unsupported tunnelling protocol: %s" % (self.peer_proto,)
254 if not self.address or not self.netprefix or not self.netmask:
255 raise RuntimeError, "Misconfigured %s iface - missing address" % (self._KIND,)
256 if self.filter_module and self.peer_proto not in ('udp','tcp',None):
257 raise RuntimeError, "Miscofnigured TUN: %s - filtered tunnels only work with udp or tcp links" % (self,)
258 if self.tun_cipher != 'PLAIN' and self.peer_proto not in ('udp','tcp',None):
259 raise RuntimeError, "Miscofnigured TUN: %s - ciphered tunnels only work with udp or tcp links" % (self,)
261 def _impl_instance(self, home_path):
262 impl = self._PROTO_MAP[self.peer_proto](
263 self, self.peer_iface, home_path, self.tun_key)
264 impl.port = self.tun_port
265 impl.cross_slice = not self.peer_iface or isinstance(self.peer_iface, _CrossIface)
270 self.peer_proto_impl = self._impl_instance(
272 self.peer_proto_impl.recover()
274 self._delay_recover = True
276 def prepare(self, home_path):
277 if not self.peer_iface and (self.peer_proto and self.peer_addr):
279 self.peer_iface = _CrossIface(
285 if not self.peer_proto_impl:
286 self.peer_proto_impl = self._impl_instance(home_path)
287 if self._delay_recover:
288 self.peer_proto_impl.recover()
291 if self.peer_proto_impl:
292 self.peer_proto_impl.launch()
295 if self.peer_proto_impl:
296 self.peer_proto_impl.shutdown()
299 if self.peer_proto_impl:
300 self.peer_proto_impl.destroy()
301 self.peer_proto_impl = None
304 if self.peer_proto_impl:
305 self.peer_proto_impl.wait()
307 def sync_trace(self, local_dir, whichtrace, tracemap = None):
308 if self.peer_proto_impl:
309 return self.peer_proto_impl.sync_trace(local_dir, whichtrace,
314 def remote_trace_path(self, whichtrace, tracemap = None):
315 if self.peer_proto_impl:
316 return self.peer_proto_impl.remote_trace_path(whichtrace, tracemap)
320 def remote_trace_name(self, whichtrace):
323 class TapIface(TunIface):
324 _PROTO_MAP = tunproto.TAP_PROTO_MAP
327 # Yep, it does nothing - yet
328 class Internet(object):
329 def __init__(self, api=None):
331 api = plcapi.PLCAPI()
334 class NetPipe(object):
335 def __init__(self, api=None):
337 api = plcapi.PLCAPI()
353 # These get initialized when the pipe is connected to its node
355 self.configured = False
359 raise RuntimeError, "Undefined NetPipe mode"
360 if not self.portList:
361 raise RuntimeError, "Undefined NetPipe port list - must always define the scope"
362 if not (self.plrIn or self.bwIn or self.delayIn):
363 raise RuntimeError, "Undefined NetPipe inbound characteristics"
364 if not (self.plrOut or self.bwOut or self.delayOut):
365 raise RuntimeError, "Undefined NetPipe outbound characteristics"
367 raise RuntimeError, "Unconnected NetPipe"
369 def _add_pipedef(self, bw, plr, delay, options):
371 options.extend(("delay","%dms" % (delay,)))
373 options.extend(("bw","%.8fMbit/s" % (bw,)))
375 options.extend(("plr","%.8f" % (plr,)))
377 def _get_ruledef(self):
380 "@" if self.addrList else "",
385 if self.bwIn or self.plrIn or self.delayIn:
387 self._add_pipedef(self.bwIn, self.plrIn, self.delayIn, options)
388 if self.bwOut or self.plrOut or self.delayOut:
389 options.append("OUT")
390 self._add_pipedef(self.bwOut, self.plrOut, self.delayOut, options)
391 options = ' '.join(options)
393 return (scope,options)
396 # Rules are safe on their nodes
397 self.configured = True
401 scope, options = self._get_ruledef()
402 command = "sudo -S netconfig config %s %s %s" % (self.mode, scope, options)
404 (out,err),proc = server.popen_ssh_command(
406 host = self.node.hostname,
408 user = self.node.slicename,
410 ident_key = self.node.ident_path,
411 server_key = self.node.server_key
415 raise RuntimeError, "Failed instal build sources: %s %s" % (out,err,)
417 # we have to clean up afterwards
418 self.configured = True
423 scope, options = self._get_ruledef()
424 command = "sudo -S netconfig refresh %s %s %s" % (self.mode, scope, options)
426 (out,err),proc = server.popen_ssh_command(
428 host = self.node.hostname,
430 user = self.node.slicename,
432 ident_key = self.node.ident_path,
433 server_key = self.node.server_key
437 raise RuntimeError, "Failed instal build sources: %s %s" % (out,err,)
442 scope, options = self._get_ruledef()
443 command = "sudo -S netconfig delete %s %s" % (self.mode, scope)
445 (out,err),proc = server.popen_ssh_command(
447 host = self.node.hostname,
449 user = self.node.slicename,
451 ident_key = self.node.ident_path,
452 server_key = self.node.server_key
456 raise RuntimeError, "Failed instal build sources: %s %s" % (out,err,)
458 self.configured = False
460 def sync_trace(self, local_dir, whichtrace):
461 if whichtrace != 'netpipeStats':
462 raise ValueError, "Unsupported trace %s" % (whichtrace,)
464 local_path = os.path.join(local_dir, "netpipe_stats_%s" % (self.mode,))
466 # create parent local folders
467 proc = subprocess.Popen(
468 ["mkdir", "-p", os.path.dirname(local_path)],
469 stdout = open("/dev/null","w"),
470 stdin = open("/dev/null","r"))
473 raise RuntimeError, "Failed to synchronize trace: %s %s" % (out,err,)
475 (out,err),proc = server.popen_ssh_command(
476 "echo 'Rules:' ; sudo -S netconfig show rules ; echo 'Pipes:' ; sudo -S netconfig show pipes",
477 host = self.node.hostname,
479 user = self.node.slicename,
481 ident_key = self.node.ident_path,
482 server_key = self.node.server_key
486 raise RuntimeError, "Failed to synchronize trace: %s %s" % (out,err,)
488 # dump results to file
489 f = open(local_path, "wb")
496 class TunFilter(object):
498 # tracename : (remotename, localname)
501 def __init__(self, api=None):
503 api = plcapi.PLCAPI()
510 # These get initialised when the filter is connected
511 self.peer_guid = None
512 self.peer_proto = None
513 self.iface_guid = None
517 def _get(what, self):
522 return getattr(wref, what)
526 def _set(what, self, val):
531 setattr(wref, what, val)
533 tun_proto = property(
534 functools.partial(_get, 'tun_proto'),
535 functools.partial(_set, 'tun_proto') )
537 functools.partial(_get, 'tun_addr'),
538 functools.partial(_set, 'tun_addr') )
540 functools.partial(_get, 'tun_port'),
541 functools.partial(_set, 'tun_port') )
543 functools.partial(_get, 'tun_key'),
544 functools.partial(_set, 'tun_key') )
545 tun_cipher = property(
546 functools.partial(_get, 'tun_cipher'),
547 functools.partial(_set, 'tun_cipher') )
552 def remote_trace_path(self, whichtrace):
554 if iface is not None:
555 return iface.remote_trace_path(whichtrace, self._TRACEMAP)
558 def remote_trace_name(self, whichtrace):
560 if iface is not None:
561 return iface.remote_trace_name(whichtrace, self._TRACEMAP)
564 def sync_trace(self, local_dir, whichtrace):
566 if iface is not None:
567 return iface.sync_trace(local_dir, whichtrace, self._TRACEMAP)
570 class ClassQueueFilter(TunFilter):
572 # tracename : (remotename, localname)
573 'dropped_stats' : ('dropped_stats', 'dropped_stats')
576 def __init__(self, api=None):
577 super(ClassQueueFilter, self).__init__(api)
579 self.module = "classqueue.py"
581 class LoggingClassQueueFilter(ClassQueueFilter):
582 _TRACEMAP = ClassQueueFilter._TRACEMAP.copy()
584 # tracename : (remotename, localname)
585 'queue_stats_f' : ('queue_stats_f', 'queue_stats_f'),
586 'queue_stats_b' : ('queue_stats_b', 'queue_stats_b'),
589 def __init__(self, api=None):
590 super(LoggingClassQueueFilter, self).__init__(api)
592 self.module = "loggingclassqueue.py classqueue.py"
596 args = dict(filter(lambda x:len(x)>1, map(lambda x:x.split('=',1),(self._args or "").split(','))))
597 args["outpath"] = "queue_stats"
598 return ",".join(map("=".join, args.iteritems()))
599 def _args_set(self, value):
601 args = property(_args_get, _args_set)
603 class ToSQueueFilter(TunFilter):
604 def __init__(self, api=None):
605 super(ToSQueueFilter, self).__init__(api)
607 self.module = "tosqueue.py"