svc/conntrack.in (kickpeers): Refactor and reformat the search loop.
[tripe] / svc / conntrack.in
1 #! @PYTHON@
2 ### -*-python-*-
3 ###
4 ### Service for automatically tracking network connection status
5 ###
6 ### (c) 2010 Straylight/Edgeware
7 ###
8
9 ###----- Licensing notice ---------------------------------------------------
10 ###
11 ### This file is part of Trivial IP Encryption (TrIPE).
12 ###
13 ### TrIPE is free software: you can redistribute it and/or modify it under
14 ### the terms of the GNU General Public License as published by the Free
15 ### Software Foundation; either version 3 of the License, or (at your
16 ### option) any later version.
17 ###
18 ### TrIPE is distributed in the hope that it will be useful, but WITHOUT
19 ### ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
20 ### FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
21 ### for more details.
22 ###
23 ### You should have received a copy of the GNU General Public License
24 ### along with TrIPE. If not, see <https://www.gnu.org/licenses/>.
25
26 VERSION = '@VERSION@'
27
28 ###--------------------------------------------------------------------------
29 ### External dependencies.
30
31 from ConfigParser import RawConfigParser
32 from optparse import OptionParser
33 import os as OS
34 import sys as SYS
35 import socket as S
36 import mLib as M
37 import tripe as T
38 import dbus as D
39 import re as RX
40 for i in ['mainloop', 'mainloop.glib']:
41 __import__('dbus.%s' % i)
42 try: from gi.repository import GLib as G
43 except ImportError: import gobject as G
44 from struct import pack, unpack
45
46 SM = T.svcmgr
47 ##__import__('rmcr').__debug = True
48
49 ###--------------------------------------------------------------------------
50 ### Utilities.
51
52 class struct (object):
53 """A simple container object."""
54 def __init__(me, **kw):
55 me.__dict__.update(kw)
56
57 ###--------------------------------------------------------------------------
58 ### Address manipulation.
59
60 class InetAddress (object):
61 def __init__(me, addrstr, maskstr = None):
62 me.addr = me._addrstr_to_int(addrstr)
63 if maskstr is None:
64 me.mask = -1
65 elif maskstr.isdigit():
66 me.mask = (1 << 32) - (1 << 32 - int(maskstr))
67 else:
68 me.mask = me._addrstr_to_int(maskstr)
69 if me.addr&~me.mask:
70 raise ValueError('network contains bits set beyond mask')
71 def _addrstr_to_int(me, addrstr):
72 return unpack('>L', S.inet_aton(addrstr))[0]
73 def _int_to_addrstr(me, n):
74 return S.inet_ntoa(pack('>L', n))
75 def sockaddr(me, port = 0):
76 if me.mask != -1: raise ValueError('not a simple address')
77 return me._int_to_addrstr(me.addr), port
78 def __str__(me):
79 addrstr = me._int_to_addrstr(me.addr)
80 if me.mask == -1:
81 return addrstr
82 else:
83 inv = me.mask ^ ((1 << 32) - 1)
84 if (inv&(inv + 1)) == 0:
85 return '%s/%d' % (addrstr, 32 - inv.bit_length())
86 else:
87 return '%s/%s' % (addrstr, me._int_to_addrstr(me.mask))
88 def withinp(me, net):
89 if (me.mask&net.mask) != net.mask: return False
90 if (me.addr ^ net.addr)&net.mask: return False
91 return True
92 def eq(me, other):
93 if me.mask != other.mask: return False
94 if me.addr != other.addr: return False
95 return True
96 @classmethod
97 def from_sockaddr(cls, sa):
98 addr, port = (lambda a, p: (a, p))(*sa)
99 return cls(addr), port
100
101 def parse_address(addrstr, maskstr = None):
102 return InetAddress(addrstr, maskstr)
103
104 def parse_net(netstr):
105 try: sl = netstr.index('/')
106 except ValueError: raise ValueError('missing mask')
107 return parse_address(netstr[:sl], netstr[sl + 1:])
108
109 def straddr(a): return a is None and '#<none>' or str(a)
110
111 ###--------------------------------------------------------------------------
112 ### Parse the configuration file.
113
114 ## Hmm. Should I try to integrate this with the peers database? It's not a
115 ## good fit; it'd need special hacks in tripe-newpeers. And the use case for
116 ## this service are largely going to be satellite notes, I don't think
117 ## scalability's going to be a problem.
118
119 TESTADDR = InetAddress('1.2.3.4')
120
121 CONFSYNTAX = [
122 ('COMMENT', RX.compile(r'^\s*($|[;#])')),
123 ('GRPHDR', RX.compile(r'^\s*\[(.*)\]\s*$')),
124 ('ASSGN', RX.compile(r'\s*([\w.-]+)\s*[:=]\s*(|\S|\S.*\S)\s*$'))]
125
126 class ConfigError (Exception):
127 def __init__(me, file, lno, msg):
128 me.file = file
129 me.lno = lno
130 me.msg = msg
131 def __str__(me):
132 return '%s:%d: %s' % (me.file, me.lno, me.msg)
133
134 class Config (object):
135 """
136 Represents a configuration file.
137
138 The most interesting thing is probably the `groups' slot, which stores a
139 list of pairs (NAME, PATTERNS); the NAME is a string, and the PATTERNS a
140 list of (TAG, PEER, NET) triples. The implication is that there should be
141 precisely one peer from the set, and that it should be named TAG, where
142 (TAG, PEER, NET) is the first triple such that the host's primary IP
143 address (if PEER is None -- or the IP address it would use for
144 communicating with PEER) is within the NET.
145 """
146
147 def __init__(me, file):
148 """
149 Construct a new Config object, reading the given FILE.
150 """
151 me._file = file
152 me._fwatch = M.FWatch(file)
153 me._update()
154
155 def check(me):
156 """
157 See whether the configuration file has been updated.
158 """
159 if me._fwatch.update():
160 me._update()
161
162 def _update(me):
163 """
164 Internal function to update the configuration from the underlying file.
165 """
166
167 if T._debug: print '# reread config'
168
169 ## Initial state.
170 testaddr = None
171 groups = {}
172 grpname = None
173 grplist = []
174
175 ## Open the file and start reading.
176 with open(me._file) as f:
177 lno = 0
178 for line in f:
179 lno += 1
180 for tag, rx in CONFSYNTAX:
181 m = rx.match(line)
182 if m: break
183 else:
184 raise ConfigError(me._file, lno, 'failed to parse line: %r' % line)
185
186 if tag == 'COMMENT':
187 ## A comment. Ignore it and hope it goes away.
188
189 continue
190
191 elif tag == 'GRPHDR':
192 ## A group header. Flush the old group and start a new one.
193 newname = m.group(1)
194
195 if grpname is not None: groups[grpname] = grplist
196 if newname in groups:
197 raise ConfigError(me._file, lno,
198 "duplicate group name `%s'" % newname)
199 grpname = newname
200 grplist = []
201
202 elif tag == 'ASSGN':
203 ## An assignment. Deal with it.
204 name, value = m.group(1), m.group(2)
205
206 if grpname is None:
207 ## We're outside of any group, so this is a global configuration
208 ## tweak.
209
210 if name == 'test-addr':
211 for astr in value.split():
212 try:
213 a = parse_address(astr)
214 except Exception, e:
215 raise ConfigError(me._file, lno,
216 "invalid IP address `%s': %s" %
217 (astr, e))
218 if testaddr is not None:
219 raise ConfigError(me._file, lno, 'duplicate test-address')
220 testaddr = a
221 else:
222 raise ConfigError(me._file, lno,
223 "unknown global option `%s'" % name)
224
225 else:
226 ## Parse a pattern and add it to the group.
227 spec = value.split()
228 i = 0
229
230 ## Check for an explicit target address.
231 if i >= len(spec) or spec[i].find('/') >= 0:
232 peer = None
233 else:
234 try:
235 peer = parse_address(spec[i])
236 except Exception, e:
237 raise ConfigError(me._file, lno,
238 "invalid IP address `%s': %s" %
239 (spec[i], e))
240 i += 1
241
242 ## Parse the local network.
243 if len(spec) != i + 1:
244 raise ConfigError(me._file, lno, 'no network defined')
245 try:
246 net = parse_net(spec[i])
247 except Exception, e:
248 raise ConfigError(me._file, lno,
249 "invalid IP network `%s': %s" %
250 (spec[i], e))
251
252 ## Add this entry to the list.
253 grplist.append((name, peer, net))
254
255 ## Fill in the default test address if necessary.
256 if testaddr is None: testaddr = TESTADDR
257
258 ## Done.
259 if grpname is not None: groups[grpname] = grplist
260 me.testaddr = testaddr
261 me.groups = groups
262
263 ### This will be a configuration file.
264 CF = None
265
266 def cmd_showconfig():
267 T.svcinfo('test-addr=%s' % CF.testaddr)
268 def cmd_showgroups():
269 for g in sorted(CF.groups.iterkeys()):
270 T.svcinfo(g)
271 def cmd_showgroup(g):
272 try: pats = CF.groups[g]
273 except KeyError: raise T.TripeJobError('unknown-group', g)
274 for t, p, n in pats:
275 T.svcinfo('peer', t,
276 'target', p and str(p) or '(default)',
277 'net', str(n))
278
279 ###--------------------------------------------------------------------------
280 ### Responding to a network up/down event.
281
282 def localaddr(peer):
283 """
284 Return the local IP address used for talking to PEER.
285 """
286 sk = S.socket(S.AF_INET, S.SOCK_DGRAM)
287 try:
288 try:
289 sk.connect(peer.sockaddr(1))
290 addr = sk.getsockname()
291 return InetAddress.from_sockaddr(addr)[0]
292 except S.error:
293 return None
294 finally:
295 sk.close()
296
297 _kick = T.Queue()
298 _delay = None
299
300 def cancel_delay():
301 global _delay
302 if _delay is not None:
303 if T._debug: print '# cancel delayed kick'
304 G.source_remove(_delay)
305 _delay = None
306
307 def netupdown(upness, reason):
308 """
309 Add or kill peers according to whether the network is up or down.
310
311 UPNESS is true if the network is up, or false if it's down.
312 """
313
314 _kick.put((upness, reason))
315
316 def delay_netupdown(upness, reason):
317 global _delay
318 cancel_delay()
319 def _func():
320 global _delay
321 if T._debug: print '# delayed %s: %s' % (upness, reason)
322 _delay = None
323 netupdown(upness, reason)
324 return False
325 if T._debug: print '# delaying %s: %s' % (upness, reason)
326 _delay = G.timeout_add(2000, _func)
327
328 def kickpeers():
329 while True:
330 upness, reason = _kick.get()
331 if T._debug: print '# kickpeers %s: %s' % (upness, reason)
332 select = []
333 cancel_delay()
334
335 ## Make sure the configuration file is up-to-date. Don't worry if we
336 ## can't do anything useful.
337 try:
338 CF.check()
339 except Exception, exc:
340 SM.warn('conntrack', 'config-file-error',
341 exc.__class__.__name__, str(exc))
342
343 ## Find the current list of peers.
344 peers = SM.list()
345
346 ## Work out the primary IP address.
347 if upness:
348 addr = localaddr(CF.testaddr)
349 if addr is None:
350 upness = False
351 else:
352 addr = None
353 if not T._debug: pass
354 elif addr: print '# local address = %s' % straddr(addr)
355 else: print '# offline'
356
357 ## Now decide what to do.
358 changes = []
359 for g, pp in CF.groups.iteritems():
360 if T._debug: print '# check group %s' % g
361
362 ## Find out which peer in the group ought to be active.
363 statemap = {}
364 want = None
365 matchp = False
366 for t, p, n in pp:
367 if p is None or not upness: ip = addr
368 else: ip = localaddr(p)
369 if T._debug:
370 info = 'peer=%s; target=%s; net=%s; local=%s' % (
371 t, p or '(default)', n, straddr(ip))
372 if upness and not matchp and \
373 ip is not None and ip.withinp(n):
374 if T._debug: print '# %s: SELECTED' % info
375 statemap[t] = 'up'
376 select.append('%s=%s' % (g, t))
377 if t == 'down' or t.startswith('down/'): want = None
378 else: want = t
379 matchp = True
380 else:
381 statemap[t] = 'down'
382 if T._debug: print '# %s: skipped' % info
383
384 ## Shut down the wrong ones.
385 found = False
386 if T._debug: print '# peer-map = %r' % statemap
387 for p in peers:
388 what = statemap.get(p, 'leave')
389 if what == 'up':
390 found = True
391 if T._debug: print '# peer %s: already up' % p
392 elif what == 'down':
393 def _(p = p):
394 try:
395 SM.kill(p)
396 except T.TripeError, exc:
397 if exc.args[0] == 'unknown-peer':
398 ## Inherently racy; don't worry about this.
399 pass
400 else:
401 raise
402 if T._debug: print '# peer %s: bring down' % p
403 changes.append(_)
404
405 ## Start the right one if necessary.
406 if want is not None and not found:
407 def _(want = want):
408 try:
409 list(SM.svcsubmit('connect', 'active', want))
410 except T.TripeError, exc:
411 SM.warn('conntrack', 'connect-failed', want, *exc.args)
412 if T._debug: print '# peer %s: bring up' % want
413 changes.append(_)
414
415 ## Commit the changes.
416 if changes:
417 SM.notify('conntrack', upness and 'up' or 'down', *select + reason)
418 for c in changes: c()
419
420 ###--------------------------------------------------------------------------
421 ### NetworkManager monitor.
422
423 DBPROPS_IFACE = 'org.freedesktop.DBus.Properties'
424
425 NM_NAME = 'org.freedesktop.NetworkManager'
426 NM_PATH = '/org/freedesktop/NetworkManager'
427 NM_IFACE = NM_NAME
428 NMCA_IFACE = NM_NAME + '.Connection.Active'
429
430 NM_STATE_CONNECTED = 3 #obsolete
431 NM_STATE_CONNECTED_LOCAL = 50
432 NM_STATE_CONNECTED_SITE = 60
433 NM_STATE_CONNECTED_GLOBAL = 70
434 NM_CONNSTATES = set([NM_STATE_CONNECTED,
435 NM_STATE_CONNECTED_LOCAL,
436 NM_STATE_CONNECTED_SITE,
437 NM_STATE_CONNECTED_GLOBAL])
438
439 class NetworkManagerMonitor (object):
440 """
441 Watch NetworkManager signals for changes in network state.
442 """
443
444 ## Strategy. There are two kinds of interesting state transitions for us.
445 ## The first one is the global are-we-connected state, which we'll use to
446 ## toggle network upness on a global level. The second is which connection
447 ## has the default route, which we'll use to tweak which peer in the peer
448 ## group is active. The former is most easily tracked using the signal
449 ## org.freedesktop.NetworkManager.StateChanged; for the latter, we track
450 ## org.freedesktop.NetworkManager.Connection.Active.PropertiesChanged and
451 ## look for when a new connection gains the default route.
452
453 def attach(me, bus):
454 try:
455 nm = bus.get_object(NM_NAME, NM_PATH)
456 state = nm.Get(NM_IFACE, 'State', dbus_interface = DBPROPS_IFACE)
457 if state in NM_CONNSTATES:
458 netupdown(True, ['nm', 'initially-connected'])
459 else:
460 netupdown(False, ['nm', 'initially-disconnected'])
461 except D.DBusException, e:
462 if T._debug: print '# exception attaching to network-manager: %s' % e
463 bus.add_signal_receiver(me._nm_state, 'StateChanged',
464 NM_IFACE, NM_NAME, NM_PATH)
465 bus.add_signal_receiver(me._nm_connchange, 'PropertiesChanged',
466 NMCA_IFACE, NM_NAME, None)
467
468 def _nm_state(me, state):
469 if state in NM_CONNSTATES:
470 delay_netupdown(True, ['nm', 'connected'])
471 else:
472 delay_netupdown(False, ['nm', 'disconnected'])
473
474 def _nm_connchange(me, props):
475 if props.get('Default', False) or props.get('Default6', False):
476 delay_netupdown(True, ['nm', 'default-connection-change'])
477
478 ##--------------------------------------------------------------------------
479 ### Connman monitor.
480
481 CM_NAME = 'net.connman'
482 CM_PATH = '/'
483 CM_IFACE = 'net.connman.Manager'
484
485 class ConnManMonitor (object):
486 """
487 Watch ConnMan signls for changes in network state.
488 """
489
490 ## Strategy. Everything seems to be usefully encoded in the `State'
491 ## property. If it's `offline', `idle' or `ready' then we don't expect a
492 ## network connection. During handover from one network to another, the
493 ## property passes through `ready' to `online'.
494
495 def attach(me, bus):
496 try:
497 cm = bus.get_object(CM_NAME, CM_PATH)
498 props = cm.GetProperties(dbus_interface = CM_IFACE)
499 state = props['State']
500 netupdown(state == 'online', ['connman', 'initially-%s' % state])
501 except D.DBusException, e:
502 if T._debug: print '# exception attaching to connman: %s' % e
503 bus.add_signal_receiver(me._cm_state, 'PropertyChanged',
504 CM_IFACE, CM_NAME, CM_PATH)
505
506 def _cm_state(me, prop, value):
507 if prop != 'State': return
508 delay_netupdown(value == 'online', ['connman', value])
509
510 ###--------------------------------------------------------------------------
511 ### Maemo monitor.
512
513 ICD_NAME = 'com.nokia.icd'
514 ICD_PATH = '/com/nokia/icd'
515 ICD_IFACE = ICD_NAME
516
517 class MaemoICdMonitor (object):
518 """
519 Watch ICd signals for changes in network state.
520 """
521
522 ## Strategy. ICd only handles one connection at a time in steady state,
523 ## though when switching between connections, it tries to bring the new one
524 ## up before shutting down the old one. This makes life a bit easier than
525 ## it is with NetworkManager. On the other hand, the notifications are
526 ## relative to particular connections only, and the indicator that the old
527 ## connection is down (`IDLE') comes /after/ the new one comes up
528 ## (`CONNECTED'), so we have to remember which one is active.
529
530 def attach(me, bus):
531 try:
532 icd = bus.get_object(ICD_NAME, ICD_PATH)
533 try:
534 iap = icd.get_ipinfo(dbus_interface = ICD_IFACE)[0]
535 me._iap = iap
536 netupdown(True, ['icd', 'initially-connected', iap])
537 except D.DBusException:
538 me._iap = None
539 netupdown(False, ['icd', 'initially-disconnected'])
540 except D.DBusException, e:
541 if T._debug: print '# exception attaching to icd: %s' % e
542 me._iap = None
543 bus.add_signal_receiver(me._icd_state, 'status_changed', ICD_IFACE,
544 ICD_NAME, ICD_PATH)
545
546 def _icd_state(me, iap, ty, state, hunoz):
547 if state == 'CONNECTED':
548 me._iap = iap
549 delay_netupdown(True, ['icd', 'connected', iap])
550 elif state == 'IDLE' and iap == me._iap:
551 me._iap = None
552 delay_netupdown(False, ['icd', 'idle'])
553
554 ###--------------------------------------------------------------------------
555 ### D-Bus connection tracking.
556
557 class DBusMonitor (object):
558 """
559 Maintains a connection to the system D-Bus, and watches for signals.
560
561 If the connection is initially down, or drops for some reason, we retry
562 periodically (every five seconds at the moment). If the connection
563 resurfaces, we reattach the monitors.
564 """
565
566 def __init__(me):
567 """
568 Initialise the object and try to establish a connection to the bus.
569 """
570 me._mons = []
571 me._loop = D.mainloop.glib.DBusGMainLoop()
572 me._state = 'startup'
573 me._reconnect()
574
575 def addmon(me, mon):
576 """
577 Add a monitor object to watch for signals.
578
579 MON.attach(BUS) is called, with BUS being the connection to the system
580 bus. MON should query its service's current status and watch for
581 relevant signals.
582 """
583 me._mons.append(mon)
584 if me._bus is not None:
585 mon.attach(me._bus)
586
587 def _reconnect(me, hunoz = None):
588 """
589 Start connecting to the bus.
590
591 If we fail the first time, retry periodically.
592 """
593 if me._state == 'startup':
594 T.aside(SM.notify, 'conntrack', 'dbus-connection', 'startup')
595 elif me._state == 'connected':
596 T.aside(SM.notify, 'conntrack', 'dbus-connection', 'lost')
597 else:
598 T.aside(SM.notify, 'conntrack', 'dbus-connection',
599 'state=%s' % me._state)
600 me._state == 'reconnecting'
601 me._bus = None
602 if me._try_connect():
603 G.timeout_add_seconds(5, me._try_connect)
604
605 def _try_connect(me):
606 """
607 Actually make a connection attempt.
608
609 If we succeed, attach the monitors.
610 """
611 try:
612 addr = OS.getenv('TRIPE_CONNTRACK_BUS')
613 if addr == 'SESSION':
614 bus = D.SessionBus(mainloop = me._loop, private = True)
615 elif addr is not None:
616 bus = D.bus.BusConnection(addr, mainloop = me._loop)
617 else:
618 bus = D.SystemBus(mainloop = me._loop, private = True)
619 for m in me._mons:
620 m.attach(bus)
621 except D.DBusException, e:
622 return True
623 me._bus = bus
624 me._state = 'connected'
625 bus.call_on_disconnection(me._reconnect)
626 T.aside(SM.notify, 'conntrack', 'dbus-connection', 'connected')
627 return False
628
629 ###--------------------------------------------------------------------------
630 ### TrIPE service.
631
632 class GIOWatcher (object):
633 """
634 Monitor I/O events using glib.
635 """
636 def __init__(me, conn, mc = G.main_context_default()):
637 me._conn = conn
638 me._watch = None
639 me._mc = mc
640 def connected(me, sock):
641 me._watch = G.io_add_watch(sock, G.IO_IN,
642 lambda *hunoz: me._conn.receive())
643 def disconnected(me):
644 G.source_remove(me._watch)
645 me._watch = None
646 def iterate(me):
647 me._mc.iteration(True)
648
649 SM.iowatch = GIOWatcher(SM)
650
651 def init():
652 """
653 Service initialization.
654
655 Add the D-Bus monitor here, because we might send commands off immediately,
656 and we want to make sure the server connection is up.
657 """
658 global DBM
659 T.Coroutine(kickpeers, name = 'kickpeers').switch()
660 DBM = DBusMonitor()
661 DBM.addmon(NetworkManagerMonitor())
662 DBM.addmon(ConnManMonitor())
663 DBM.addmon(MaemoICdMonitor())
664 G.timeout_add_seconds(30, lambda: (_delay is not None or
665 netupdown(True, ['interval-timer']) or
666 True))
667
668 def parse_options():
669 """
670 Parse the command-line options.
671
672 Automatically changes directory to the requested configdir, and turns on
673 debugging. Returns the options object.
674 """
675 op = OptionParser(usage = '%prog [-a FILE] [-d DIR]',
676 version = '%%prog %s' % VERSION)
677
678 op.add_option('-a', '--admin-socket',
679 metavar = 'FILE', dest = 'tripesock', default = T.tripesock,
680 help = 'Select socket to connect to [default %default]')
681 op.add_option('-d', '--directory',
682 metavar = 'DIR', dest = 'dir', default = T.configdir,
683 help = 'Select current diretory [default %default]')
684 op.add_option('-c', '--config',
685 metavar = 'FILE', dest = 'conf', default = 'conntrack.conf',
686 help = 'Select configuration [default %default]')
687 op.add_option('--daemon', dest = 'daemon',
688 default = False, action = 'store_true',
689 help = 'Become a daemon after successful initialization')
690 op.add_option('--debug', dest = 'debug',
691 default = False, action = 'store_true',
692 help = 'Emit debugging trace information')
693 op.add_option('--startup', dest = 'startup',
694 default = False, action = 'store_true',
695 help = 'Being called as part of the server startup')
696
697 opts, args = op.parse_args()
698 if args: op.error('no arguments permitted')
699 OS.chdir(opts.dir)
700 T._debug = opts.debug
701 return opts
702
703 ## Service table, for running manually.
704 def cmd_updown(upness):
705 return lambda *args: T.defer(netupdown, upness, ['manual'] + list(args))
706 service_info = [('conntrack', VERSION, {
707 'up': (0, None, '', cmd_updown(True)),
708 'down': (0, None, '', cmd_updown(False)),
709 'show-config': (0, 0, '', cmd_showconfig),
710 'show-groups': (0, 0, '', cmd_showgroups),
711 'show-group': (1, 1, 'GROUP', cmd_showgroup)
712 })]
713
714 if __name__ == '__main__':
715 opts = parse_options()
716 CF = Config(opts.conf)
717 T.runservices(opts.tripesock, service_info,
718 init = init, daemon = opts.daemon)
719
720 ###----- That's all, folks --------------------------------------------------