A series of synchronization fixes:
[nepi.git] / src / nepi / testbeds / planetlab / interfaces.py
1 #!/usr/bin/env python
2 # -*- coding: utf-8 -*-
3
4 from constants import TESTBED_ID
5 import nepi.util.ipaddr2 as ipaddr2
6 import nepi.util.server as server
7 import plcapi
8 import subprocess
9 import os
10 import os.path
11 import random
12 import ipaddr
13
14 import tunproto
15
16 class NodeIface(object):
17     def __init__(self, api=None):
18         if not api:
19             api = plcapi.PLCAPI()
20         self._api = api
21         
22         # Attributes
23         self.primary = True
24
25         # These get initialized at configuration time
26         self.address = None
27         self.lladdr = None
28         self.netprefix = None
29         self.netmask = None
30         self.broadcast = True
31         self._interface_id = None
32
33         # These get initialized when the iface is connected to its node
34         self.node = None
35
36         # These get initialized when the iface is connected to the internet
37         self.has_internet = False
38
39     def __str__(self):
40         return "%s<ip:%s/%s up mac:%s>" % (
41             self.__class__.__name__,
42             self.address, self.netmask,
43             self.lladdr,
44         )
45     
46     __repr__ = __str__
47
48     def add_address(self, address, netprefix, broadcast):
49         raise RuntimeError, "Cannot add explicit addresses to public interface"
50     
51     def pick_iface(self, siblings):
52         """
53         Picks an interface using the PLCAPI to query information about the node.
54         
55         Needs an assigned node.
56         
57         Params:
58             siblings: other NodeIface elements attached to the same node
59         """
60         
61         if self.node is None or self.node._node_id is None:
62             raise RuntimeError, "Cannot pick interface without an assigned node"
63         
64         avail = self._api.GetInterfaces(
65             node_id=self.node._node_id, 
66             is_primary=self.primary,
67             fields=('interface_id','mac','netmask','ip') )
68         
69         used = set([sibling._interface_id for sibling in siblings
70                     if sibling._interface_id is not None])
71         
72         for candidate in avail:
73             candidate_id = candidate['interface_id']
74             if candidate_id not in used:
75                 # pick it!
76                 self._interface_id = candidate_id
77                 self.address = candidate['ip']
78                 self.lladdr = candidate['mac']
79                 self.netprefix = candidate['netmask']
80                 self.netmask = ipaddr2.ipv4_dot2mask(self.netprefix) if self.netprefix else None
81                 return
82         else:
83             raise RuntimeError, "Cannot configure interface: cannot find suitable interface in PlanetLab node"
84
85     def validate(self):
86         if not self.has_internet:
87             raise RuntimeError, "All external interface devices must be connected to the Internet"
88     
89
90 class _CrossIface(object):
91     def __init__(self, proto, addr, port):
92         self.tun_proto = proto
93         self.tun_addr = addr
94         self.tun_port = port
95         
96         # Cannot access cross peers
97         self.peer_proto_impl = None
98     
99     def __str__(self):
100         return "%s%r" % (
101             self.__class__.__name__,
102             ( self.tun_proto,
103               self.tun_addr,
104               self.tun_port ) 
105         )
106     
107     __repr__ = __str__
108
109 class TunIface(object):
110     _PROTO_MAP = tunproto.TUN_PROTO_MAP
111     _KIND = 'TUN'
112
113     def __init__(self, api=None):
114         if not api:
115             api = plcapi.PLCAPI()
116         self._api = api
117         
118         # Attributes
119         self.address = None
120         self.netprefix = None
121         self.netmask = None
122         
123         self.up = None
124         self.device_name = None
125         self.mtu = None
126         self.snat = False
127         self.txqueuelen = None
128         self.pointopoint = None
129         
130         # Enabled traces
131         self.capture = False
132
133         # These get initialized when the iface is connected to its node
134         self.node = None
135         
136         # These get initialized when the iface is configured
137         self.external_iface = None
138         
139         # These get initialized when the iface is configured
140         # They're part of the TUN standard attribute set
141         self.tun_port = None
142         self.tun_addr = None
143         
144         # These get initialized when the iface is connected to its peer
145         self.peer_iface = None
146         self.peer_proto = None
147         self.peer_addr = None
148         self.peer_port = None
149         self.peer_proto_impl = None
150
151         # same as peer proto, but for execute-time standard attribute lookups
152         self.tun_proto = None 
153         
154         
155         # Generate an initial random cryptographic key to use for tunnelling
156         # Upon connection, both endpoints will agree on a common one based on
157         # this one.
158         self.tun_key = ( ''.join(map(chr, [ 
159                     r.getrandbits(8) 
160                     for i in xrange(32) 
161                     for r in (random.SystemRandom(),) ])
162                 ).encode("base64").strip() )        
163         
164
165     def __str__(self):
166         return "%s<ip:%s/%s %s%s%s>" % (
167             self.__class__.__name__,
168             self.address, self.netprefix,
169             " up" if self.up else " down",
170             " snat" if self.snat else "",
171             (" p2p %s" % (self.pointopoint,)) if self.pointopoint else "",
172         )
173     
174     __repr__ = __str__
175     
176     @property
177     def if_name(self):
178         if self.peer_proto_impl:
179             return self.peer_proto_impl.if_name
180
181     def routes_here(self, route):
182         """
183         Returns True if the route should be attached to this interface
184         (ie, it references a gateway in this interface's network segment)
185         """
186         if self.address and self.netprefix:
187             addr, prefix = self.address, self.netprefix
188             if self.pointopoint:
189                 prefix = 32
190                 
191             dest, destprefix, nexthop = route
192             
193             myNet = ipaddr.IPNetwork("%s/%d" % (addr, prefix))
194             gwIp = ipaddr.IPNetwork(nexthop)
195             
196             if self.pointopoint:
197                 peerIp = ipaddr.IPNetwork(self.pointopoint)
198                 
199                 if gwIp == peerIp:
200                     return True
201             else:
202                 if gwIp in myNet:
203                     return True
204         return False
205     
206     def add_address(self, address, netprefix, broadcast):
207         if (self.address or self.netprefix or self.netmask) is not None:
208             raise RuntimeError, "Cannot add more than one address to %s interfaces" % (self._KIND,)
209         if broadcast:
210             raise ValueError, "%s interfaces cannot broadcast in PlanetLab" % (self._KIND,)
211         
212         self.address = address
213         self.netprefix = netprefix
214         self.netmask = ipaddr2.ipv4_mask2dot(netprefix)
215     
216     def validate(self):
217         if not self.node:
218             raise RuntimeError, "Unconnected %s iface - missing node" % (self._KIND,)
219         if self.peer_iface and self.peer_proto not in self._PROTO_MAP:
220             raise RuntimeError, "Unsupported tunnelling protocol: %s" % (self.peer_proto,)
221         if not self.address or not self.netprefix or not self.netmask:
222             raise RuntimeError, "Misconfigured %s iface - missing address" % (self._KIND,)
223     
224     def _impl_instance(self, home_path, listening):
225         impl = self._PROTO_MAP[self.peer_proto](
226             self, self.peer_iface, home_path, self.tun_key, listening)
227         impl.port = self.tun_port
228         return impl
229     
230     def prepare(self, home_path, listening):
231         if not self.peer_iface and (self.peer_proto and (listening or (self.peer_addr and self.peer_port))):
232             # Ad-hoc peer_iface
233             self.peer_iface = _CrossIface(
234                 self.peer_proto,
235                 self.peer_addr,
236                 self.peer_port)
237         if self.peer_iface:
238             if not self.peer_proto_impl:
239                 self.peer_proto_impl = self._impl_instance(home_path, listening)
240             self.peer_proto_impl.prepare()
241     
242     def setup(self):
243         if self.peer_proto_impl:
244             self.peer_proto_impl.setup()
245     
246     def cleanup(self):
247         if self.peer_proto_impl:
248             self.peer_proto_impl.shutdown()
249
250     def destroy(self):
251         if self.peer_proto_impl:
252             self.peer_proto_impl.destroy()
253             self.peer_proto_impl = None
254
255     def async_launch_wait(self):
256         if self.peer_proto_impl:
257             self.peer_proto_impl.async_launch_wait()
258
259     def sync_trace(self, local_dir, whichtrace):
260         if self.peer_proto_impl:
261             return self.peer_proto_impl.sync_trace(local_dir, whichtrace)
262         else:
263             return None
264
265 class TapIface(TunIface):
266     _PROTO_MAP = tunproto.TAP_PROTO_MAP
267     _KIND = 'TAP'
268
269 # Yep, it does nothing - yet
270 class Internet(object):
271     def __init__(self, api=None):
272         if not api:
273             api = plcapi.PLCAPI()
274         self._api = api
275
276 class NetPipe(object):
277     def __init__(self, api=None):
278         if not api:
279             api = plcapi.PLCAPI()
280         self._api = api
281
282         # Attributes
283         self.mode = None
284         self.addrList = None
285         self.portList = None
286         
287         self.plrIn = None
288         self.bwIn = None
289         self.delayIn = None
290
291         self.plrOut = None
292         self.bwOut = None
293         self.delayOut = None
294         
295         # These get initialized when the pipe is connected to its node
296         self.node = None
297         self.configured = False
298     
299     def validate(self):
300         if not self.mode:
301             raise RuntimeError, "Undefined NetPipe mode"
302         if not self.portList:
303             raise RuntimeError, "Undefined NetPipe port list - must always define the scope"
304         if not (self.plrIn or self.bwIn or self.delayIn):
305             raise RuntimeError, "Undefined NetPipe inbound characteristics"
306         if not (self.plrOut or self.bwOut or self.delayOut):
307             raise RuntimeError, "Undefined NetPipe outbound characteristics"
308         if not self.node:
309             raise RuntimeError, "Unconnected NetPipe"
310     
311     def _add_pipedef(self, bw, plr, delay, options):
312         if delay:
313             options.extend(("delay","%dms" % (delay,)))
314         if bw:
315             options.extend(("bw","%.8fMbit/s" % (bw,)))
316         if plr:
317             options.extend(("plr","%.8f" % (plr,)))
318     
319     def _get_ruledef(self):
320         scope = "%s%s%s" % (
321             self.portList,
322             "@" if self.addrList else "",
323             self.addrList or "",
324         )
325         
326         options = []
327         if self.bwIn or self.plrIn or self.delayIn:
328             options.append("IN")
329             self._add_pipedef(self.bwIn, self.plrIn, self.delayIn, options)
330         if self.bwOut or self.plrOut or self.delayOut:
331             options.append("OUT")
332             self._add_pipedef(self.bwOut, self.plrOut, self.delayOut, options)
333         options = ' '.join(options)
334         
335         return (scope,options)
336
337     def configure(self):
338         # set up rule
339         scope, options = self._get_ruledef()
340         command = "sudo -S netconfig config %s %s %s" % (self.mode, scope, options)
341         
342         (out,err),proc = server.popen_ssh_command(
343             command,
344             host = self.node.hostname,
345             port = None,
346             user = self.node.slicename,
347             agent = None,
348             ident_key = self.node.ident_path,
349             server_key = self.node.server_key
350             )
351     
352         if proc.wait():
353             raise RuntimeError, "Failed instal build sources: %s %s" % (out,err,)
354         
355         # we have to clean up afterwards
356         self.configured = True
357     
358     def refresh(self):
359         if self.configured:
360             # refresh rule
361             scope, options = self._get_ruledef()
362             command = "sudo -S netconfig refresh %s %s %s" % (self.mode, scope, options)
363             
364             (out,err),proc = server.popen_ssh_command(
365                 command,
366                 host = self.node.hostname,
367                 port = None,
368                 user = self.node.slicename,
369                 agent = None,
370                 ident_key = self.node.ident_path,
371                 server_key = self.node.server_key
372                 )
373         
374             if proc.wait():
375                 raise RuntimeError, "Failed instal build sources: %s %s" % (out,err,)
376     
377     def cleanup(self):
378         if self.configured:
379             # remove rule
380             scope, options = self._get_ruledef()
381             command = "sudo -S netconfig delete %s %s" % (self.mode, scope)
382             
383             (out,err),proc = server.popen_ssh_command(
384                 command,
385                 host = self.node.hostname,
386                 port = None,
387                 user = self.node.slicename,
388                 agent = None,
389                 ident_key = self.node.ident_path,
390                 server_key = self.node.server_key
391                 )
392         
393             if proc.wait():
394                 raise RuntimeError, "Failed instal build sources: %s %s" % (out,err,)
395             
396             self.configured = False
397     
398     def sync_trace(self, local_dir, whichtrace):
399         if whichtrace != 'netpipeStats':
400             raise ValueError, "Unsupported trace %s" % (whichtrace,)
401         
402         local_path = os.path.join(local_dir, "netpipe_stats_%s" % (self.mode,))
403         
404         # create parent local folders
405         proc = subprocess.Popen(
406             ["mkdir", "-p", os.path.dirname(local_path)],
407             stdout = open("/dev/null","w"),
408             stdin = open("/dev/null","r"))
409
410         if proc.wait():
411             raise RuntimeError, "Failed to synchronize trace: %s %s" % (out,err,)
412         
413         (out,err),proc = server.popen_ssh_command(
414             "echo 'Rules:' ; sudo -S netconfig show rules ; echo 'Pipes:' ; sudo -S netconfig show pipes",
415             host = self.node.hostname,
416             port = None,
417             user = self.node.slicename,
418             agent = None,
419             ident_key = self.node.ident_path,
420             server_key = self.node.server_key
421             )
422         
423         if proc.wait():
424             raise RuntimeError, "Failed to synchronize trace: %s %s" % (out,err,)
425         
426         # dump results to file
427         f = open(local_path, "wb")
428         f.write(err or "")
429         f.write(out or "")
430         f.close()
431         
432         return local_path
433