To get this branch, use:
bzr branch
http://bzr.recompile.se/loggerhead/mandos/trunk
« back to all changes in this revision
Viewing changes to server.py
- browse files at revision 11
- compare with another revision
- compare with revision 13
- download diff
- download tarball
- view history from revision 11
- Committer: Teddy Hogeborn
- Date: 2008-06-30 01:43:39 UTC
- Revision ID: teddy@fukt.bsnet.se-20080630014339-rsuztydpl2w5ml83
* server.py: Rewritten to use Zeroconf (mDNS/DNS-SD) in place of the
old broadcast-UDP-to-port-49001 method.
old broadcast-UDP-to-port-49001 method.
- files modified:
- server.py
added
removed
server.py
19
19
from sets import Set
20
20
import subprocess
21
21
22
import dbus
23
import gobject
24
import avahi
25
from dbus.mainloop.glib import DBusGMainLoop
26
27
# This variable is used to optionally bind to a specified
28
# interface.
29
serviceInterface = avahi.IF_UNSPEC
30
# It is a global variable to fit in with the rest of the
31
# variables from the Avahi server example code:
32
serviceName = "Mandos"
33
serviceType = "_mandos._tcp" # http://www.dns-sd.org/ServiceTypes.html
34
servicePort = None # Not known at startup
35
serviceTXT = [] # TXT record for the service
36
domain = "" # Domain to publish on, default to .local
37
host = "" # Host to publish records for, default to localhost
38
group = None #our entry group
39
rename_count = 12 # Counter so we only rename after collisions a
40
# sensible number of times
41
# End of Avahi example code
42
43
22
44
class Client(object):
23
def __init__(self, name=None, options=None, dn=None,
24
password=None, passfile=None, fqdn=None,
45
"""A representation of a client host served by this server.
46
Attributes:
47
password: string
48
fqdn: string, FQDN (used by the checker)
49
created: datetime.datetime()
50
last_seen: datetime.datetime() or None if not yet seen
51
timeout: datetime.timedelta(); How long from last_seen until
52
this client is invalid
53
interval: datetime.timedelta(); How often to start a new checker
54
timeout_milliseconds: Used by gobject.timeout_add()
55
interval_milliseconds: - '' -
56
stop_hook: If set, called by stop() as stop_hook(self)
57
checker: subprocess.Popen(); a running checker process used
58
to see if the client lives.
59
Is None if no process is running.
60
checker_initiator_tag: a gobject event source tag, or None
61
stop_initiator_tag: - '' -
62
checker_callback_tag: - '' -
63
"""
64
def __init__(self, name=None, options=None, stop_hook=None,
65
dn=None, password=None, passfile=None, fqdn=None,
25
66
timeout=None, interval=-1):
26
67
self.name = name
27
68
self.dn = dn
30
71
elif passfile:
31
72
self.password = open(passfile).readall()
32
73
else:
33
print "No Password or Passfile in client config file"
34
# raise RuntimeError XXX
35
self.password = "gazonk"
74
raise RuntimeError(u"No Password or Passfile for client %s"
75
% self.name)
36
76
self.fqdn = fqdn # string
37
77
self.created = datetime.datetime.now()
38
self.last_seen = None # datetime.datetime()
78
self.last_seen = None
39
79
if timeout is None:
40
80
timeout = options.timeout
41
self.timeout = timeout # datetime.timedelta()
81
self.timeout = timeout
82
self.timeout_milliseconds = ((self.timeout.days
83
* 24 * 60 * 60 * 1000)
84
+ (self.timeout.seconds * 1000)
85
+ (self.timeout.microseconds
86
// 1000))
42
87
if interval == -1:
43
88
interval = options.interval
44
89
else:
45
90
interval = string_to_delta(interval)
46
self.interval = interval # datetime.timedelta()
47
self.next_check = datetime.datetime.now() # datetime.datetime()
48
# Note: next_check may be in the past if checker is not None
49
self.checker = None # or a subprocess.Popen()
50
def check_action(self):
51
"""The checker said something and might have completed.
52
Check if is has, and take appropriate actions."""
53
if self.checker.poll() is None:
54
# False alarm, no result yet
55
#self.checker.read()
56
#print "Checker for %(name)s said nothing?" % vars(self)
57
return
91
self.interval = interval
92
self.interval_milliseconds = ((self.interval.days
93
* 24 * 60 * 60 * 1000)
94
+ (self.interval.seconds * 1000)
95
+ (self.interval.microseconds
96
// 1000))
97
self.stop_hook = stop_hook
98
self.checker = None
99
self.checker_initiator_tag = None
100
self.stop_initiator_tag = None
101
self.checker_callback_tag = None
102
def start(self):
103
"""Start this clients checker and timeout hooks"""
104
# Schedule a new checker to be started an 'interval' from now,
105
# and every interval from then on.
106
self.checker_initiator_tag = gobject.\
107
timeout_add(self.interval_milliseconds,
108
self.start_checker)
109
# Also start a new checker *right now*.
110
self.start_checker()
111
# Schedule a stop() when 'timeout' has passed
112
self.stop_initiator_tag = gobject.\
113
timeout_add(self.timeout_milliseconds,
114
self.stop)
115
def stop(self):
116
"""Stop this client.
117
The possibility that this client might be restarted is left
118
open, but not currently used."""
119
# print "Stopping client", self.name
120
self.password = None
121
if self.stop_initiator_tag:
122
gobject.source_remove(self.stop_initiator_tag)
123
self.stop_initiator_tag = None
124
if self.checker_initiator_tag:
125
gobject.source_remove(self.checker_initiator_tag)
126
self.checker_initiator_tag = None
127
self.stop_checker()
128
if self.stop_hook:
129
self.stop_hook(self)
130
# Do not run this again if called by a gobject.timeout_add
131
return False
132
def __del__(self):
133
# Some code duplication here and in stop()
134
if hasattr(self, "stop_initiator_tag") \
135
and self.stop_initiator_tag:
136
gobject.source_remove(self.stop_initiator_tag)
137
self.stop_initiator_tag = None
138
if hasattr(self, "checker_initiator_tag") \
139
and self.checker_initiator_tag:
140
gobject.source_remove(self.checker_initiator_tag)
141
self.checker_initiator_tag = None
142
self.stop_checker()
143
def checker_callback(self, pid, condition):
144
"""The checker has completed, so take appropriate actions."""
58
145
now = datetime.datetime.now()
59
if self.checker.returncode == 0:
60
print "Checker for %(name)s succeeded" % vars(self)
146
if os.WIFEXITED(condition) \
147
and (os.WEXITSTATUS(condition) == 0):
148
#print "Checker for %(name)s succeeded" % vars(self)
61
149
self.last_seen = now
62
else:
63
print "Checker for %(name)s failed" % vars(self)
64
while self.next_check <= now:
65
self.next_check += self.interval
150
gobject.source_remove(self.stop_initiator_tag)
151
self.stop_initiator_tag = gobject.\
152
timeout_add(self.timeout_milliseconds,
153
self.stop)
154
#else:
155
# if not os.WIFEXITED(condition):
156
# print "Checker for %(name)s crashed?" % vars(self)
157
# else:
158
# print "Checker for %(name)s failed" % vars(self)
66
159
self.checker = None
67
handle_request = check_action
160
self.checker_callback_tag = None
68
161
def start_checker(self):
69
self.stop_checker()
70
try:
71
self.checker = subprocess.Popen("sleep 10; fping -q -- %s"
72
% re.escape(self.fqdn),
73
stdout=subprocess.PIPE,
74
close_fds=True,
75
shell=True, cwd="/")
76
except subprocess.OSError, e:
77
print "Failed to start subprocess:", e
162
"""Start a new checker subprocess if one is not running.
163
If a checker already exists, leave it running and do
164
nothing."""
165
if self.checker is None:
166
#print "Starting checker for", self.name
167
try:
168
self.checker = subprocess.\
169
Popen("sleep 1; fping -q -- %s"
170
% re.escape(self.fqdn),
171
stdout=subprocess.PIPE,
172
close_fds=True, shell=True,
173
cwd="/")
174
self.checker_callback_tag = gobject.\
175
child_watch_add(self.checker.pid,
176
self.\
177
checker_callback)
178
except subprocess.OSError, error:
179
sys.stderr.write(u"Failed to start subprocess: %s\n"
180
% error)
181
# Re-run this periodically if run by gobject.timeout_add
182
return True
78
183
def stop_checker(self):
79
if self.checker is None:
184
"""Force the checker process, if any, to stop."""
185
if not hasattr(self, "checker") or self.checker is None:
80
186
return
187
gobject.source_remove(self.checker_callback_tag)
188
self.checker_callback_tag = None
81
189
os.kill(self.checker.pid, signal.SIGTERM)
82
190
if self.checker.poll() is None:
83
191
os.kill(self.checker.pid, signal.SIGKILL)
84
192
self.checker = None
85
__del__ = stop_checker
86
def fileno(self):
87
if self.checker is None:
88
return None
89
return self.checker.stdout.fileno()
90
def next_stop(self):
91
"""The time when something must be done about this client
92
May be in the past."""
93
if self.last_seen is None:
94
# This client has never been seen
95
next_timeout = self.created + self.timeout
96
else:
97
next_timeout = self.last_seen + self.timeout
98
if self.checker is None:
99
return min(next_timeout, self.next_check)
100
else:
101
return next_timeout
102
193
def still_valid(self, now=None):
103
"""Has this client's timeout not passed?"""
194
"""Has the timeout not yet passed for this client?"""
104
195
if now is None:
105
196
now = datetime.datetime.now()
106
197
if self.last_seen is None:
107
198
return now < (self.created + self.timeout)
108
199
else:
109
200
return now < (self.last_seen + self.timeout)
110
def it_is_time_to_check(self, now=None):
111
if now is None:
112
now = datetime.datetime.now()
113
return self.next_check <= now
114
115
116
class server_metaclass(type):
117
"Common behavior for the UDP and TCP server classes"
118
def __new__(cls, name, bases, attrs):
119
attrs["address_family"] = socket.AF_INET6
120
attrs["allow_reuse_address"] = True
121
def server_bind(self):
122
if self.options.interface:
123
if not hasattr(socket, "SO_BINDTODEVICE"):
124
# From /usr/include/asm-i486/socket.h
125
socket.SO_BINDTODEVICE = 25
126
try:
127
self.socket.setsockopt(socket.SOL_SOCKET,
128
socket.SO_BINDTODEVICE,
129
self.options.interface)
130
except socket.error, error:
131
if error[0] == errno.EPERM:
132
print "Warning: No permission to bind to interface", \
133
self.options.interface
134
else:
135
raise error
136
return super(type(self), self).server_bind()
137
attrs["server_bind"] = server_bind
138
def init(self, *args, **kwargs):
139
if "options" in kwargs:
140
self.options = kwargs["options"]
141
del kwargs["options"]
142
if "clients" in kwargs:
143
self.clients = kwargs["clients"]
144
del kwargs["clients"]
145
if "credentials" in kwargs:
146
self.credentials = kwargs["credentials"]
147
del kwargs["credentials"]
148
return super(type(self), self).__init__(*args, **kwargs)
149
attrs["__init__"] = init
150
return type.__new__(cls, name, bases, attrs)
151
152
153
class udp_handler(SocketServer.DatagramRequestHandler, object):
154
def handle(self):
155
self.wfile.write("Polo")
156
print "UDP request answered"
157
158
159
class IPv6_UDPServer(SocketServer.UDPServer, object):
160
__metaclass__ = server_metaclass
161
def verify_request(self, request, client_address):
162
print "UDP request came"
163
return request[0] == "Marco"
164
201
165
202
166
203
class tcp_handler(SocketServer.BaseRequestHandler, object):
204
"""A TCP request handler class.
205
Instantiated by IPv6_TCPServer for each request to handle it.
206
Note: This will run in its own forked process."""
167
207
def handle(self):
168
print "TCP request came"
169
print "Request:", self.request
170
print "Client Address:", self.client_address
171
print "Server:", self.server
208
#print u"TCP request came"
209
#print u"Request:", self.request
210
#print u"Client Address:", self.client_address
211
#print u"Server:", self.server
172
212
session = gnutls.connection.ServerSession(self.request,
173
self.server.credentials)
174
session.handshake()
175
if session.peer_certificate:
176
print "DN:", session.peer_certificate.subject
213
self.server\
214
.credentials)
215
try:
216
session.handshake()
217
except gnutls.errors.GNUTLSError, error:
218
#sys.stderr.write(u"Handshake failed: %s\n" % error)
219
# Do not run session.bye() here: the session is not
220
# established. Just abandon the request.
221
return
222
#if session.peer_certificate:
223
# print "DN:", session.peer_certificate.subject
177
224
try:
178
225
session.verify_peer()
179
226
except gnutls.errors.CertificateError, error:
180
print "Verify failed", error
227
#sys.stderr.write(u"Verify failed: %s\n" % error)
181
228
session.bye()
182
229
return
183
try:
184
session.send([client.password
185
for client in self.server.clients
186
if (client.dn ==
187
session.peer_certificate.subject)][0])
188
except IndexError:
189
session.send("gazonk")
190
# Log maybe? XXX
230
client = None
231
for c in clients:
232
if c.dn == session.peer_certificate.subject:
233
client = c
234
break
235
# Have to check if client.still_valid(), since it is possible
236
# that the client timed out while establishing the GnuTLS
237
# session.
238
if client and client.still_valid():
239
session.send(client.password)
240
else:
241
#if client:
242
# sys.stderr.write(u"Client %(name)s is invalid\n"
243
# % vars(client))
244
#else:
245
# sys.stderr.write(u"Client not found for DN: %s\n"
246
# % session.peer_certificate.subject)
247
#session.send("gazonk")
248
pass
191
249
session.bye()
192
250
193
251
194
252
class IPv6_TCPServer(SocketServer.ForkingTCPServer, object):
195
__metaclass__ = server_metaclass
253
"""IPv6 TCP server. Accepts 'None' as address and/or port.
254
Attributes:
255
options: Command line options
256
clients: Set() of Client objects
257
credentials: GnuTLS X.509 credentials
258
"""
259
address_family = socket.AF_INET6
260
def __init__(self, *args, **kwargs):
261
if "options" in kwargs:
262
self.options = kwargs["options"]
263
del kwargs["options"]
264
if "clients" in kwargs:
265
self.clients = kwargs["clients"]
266
del kwargs["clients"]
267
if "credentials" in kwargs:
268
self.credentials = kwargs["credentials"]
269
del kwargs["credentials"]
270
return super(type(self), self).__init__(*args, **kwargs)
271
def server_bind(self):
272
"""This overrides the normal server_bind() function
273
to bind to an interface if one was specified, and also NOT to
274
bind to an address or port if they were not specified."""
275
if self.options.interface:
276
if not hasattr(socket, "SO_BINDTODEVICE"):
277
# From /usr/include/asm-i486/socket.h
278
socket.SO_BINDTODEVICE = 25
279
try:
280
self.socket.setsockopt(socket.SOL_SOCKET,
281
socket.SO_BINDTODEVICE,
282
self.options.interface)
283
except socket.error, error:
284
if error[0] == errno.EPERM:
285
sys.stderr.write(u"Warning: No permission to bind to interface %s\n"
286
% self.options.interface)
287
else:
288
raise error
289
# Only bind(2) the socket if we really need to.
290
if self.server_address[0] or self.server_address[1]:
291
if not self.server_address[0]:
292
in6addr_any = "::"
293
self.server_address = (in6addr_any,
294
self.server_address[1])
295
elif self.server_address[1] is None:
296
self.server_address = (self.server_address[0],
297
0)
298
return super(type(self), self).server_bind()
196
299
197
300
198
301
def string_to_delta(interval):
229
332
return delta
230
333
231
334
232
def main():
335
def add_service():
336
"""From the Avahi server example code"""
337
global group, serviceName, serviceType, servicePort, serviceTXT, \
338
domain, host
339
if group is None:
340
group = dbus.Interface(
341
bus.get_object( avahi.DBUS_NAME,
342
server.EntryGroupNew()),
343
avahi.DBUS_INTERFACE_ENTRY_GROUP)
344
group.connect_to_signal('StateChanged',
345
entry_group_state_changed)
346
347
# print "Adding service '%s' of type '%s' ..." % (serviceName,
348
# serviceType)
349
350
group.AddService(
351
serviceInterface, # interface
352
avahi.PROTO_INET6, # protocol
353
dbus.UInt32(0), # flags
354
serviceName, serviceType,
355
domain, host,
356
dbus.UInt16(servicePort),
357
avahi.string_array_to_txt_array(serviceTXT))
358
group.Commit()
359
360
361
def remove_service():
362
"""From the Avahi server example code"""
363
global group
364
365
if not group is None:
366
group.Reset()
367
368
369
def server_state_changed(state):
370
"""From the Avahi server example code"""
371
if state == avahi.SERVER_COLLISION:
372
print "WARNING: Server name collision"
373
remove_service()
374
elif state == avahi.SERVER_RUNNING:
375
add_service()
376
377
378
def entry_group_state_changed(state, error):
379
"""From the Avahi server example code"""
380
global serviceName, server, rename_count
381
382
# print "state change: %i" % state
383
384
if state == avahi.ENTRY_GROUP_ESTABLISHED:
385
pass
386
# print "Service established."
387
elif state == avahi.ENTRY_GROUP_COLLISION:
388
389
rename_count = rename_count - 1
390
if rename_count > 0:
391
name = server.GetAlternativeServiceName(name)
392
print "WARNING: Service name collision, changing name to '%s' ..." % name
393
remove_service()
394
add_service()
395
396
else:
397
print "ERROR: No suitable service name found after %i retries, exiting." % n_rename
398
main_loop.quit()
399
elif state == avahi.ENTRY_GROUP_FAILURE:
400
print "Error in group state changed", error
401
main_loop.quit()
402
return
403
404
405
def if_nametoindex(interface):
406
"""Call the C function if_nametoindex()"""
407
try:
408
if "ctypes" not in sys.modules:
409
import ctypes
410
libc = ctypes.cdll.LoadLibrary("libc.so.6")
411
return libc.if_nametoindex(interface)
412
except (ImportError, OSError, AttributeError):
413
if "struct" not in sys.modules:
414
import struct
415
if "fcntl" not in sys.modules:
416
import fcntl
417
SIOCGIFINDEX = 0x8933 # From /usr/include/linux/sockios.h
418
s = socket.socket()
419
ifreq = fcntl.ioctl(s, SIOCGIFINDEX,
420
struct.pack("16s16x", interface))
421
s.close()
422
interface_index = struct.unpack("I", ifreq[16:20])[0]
423
return interface_index
424
425
426
if __name__ == '__main__':
233
427
parser = OptionParser()
234
428
parser.add_option("-i", "--interface", type="string",
235
default="eth0", metavar="IF",
236
help="Interface to bind to")
429
default=None, metavar="IF",
430
help="Bind to interface IF")
237
431
parser.add_option("--cert", type="string", default="cert.pem",
238
432
metavar="FILE",
239
help="Public key certificate to use")
433
help="Public key certificate PEM file to use")
240
434
parser.add_option("--key", type="string", default="key.pem",
241
435
metavar="FILE",
242
help="Private key to use")
436
help="Private key PEM file to use")
243
437
parser.add_option("--ca", type="string", default="ca.pem",
244
438
metavar="FILE",
245
help="Certificate Authority certificate to use")
439
help="Certificate Authority certificate PEM file to use")
246
440
parser.add_option("--crl", type="string", default="crl.pem",
247
441
metavar="FILE",
248
help="Certificate Revokation List to use")
249
parser.add_option("-p", "--port", type="int", default=49001,
442
help="Certificate Revokation List PEM file to use")
443
parser.add_option("-p", "--port", type="int", default=None,
250
444
help="Port number to receive requests on")
251
parser.add_option("--dh", type="int", metavar="BITS",
252
help="DH group to use")
253
parser.add_option("-t", "--timeout", type="string", # Parsed later
254
default="15m",
445
parser.add_option("--timeout", type="string", # Parsed later
446
default="1h",
255
447
help="Amount of downtime allowed for clients")
256
448
parser.add_option("--interval", type="string", # Parsed later
257
449
default="5m",
259
451
parser.add_option("--check", action="store_true", default=False,
260
452
help="Run self-test")
261
453
(options, args) = parser.parse_args()
262
454
263
455
if options.check:
264
456
import doctest
265
457
doctest.testmod()
270
462
options.timeout = string_to_delta(options.timeout)
271
463
except ValueError:
272
464
parser.error("option --timeout: Unparseable time")
273
274
465
try:
275
466
options.interval = string_to_delta(options.interval)
276
467
except ValueError:
284
475
285
476
# Parse config file
286
477
defaults = {}
287
client_config_object = ConfigParser.SafeConfigParser(defaults)
288
client_config_object.read("mandos-clients.conf")
289
clients = Set(Client(name=section, options=options,
290
**(dict(client_config_object\
291
.items(section))))
292
for section in client_config_object.sections())
293
294
in6addr_any = "::"
295
udp_server = IPv6_UDPServer((in6addr_any, options.port),
296
udp_handler,
297
options=options)
298
299
tcp_server = IPv6_TCPServer((in6addr_any, options.port),
478
client_config = ConfigParser.SafeConfigParser(defaults)
479
#client_config.readfp(open("secrets.conf"), "secrets.conf")
480
client_config.read("mandos-clients.conf")
481
482
# From the Avahi server example code
483
DBusGMainLoop(set_as_default=True )
484
main_loop = gobject.MainLoop()
485
bus = dbus.SystemBus()
486
server = dbus.Interface(
487
bus.get_object( avahi.DBUS_NAME, avahi.DBUS_PATH_SERVER ),
488
avahi.DBUS_INTERFACE_SERVER )
489
# End of Avahi example code
490
491
clients = Set()
492
def remove_from_clients(client):
493
clients.remove(client)
494
if not clients:
495
print "No clients left, exiting"
496
main_loop.quit()
497
498
clients.update(Set(Client(name=section, options=options,
499
stop_hook = remove_from_clients,
500
**(dict(client_config\
501
.items(section))))
502
for section in client_config.sections()))
503
for client in clients:
504
client.start()
505
506
tcp_server = IPv6_TCPServer((None, options.port),
300
507
tcp_handler,
301
508
options=options,
302
509
clients=clients,
303
510
credentials=cred)
304
305
while True:
306
if not clients:
307
break
308
try:
309
next_stop = min(client.next_stop() for client in clients)
310
now = datetime.datetime.now()
311
if next_stop > now:
312
delay = next_stop - now
313
delay_seconds = (delay.days * 24 * 60 * 60
314
+ delay.seconds
315
+ delay.microseconds / 1000000)
316
clients_with_checkers = tuple(client for client in
317
clients
318
if client.checker
319
is not None)
320
input_checks = (udp_server, tcp_server) \
321
+ clients_with_checkers
322
print "Waiting for network",
323
if clients_with_checkers:
324
print "and checkers for:",
325
for client in clients_with_checkers:
326
print client.name,
327
print
328
input, out, err = select.select(input_checks, (), (),
329
delay_seconds)
330
for obj in input:
331
obj.handle_request()
332
# start new checkers
333
for client in clients:
334
if client.it_is_time_to_check(now=now) and \
335
client.checker is None:
336
print "Starting checker for client %(name)s" \
337
% vars(client)
338
client.start_checker()
339
# delete timed-out clients
340
for client in clients.copy():
341
if not client.still_valid(now=now):
342
# log xxx
343
print "Removing client %(name)s" % vars(client)
344
clients.remove(client)
345
except KeyboardInterrupt:
346
break
511
# Find out what random port we got
512
servicePort = tcp_server.socket.getsockname()[1]
513
#sys.stderr.write("Now listening on port %d\n" % servicePort)
514
515
if options.interface is not None:
516
serviceInterface = if_nametoindex(options.interface)
517
518
# From the Avahi server example code
519
server.connect_to_signal("StateChanged", server_state_changed)
520
server_state_changed(server.GetState())
521
# End of Avahi example code
522
523
gobject.io_add_watch(tcp_server.fileno(), gobject.IO_IN,
524
lambda *args, **kwargs:
525
tcp_server.handle_request(*args[2:],
526
**kwargs) or True)
527
try:
528
main_loop.run()
529
except KeyboardInterrupt:
530
print
347
531
348
532
# Cleanup here
533
534
# From the Avahi server example code
535
if not group is None:
536
group.Free()
537
# End of Avahi example code
538
349
539
for client in clients:
350
client.stop_checker()
351
352
353
if __name__ == "__main__":
354
main()
355
540
client.stop_hook = None
541
client.stop()
Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0