Context Navigation

source: fedd/federation/emulab_access.py @ 4692a16

axis_examplecompt_changesinfo-ops

Last change on this file since 4692a16 was 4692a16, checked in by Ted Faber <faber@…>, 14 years ago
Gratuitious use of lambda
Property mode set to `100644`
File size: 39.3 KB

Line
1	#!/usr/local/bin/python
2
3	import os,sys
4	import stat # for chmod constants
5	import re
6	import random
7	import string
8	import copy
9	import pickle
10	import logging
11	import subprocess
12	import traceback
13
14	from threading import *
15	from M2Crypto.SSL import SSLError
16
17	from access import access_base
18	from legacy_access import legacy_access
19
20	from util import *
21	from allocate_project import allocate_project_local, allocate_project_remote
22	from fedid import fedid, generate_fedid
23	from authorizer import authorizer, abac_authorizer
24	from service_error import service_error
25	from remote_service import xmlrpc_handler, soap_handler, service_caller
26
27	import httplib
28	import tempfile
29	from urlparse import urlparse
30
31	import topdl
32	import list_log
33	import proxy_emulab_segment
34	import local_emulab_segment
35
36
37	# Make log messages disappear if noone configures a fedd logger
38	class nullHandler(logging.Handler):
39	def emit(self, record): pass
40
41	fl = logging.getLogger("fedd.access")
42	fl.addHandler(nullHandler())
43
44	class access(access_base, legacy_access):
45	"""
46	The implementation of access control based on mapping users to projects.
47
48	Users can be mapped to existing projects or have projects created
49	dynamically. This implements both direct requests and proxies.
50	"""
51
52	def __init__(self, config=None, auth=None):
53	"""
54	Initializer. Pulls parameters out of the ConfigParser's access section.
55	"""
56
57	access_base.__init__(self, config, auth)
58
59	self.allow_proxy = config.getboolean("access", "allow_proxy")
60
61	self.domain = config.get("access", "domain")
62	self.userconfdir = config.get("access","userconfdir")
63	self.userconfcmd = config.get("access","userconfcmd")
64	self.userconfurl = config.get("access","userconfurl")
65	self.federation_software = config.get("access", "federation_software")
66	self.portal_software = config.get("access", "portal_software")
67	self.local_seer_software = config.get("access", "local_seer_software")
68	self.local_seer_image = config.get("access", "local_seer_image")
69	self.local_seer_start = config.get("access", "local_seer_start")
70	self.seer_master_start = config.get("access", "seer_master_start")
71	self.ssh_privkey_file = config.get("access","ssh_privkey_file")
72	self.ssh_pubkey_file = config.get("access","ssh_pubkey_file")
73	self.ssh_port = config.get("access","ssh_port") or "22"
74	self.boss = config.get("access", "boss")
75	self.ops = config.get("access", "ops")
76	self.xmlrpc_cert = config.get("access", "xmlrpc_cert")
77	self.xmlrpc_certpw = config.get("access", "xmlrpc_certpw")
78
79	self.dragon_endpoint = config.get("access", "dragon")
80	self.dragon_vlans = config.get("access", "dragon_vlans")
81	self.deter_internal = config.get("access", "deter_internal")
82
83	self.tunnel_config = config.getboolean("access", "tunnel_config")
84	self.portal_command = config.get("access", "portal_command")
85	self.portal_image = config.get("access", "portal_image")
86	self.portal_type = config.get("access", "portal_type") or "pc"
87	self.portal_startcommand = config.get("access", "portal_startcommand")
88	self.node_startcommand = config.get("access", "node_startcommand")
89
90	self.federation_software = self.software_list(self.federation_software)
91	self.portal_software = self.software_list(self.portal_software)
92	self.local_seer_software = self.software_list(self.local_seer_software)
93
94	self.access_type = self.access_type.lower()
95	if self.access_type == 'remote_emulab':
96	self.start_segment = proxy_emulab_segment.start_segment
97	self.stop_segment = proxy_emulab_segment.stop_segment
98	elif self.access_type == 'local_emulab':
99	self.start_segment = local_emulab_segment.start_segment
100	self.stop_segment = local_emulab_segment.stop_segment
101	else:
102	self.start_segment = None
103	self.stop_segment = None
104
105	self.restricted = [ ]
106	# XXX: this should go?
107	#if config.has_option("access", "accessdb"):
108	# self.read_access(config.get("access", "accessdb"))
109	tb = config.get('access', 'testbed')
110	if tb: self.testbed = [ t.strip() for t in tb.split(',') ]
111	else: self.testbed = [ ]
112
113	# authorization information
114	self.auth_type = config.get('access', 'auth_type') \
115	or 'legacy'
116	self.auth_dir = config.get('access', 'auth_dir')
117	accessdb = config.get("access", "accessdb")
118	# initialize the authorization system
119	if self.auth_type == 'legacy':
120	self.access = { }
121	if accessdb:
122	self.legacy_read_access(accessdb, self.legacy_access_tuple)
123	elif self.auth_type == 'abac':
124	self.auth = abac_authorizer(load=self.auth_dir)
125	self.access = [ ]
126	if accessdb:
127	self.read_access(accessdb, self.access_tuple)
128	else:
129	raise service_error(service_error.internal,
130	"Unknown auth_type: %s" % self.auth_type)
131
132	# read_state in the base_class
133	self.state_lock.acquire()
134	for a in ('allocation', 'projects', 'keys', 'types'):
135	if a not in self.state:
136	self.state[a] = { }
137	self.allocation = self.state['allocation']
138	self.projects = self.state['projects']
139	self.keys = self.state['keys']
140	self.types = self.state['types']
141	if self.auth_type == "legacy":
142	# Add the ownership attributes to the authorizer. Note that the
143	# indices of the allocation dict are strings, but the attributes are
144	# fedids, so there is a conversion.
145	for k in self.allocation.keys():
146	for o in self.allocation[k].get('owners', []):
147	self.auth.set_attribute(o, fedid(hexstr=k))
148	if self.allocation[k].has_key('userconfig'):
149	sfid = self.allocation[k]['userconfig']
150	fid = fedid(hexstr=sfid)
151	self.auth.set_attribute(fid, "/%s" % sfid)
152	self.state_lock.release()
153	self.exports = {
154	'SMB': self.export_SMB,
155	'seer': self.export_seer,
156	'tmcd': self.export_tmcd,
157	'userconfig': self.export_userconfig,
158	'project_export': self.export_project_export,
159	'local_seer_control': self.export_local_seer,
160	'seer_master': self.export_seer_master,
161	'hide_hosts': self.export_hide_hosts,
162	}
163
164	if not self.local_seer_image or not self.local_seer_software or \
165	not self.local_seer_start:
166	if 'local_seer_control' in self.exports:
167	del self.exports['local_seer_control']
168
169	if not self.local_seer_image or not self.local_seer_software or \
170	not self.seer_master_start:
171	if 'seer_master' in self.exports:
172	del self.exports['seer_master']
173
174
175	self.soap_services = {\
176	'RequestAccess': soap_handler("RequestAccess", self.RequestAccess),
177	'ReleaseAccess': soap_handler("ReleaseAccess", self.ReleaseAccess),
178	'StartSegment': soap_handler("StartSegment", self.StartSegment),
179	'TerminateSegment': soap_handler("TerminateSegment",
180	self.TerminateSegment),
181	}
182	self.xmlrpc_services = {\
183	'RequestAccess': xmlrpc_handler('RequestAccess',
184	self.RequestAccess),
185	'ReleaseAccess': xmlrpc_handler('ReleaseAccess',
186	self.ReleaseAccess),
187	'StartSegment': xmlrpc_handler("StartSegment", self.StartSegment),
188	'TerminateSegment': xmlrpc_handler('TerminateSegment',
189	self.TerminateSegment),
190	}
191
192	self.call_SetValue = service_caller('SetValue')
193	self.call_GetValue = service_caller('GetValue', log=self.log)
194
195	if not config.has_option("allocate", "uri"):
196	self.allocate_project = \
197	allocate_project_local(config, auth)
198	else:
199	self.allocate_project = \
200	allocate_project_remote(config, auth)
201
202
203	# If the project allocator exports services, put them in this object's
204	# maps so that classes that instantiate this can call the services.
205	self.soap_services.update(self.allocate_project.soap_services)
206	self.xmlrpc_services.update(self.allocate_project.xmlrpc_services)
207
208	@staticmethod
209	def legacy_access_tuple(str):
210	"""
211	Convert a string of the form (id[:resources:resouces], id, id) into a
212	tuple of the form (project, user, user) where users may be names or
213	fedids. The resources strings are obsolete and ignored.
214	"""
215	def parse_name(n):
216	if n.startswith('fedid:'): return fedid(hexstr=n[len('fedid:'):])
217	else: return n
218
219	str = str.strip()
220	if str.startswith('(') and str.endswith(')'):
221	str = str[1:-1]
222	names = [ s.strip() for s in str.split(",")]
223	if len(names) > 3:
224	raise self.parse_error("More than three fields in name")
225	first = names[0].split(":")
226	if first == 'fedid:':
227	del first[0]
228	first[0] = fedid(hexstr=first[0])
229	names[0] = first[0]
230
231	for i in range(1,2):
232	names[i] = parse_name(names[i])
233
234	return tuple(names)
235	else:
236	raise self.parse_error('Bad mapping (unbalanced parens)')
237
238	@staticmethod
239	def access_tuple(str):
240	"""
241	Convert a string of the form (id, id) into an access_project. This is
242	called by read_access to convert to local attributes. It returns
243	a tuple of the form (project, user, user) where the two users are
244	always the same.
245	"""
246
247	str = str.strip()
248	if str.startswith('(') and str.endswith(')') and str.count(',') == 1:
249	# The slice takes the parens off the string.
250	proj, user = str[1:-1].split(',')
251	return (proj.strip(), user.strip(), user.strip())
252	else:
253	raise self.parse_error(
254	'Bad mapping (unbalanced parens or more than 1 comma)')
255
256
257	# RequestAccess support routines
258
259	def legacy_lookup_access(self, req, fid):
260	"""
261	Look up the local access control information mapped to this fedid and
262	credentials. In this case it is a (project, create_user, access_user)
263	triple, and a triple of three booleans indicating which, if any will
264	need to be dynamically created. Finally a list of owners for that
265	allocation is returned.
266
267	lookup_access_base pulls the first triple out, and it is parsed by this
268	routine into the boolean map. Owners is always the controlling fedid.
269	"""
270	# Return values
271	rp = None
272	ru = None
273	# This maps a valid user to the Emulab projects and users to use
274	found, match = self.legacy_lookup_access_base(req, fid)
275	tb, project, user = match
276
277	if found == None:
278	raise service_error(service_error.access,
279	"Access denied - cannot map access")
280
281	# resolve <dynamic> and <same> in found
282	dyn_proj = False
283	dyn_create_user = False
284	dyn_service_user = False
285
286	if found[0] == "<same>":
287	if project != None:
288	rp = project
289	else :
290	raise service_error(\
291	service_error.server_config,
292	"Project matched <same> when no project given")
293	elif found[0] == "<dynamic>":
294	rp = None
295	dyn_proj = True
296	else:
297	rp = found[0]
298
299	if found[1] == "<same>":
300	if user_match == "<any>":
301	if user != None: rcu = user[0]
302	else: raise service_error(\
303	service_error.server_config,
304	"Matched <same> on anonymous request")
305	else:
306	rcu = user_match
307	elif found[1] == "<dynamic>":
308	rcu = None
309	dyn_create_user = True
310	else:
311	rcu = found[1]
312
313	if found[2] == "<same>":
314	if user_match == "<any>":
315	if user != None: rsu = user[0]
316	else: raise service_error(\
317	service_error.server_config,
318	"Matched <same> on anonymous request")
319	else:
320	rsu = user_match
321	elif found[2] == "<dynamic>":
322	rsu = None
323	dyn_service_user = True
324	else:
325	rsu = found[2]
326
327	return (rp, rcu, rsu), (dyn_create_user, dyn_service_user, dyn_proj),\
328	[ fid ]
329
330	def lookup_access(self, req, fid, filter=None, compare=None):
331	"""
332	Check all the attributes that this controller knows how to map and see
333	if the requester is allowed to use any of them. If so return one.
334	Filter defined the objects to check - it's a function that returns true
335	for the objects to check - and cmp defines the order to check them in
336	as the cmp field of sorted(). If filter is None, all possibilities are
337	checked. If cmp is None, the choices are sorted by priority.
338	"""
339
340	# Import request credentials into this (clone later??)
341	if self.auth.import_credentials(
342	data_list=req.get('abac_credential', [])):
343	self.auth.save()
344
345	# NB: in the default case (the else), the comparison order is reversed
346	# so numerically larger priorities are checked first.
347	if compare: c = compare
348	else: c = lambda(a, b): cmp(b,a)
349
350	if filter: f = filter
351	else: f = lambda(x): True
352
353	check = sorted([ a for a in self.access if f(a)], cmp=c)
354
355	# Check every attribute that we know how to map and take the first
356	# success.
357	for attr in check:
358	if self.auth.check_attribute(fid, attr.attr):
359	self.log.debug("Access succeeded for %s %s" % (attr.attr, fid))
360	# XXX: needs to deal with dynamics
361	return copy.copy(attr.value), (False, False, False), \
362	[ fid ]
363	else:
364	self.log.debug("Access failed for %s %s" % (attr.attr, fid))
365	else:
366	raise service_error(service_error.access, "Access denied")
367
368
369	def do_project_allocation(self, dyn, project, user):
370	"""
371	Call the project allocation routines and return the info.
372	"""
373	if dyn:
374	# Compose the dynamic project request
375	# (only dynamic, dynamic currently allowed)
376	preq = { 'AllocateProjectRequestBody': \
377	{ 'project' : {\
378	'user': [ \
379	{ \
380	'access': [ {
381	'sshPubkey': self.ssh_pubkey_file } ],
382	'role': "serviceAccess",\
383	}, \
384	{ \
385	'access': [ {
386	'sshPubkey': self.ssh_pubkey_file } ],
387	'role': "experimentCreation",\
388	}, \
389	], \
390	}\
391	}\
392	}
393	return self.allocate_project.dynamic_project(preq)
394	else:
395	preq = {'StaticProjectRequestBody' : \
396	{ 'project': \
397	{ 'name' : { 'localname' : project },\
398	'user' : [ \
399	{\
400	'userID': { 'localname' : user }, \
401	'access': [ {
402	'sshPubkey': self.ssh_pubkey_file } ],
403	'role': 'experimentCreation'\
404	},\
405	{\
406	'userID': { 'localname' : user}, \
407	'access': [ {
408	'sshPubkey': self.ssh_pubkey_file } ],
409	'role': 'serviceAccess'\
410	},\
411	]}\
412	}\
413	}
414	return self.allocate_project.static_project(preq)
415
416	def save_project_state(self, aid, ap, dyn, owners):
417	"""
418	Parse out and save the information relevant to the project created for
419	this experiment. That info is largely in ap and owners. dyn indicates
420	that the project was created dynamically. Return the user and project
421	names.
422	"""
423	self.state_lock.acquire()
424	self.allocation[aid] = { }
425	try:
426	pname = ap['project']['name']['localname']
427	except KeyError:
428	pname = None
429
430	if dyn:
431	if not pname:
432	self.state_lock.release()
433	raise service_error(service_error.internal,
434	"Misformed allocation response?")
435	if pname in self.projects: self.projects[pname] += 1
436	else: self.projects[pname] = 1
437	self.allocation[aid]['project'] = pname
438	else:
439	# sproject is a static project associated with this allocation.
440	self.allocation[aid]['sproject'] = pname
441
442	self.allocation[aid]['keys'] = [ ]
443
444	try:
445	for u in ap['project']['user']:
446	uname = u['userID']['localname']
447	if u['role'] == 'experimentCreation':
448	self.allocation[aid]['user'] = uname
449	for k in [ k['sshPubkey'] for k in u['access'] \
450	if k.has_key('sshPubkey') ]:
451	kv = "%s:%s" % (uname, k)
452	if self.keys.has_key(kv): self.keys[kv] += 1
453	else: self.keys[kv] = 1
454	self.allocation[aid]['keys'].append((uname, k))
455	except KeyError:
456	self.state_lock.release()
457	raise service_error(service_error.internal,
458	"Misformed allocation response?")
459
460	self.allocation[aid]['owners'] = owners
461	self.write_state()
462	self.state_lock.release()
463	return (pname, uname)
464
465	# End of RequestAccess support routines
466
467	def RequestAccess(self, req, fid):
468	"""
469	Handle the access request. Proxy if not for us.
470
471	Parse out the fields and make the allocations or rejections if for us,
472	otherwise, assuming we're willing to proxy, proxy the request out.
473	"""
474
475	def gateway_hardware(h):
476	if h == 'GWTYPE': return self.portal_type or 'GWTYPE'
477	else: return h
478
479	def get_export_project(svcs):
480	"""
481	if the service requests includes one to export a project, return
482	that project.
483	"""
484	rv = None
485	for s in svcs:
486	if s.get('name', '') == 'project_export' and \
487	s.get('visibility', '') == 'export':
488	if not rv:
489	for a in s.get('fedAttr', []):
490	if a.get('attribute', '') == 'project' \
491	and 'value' in a:
492	rv = a['value']
493	else:
494	raise service_error(service_error, access,
495	'Requesting multiple project exports is ' + \
496	'not supported');
497	return rv
498
499	# The dance to get into the request body
500	if req.has_key('RequestAccessRequestBody'):
501	req = req['RequestAccessRequestBody']
502	else:
503	raise service_error(service_error.req, "No request!?")
504
505	# if this includes a project export request, construct a filter such
506	# that only the ABAC attributes mapped to that project are checked for
507	# access.
508	if 'service' in req:
509	ep = get_export_project(req['service'])
510	pf = lambda(a): a.value[0] == ep
511	else:
512	ep = None
513	pf = None
514
515	if self.auth.import_credentials(
516	data_list=req.get('abac_credential', [])):
517	self.auth.save()
518
519	if self.auth_type == "legacy":
520	found, dyn, owners = self.legacy_lookup_access(req, fid)
521	elif self.auth_type == 'abac':
522	found, dyn, owners = self.lookup_access(req, fid, filter=pf)
523	else:
524	raise service_error(service_error.internal,
525	'Unknown auth_type: %s' % self.auth_type)
526	ap = None
527
528	# This only happens in legacy lookups, but if this user has access to
529	# the testbed but not the project to be exported, raise the error.
530	if ep and ep != found[0]:
531	raise service_error(service_error.access,
532	"Cannot export %s" % ep)
533
534	if self.ssh_pubkey_file:
535	ap = self.do_project_allocation(dyn[1], found[0], found[1])
536	else:
537	raise service_error(service_error.internal,
538	"SSH access parameters required")
539	# keep track of what's been added
540	allocID, alloc_cert = generate_fedid(subj="alloc", log=self.log)
541	aid = unicode(allocID)
542
543	pname, uname = self.save_project_state(aid, ap, dyn[1], owners)
544
545	services, svc_state = self.export_services(req.get('service',[]),
546	pname, uname)
547	self.state_lock.acquire()
548	# Store services state in global state
549	for k, v in svc_state.items():
550	self.allocation[aid][k] = v
551	self.write_state()
552	self.state_lock.release()
553	# Give the owners the right to change this allocation
554	for o in owners:
555	self.auth.set_attribute(o, allocID)
556	self.auth.save()
557	try:
558	f = open("%s/%s.pem" % (self.certdir, aid), "w")
559	print >>f, alloc_cert
560	f.close()
561	except EnvironmentError, e:
562	raise service_error(service_error.internal,
563	"Can't open %s/%s : %s" % (self.certdir, aid, e))
564	resp = self.build_access_response({ 'fedid': allocID } ,
565	ap, services)
566	return resp
567
568	def do_release_project(self, del_project, del_users, del_types):
569	"""
570	If a project and users has to be deleted, make the call.
571	"""
572	msg = { 'project': { }}
573	if del_project:
574	msg['project']['name']= {'localname': del_project}
575	users = [ ]
576	for u in del_users.keys():
577	users.append({ 'userID': { 'localname': u },\
578	'access' : \
579	[ {'sshPubkey' : s } for s in del_users[u]]\
580	})
581	if users:
582	msg['project']['user'] = users
583	if len(del_types) > 0:
584	msg['resources'] = { 'node': \
585	[ {'hardware': [ h ] } for h in del_types ]\
586	}
587	if self.allocate_project.release_project:
588	msg = { 'ReleaseProjectRequestBody' : msg}
589	self.allocate_project.release_project(msg)
590
591	def ReleaseAccess(self, req, fid):
592	# The dance to get into the request body
593	if req.has_key('ReleaseAccessRequestBody'):
594	req = req['ReleaseAccessRequestBody']
595	else:
596	raise service_error(service_error.req, "No request!?")
597
598	try:
599	if req['allocID'].has_key('localname'):
600	auth_attr = aid = req['allocID']['localname']
601	elif req['allocID'].has_key('fedid'):
602	aid = unicode(req['allocID']['fedid'])
603	auth_attr = req['allocID']['fedid']
604	else:
605	raise service_error(service_error.req,
606	"Only localnames and fedids are understood")
607	except KeyError:
608	raise service_error(service_error.req, "Badly formed request")
609
610	self.log.debug("[access] deallocation requested for %s by %s" % \
611	(aid, fid))
612	if not self.auth.check_attribute(fid, auth_attr):
613	self.log.debug("[access] deallocation denied for %s", aid)
614	raise service_error(service_error.access, "Access Denied")
615
616	# If we know this allocation, reduce the reference counts and
617	# remove the local allocations. Otherwise report an error. If
618	# there is an allocation to delete, del_users will be a dictonary
619	# of sets where the key is the user that owns the keys in the set.
620	# We use a set to avoid duplicates. del_project is just the name
621	# of any dynamic project to delete. We're somewhat lazy about
622	# deleting authorization attributes. Having access to something
623	# that doesn't exist isn't harmful.
624	del_users = { }
625	del_project = None
626	del_types = set()
627
628	self.state_lock.acquire()
629	if aid in self.allocation:
630	self.log.debug("Found allocation for %s" %aid)
631	for k in self.allocation[aid]['keys']:
632	kk = "%s:%s" % k
633	self.keys[kk] -= 1
634	if self.keys[kk] == 0:
635	if not del_users.has_key(k[0]):
636	del_users[k[0]] = set()
637	del_users[k[0]].add(k[1])
638	del self.keys[kk]
639
640	if 'project' in self.allocation[aid]:
641	pname = self.allocation[aid]['project']
642	self.projects[pname] -= 1
643	if self.projects[pname] == 0:
644	del_project = pname
645	del self.projects[pname]
646
647	if 'types' in self.allocation[aid]:
648	for t in self.allocation[aid]['types']:
649	self.types[t] -= 1
650	if self.types[t] == 0:
651	if not del_project: del_project = t[0]
652	del_types.add(t[1])
653	del self.types[t]
654
655	del self.allocation[aid]
656	self.write_state()
657	self.state_lock.release()
658	# If we actually have resources to deallocate, prepare the call.
659	if del_project or del_users:
660	self.do_release_project(del_project, del_users, del_types)
661	# And remove the access cert
662	cf = "%s/%s.pem" % (self.certdir, aid)
663	self.log.debug("Removing %s" % cf)
664	os.remove(cf)
665	return { 'allocID': req['allocID'] }
666	else:
667	self.state_lock.release()
668	raise service_error(service_error.req, "No such allocation")
669
670	# These are subroutines for StartSegment
671	def generate_ns2(self, topo, expfn, softdir, connInfo):
672	"""
673	Convert topo into an ns2 file, decorated with appropriate commands for
674	the particular testbed setup. Convert all requests for software, etc
675	to point at the staged copies on this testbed and add the federation
676	startcommands.
677	"""
678	class dragon_commands:
679	"""
680	Functor to spit out approrpiate dragon commands for nodes listed in
681	the connectivity description. The constructor makes a dict mapping
682	dragon nodes to their parameters and the __call__ checks each
683	element in turn for membership.
684	"""
685	def __init__(self, map):
686	self.node_info = map
687
688	def __call__(self, e):
689	s = ""
690	if isinstance(e, topdl.Computer):
691	if self.node_info.has_key(e.name):
692	info = self.node_info[e.name]
693	for ifname, vlan, type in info:
694	for i in e.interface:
695	if i.name == ifname:
696	addr = i.get_attribute('ip4_address')
697	subs = i.substrate[0]
698	break
699	else:
700	raise service_error(service_error.internal,
701	"No interface %s on element %s" % \
702	(ifname, e.name))
703	# XXX: do netmask right
704	if type =='link':
705	s = ("tb-allow-external ${%s} " + \
706	"dragonportal ip %s vlan %s " + \
707	"netmask 255.255.255.0\n") % \
708	(topdl.to_tcl_name(e.name), addr, vlan)
709	elif type =='lan':
710	s = ("tb-allow-external ${%s} " + \
711	"dragonportal " + \
712	"ip %s vlan %s usurp %s\n") % \
713	(topdl.to_tcl_name(e.name), addr,
714	vlan, subs)
715	else:
716	raise service_error(service_error_internal,
717	"Unknown DRAGON type %s" % type)
718	return s
719
720	class not_dragon:
721	"""
722	Return true if a node is in the given map of dragon nodes.
723	"""
724	def __init__(self, map):
725	self.nodes = set(map.keys())
726
727	def __call__(self, e):
728	return e.name not in self.nodes
729
730	# Main line of generate_ns2
731	t = topo.clone()
732
733	# Create the map of nodes that need direct connections (dragon
734	# connections) from the connInfo
735	dragon_map = { }
736	for i in [ i for i in connInfo if i['type'] == 'transit']:
737	for a in i.get('fedAttr', []):
738	if a['attribute'] == 'vlan_id':
739	vlan = a['value']
740	break
741	else:
742	raise service_error(service_error.internal, "No vlan tag")
743	members = i.get('member', [])
744	if len(members) > 1: type = 'lan'
745	else: type = 'link'
746
747	try:
748	for m in members:
749	if m['element'] in dragon_map:
750	dragon_map[m['element']].append(( m['interface'],
751	vlan, type))
752	else:
753	dragon_map[m['element']] = [( m['interface'],
754	vlan, type),]
755	except KeyError:
756	raise service_error(service_error.req,
757	"Missing connectivity info")
758
759	# Weed out the things we aren't going to instantiate: Segments, portal
760	# substrates, and portal interfaces. (The copy in the for loop allows
761	# us to delete from e.elements in side the for loop). While we're
762	# touching all the elements, we also adjust paths from the original
763	# testbed to local testbed paths and put the federation commands into
764	# the start commands
765	for e in [e for e in t.elements]:
766	if isinstance(e, topdl.Segment):
767	t.elements.remove(e)
768	if isinstance(e, topdl.Computer):
769	self.add_kit(e, self.federation_software)
770	if e.get_attribute('portal') and self.portal_startcommand:
771	# Add local portal support software
772	self.add_kit(e, self.portal_software)
773	# Portals never have a user-specified start command
774	e.set_attribute('startup', self.portal_startcommand)
775	elif self.node_startcommand:
776	if e.get_attribute('startup'):
777	e.set_attribute('startup', "%s \\$USER '%s'" % \
778	(self.node_startcommand,
779	e.get_attribute('startup')))
780	else:
781	e.set_attribute('startup', self.node_startcommand)
782
783	dinf = [i[0] for i in dragon_map.get(e.name, []) ]
784	# Remove portal interfaces that do not connect to DRAGON
785	e.interface = [i for i in e.interface \
786	if not i.get_attribute('portal') or i.name in dinf ]
787	# Fix software paths
788	for s in getattr(e, 'software', []):
789	s.location = re.sub("^.*/", softdir, s.location)
790
791	t.substrates = [ s.clone() for s in t.substrates ]
792	t.incorporate_elements()
793
794	# Customize the ns2 output for local portal commands and images
795	filters = []
796
797	if self.dragon_endpoint:
798	add_filter = not_dragon(dragon_map)
799	filters.append(dragon_commands(dragon_map))
800	else:
801	add_filter = None
802
803	if self.portal_command:
804	filters.append(topdl.generate_portal_command_filter(
805	self.portal_command, add_filter=add_filter))
806
807	if self.portal_image:
808	filters.append(topdl.generate_portal_image_filter(
809	self.portal_image))
810
811	if self.portal_type:
812	filters.append(topdl.generate_portal_hardware_filter(
813	self.portal_type))
814
815	# Convert to ns and write it out
816	expfile = topdl.topology_to_ns2(t, filters)
817	try:
818	f = open(expfn, "w")
819	print >>f, expfile
820	f.close()
821	except EnvironmentError:
822	raise service_error(service_error.internal,
823	"Cannot write experiment file %s: %s" % (expfn,e))
824
825	def export_store_info(self, cf, proj, ename, connInfo):
826	"""
827	For the export requests in the connection info, install the peer names
828	at the experiment controller via SetValue calls.
829	"""
830
831	for c in connInfo:
832	for p in [ p for p in c.get('parameter', []) \
833	if p.get('type', '') == 'output']:
834
835	if p.get('name', '') == 'peer':
836	k = p.get('key', None)
837	surl = p.get('store', None)
838	if surl and k and k.index('/') != -1:
839	value = "%s.%s.%s%s" % \
840	(k[k.index('/')+1:], ename, proj, self.domain)
841	req = { 'name': k, 'value': value }
842	self.log.debug("Setting %s to %s on %s" % \
843	(k, value, surl))
844	self.call_SetValue(surl, req, cf)
845	else:
846	self.log.error("Bad export request: %s" % p)
847	elif p.get('name', '') == 'ssh_port':
848	k = p.get('key', None)
849	surl = p.get('store', None)
850	if surl and k:
851	req = { 'name': k, 'value': self.ssh_port }
852	self.log.debug("Setting %s to %s on %s" % \
853	(k, self.ssh_port, surl))
854	self.call_SetValue(surl, req, cf)
855	else:
856	self.log.error("Bad export request: %s" % p)
857	else:
858	self.log.error("Unknown export parameter: %s" % \
859	p.get('name'))
860	continue
861
862	def add_seer_node(self, topo, name, startup):
863	"""
864	Add a seer node to the given topology, with the startup command passed
865	in. Used by configure seer_services.
866	"""
867	c_node = topdl.Computer(
868	name=name,
869	os= topdl.OperatingSystem(
870	attribute=[
871	{ 'attribute': 'osid',
872	'value': self.local_seer_image },
873	]),
874	attribute=[
875	{ 'attribute': 'startup', 'value': startup },
876	]
877	)
878	self.add_kit(c_node, self.local_seer_software)
879	topo.elements.append(c_node)
880
881	def configure_seer_services(self, services, topo, softdir):
882	"""
883	Make changes to the topology required for the seer requests being made.
884	Specifically, add any control or master nodes required and set up the
885	start commands on the nodes to interconnect them.
886	"""
887	local_seer = False # True if we need to add a control node
888	collect_seer = False # True if there is a seer-master node
889	seer_master= False # True if we need to add the seer-master
890	for s in services:
891	s_name = s.get('name', '')
892	s_vis = s.get('visibility','')
893
894	if s_name == 'local_seer_control' and s_vis == 'export':
895	local_seer = True
896	elif s_name == 'seer_master':
897	if s_vis == 'import':
898	collect_seer = True
899	elif s_vis == 'export':
900	seer_master = True
901
902	# We've got the whole picture now, so add nodes if needed and configure
903	# them to interconnect properly.
904	if local_seer or seer_master:
905	# Copy local seer control node software to the tempdir
906	for l, f in self.local_seer_software:
907	base = os.path.basename(f)
908	copy_file(f, "%s/%s" % (softdir, base))
909	# If we're collecting seers somewhere the controllers need to talk to
910	# the master. In testbeds that export the service, that will be a
911	# local node that we'll add below. Elsewhere it will be the control
912	# portal that will port forward to the exporting master.
913	if local_seer:
914	if collect_seer:
915	startup = "%s -C %s" % (self.local_seer_start, "seer-master")
916	else:
917	startup = self.local_seer_start
918	self.add_seer_node(topo, 'control', startup)
919	# If this is the seer master, add that node, too.
920	if seer_master:
921	self.add_seer_node(topo, 'seer-master',
922	"%s -R -n -R seer-master -R -A -R sink" % \
923	self.seer_master_start)
924
925	def retrieve_software(self, topo, certfile, softdir):
926	"""
927	Collect the software that nodes in the topology need loaded and stage
928	it locally. This implies retrieving it from the experiment_controller
929	and placing it into softdir. Certfile is used to prove that this node
930	has access to that data (it's the allocation/segment fedid). Finally
931	local portal and federation software is also copied to the same staging
932	directory for simplicity - all software needed for experiment creation
933	is in softdir.
934	"""
935	sw = set()
936	for e in topo.elements:
937	for s in getattr(e, 'software', []):
938	sw.add(s.location)
939	for s in sw:
940	self.log.debug("Retrieving %s" % s)
941	try:
942	get_url(s, certfile, softdir)
943	except:
944	t, v, st = sys.exc_info()
945	raise service_error(service_error.internal,
946	"Error retrieving %s: %s" % (s, v))
947
948	# Copy local federation and portal node software to the tempdir
949	for s in (self.federation_software, self.portal_software):
950	for l, f in s:
951	base = os.path.basename(f)
952	copy_file(f, "%s/%s" % (softdir, base))
953
954
955	def initialize_experiment_info(self, attrs, aid, certfile, tmpdir):
956	"""
957	Gather common configuration files, retrieve or create an experiment
958	name and project name, and return the ssh_key filenames. Create an
959	allocation log bound to the state log variable as well.
960	"""
961	configs = set(('hosts', 'ssh_pubkey', 'ssh_secretkey'))
962	ename = None
963	pubkey_base = None
964	secretkey_base = None
965	proj = None
966	user = None
967	alloc_log = None
968
969	for a in attrs:
970	if a['attribute'] in configs:
971	try:
972	self.log.debug("Retrieving %s from %s" % \
973	(a['attribute'], a['value']))
974	get_url(a['value'], certfile, tmpdir)
975	except:
976	t, v, st = sys.exc_info()
977	raise service_error(service_error.internal,
978	"Error retrieving %s: %s" % (a.get('value', ""), v))
979	if a['attribute'] == 'ssh_pubkey':
980	pubkey_base = a['value'].rpartition('/')[2]
981	if a['attribute'] == 'ssh_secretkey':
982	secretkey_base = a['value'].rpartition('/')[2]
983	if a['attribute'] == 'experiment_name':
984	ename = a['value']
985
986	if not ename:
987	ename = ""
988	for i in range(0,5):
989	ename += random.choice(string.ascii_letters)
990	self.log.warn("No experiment name: picked one randomly: %s" \
991	% ename)
992
993	if not pubkey_base:
994	raise service_error(service_error.req,
995	"No public key attribute")
996
997	if not secretkey_base:
998	raise service_error(service_error.req,
999	"No secret key attribute")
1000
1001	self.state_lock.acquire()
1002	if aid in self.allocation:
1003	proj = self.allocation[aid].get('project', None)
1004	if not proj:
1005	proj = self.allocation[aid].get('sproject', None)
1006	user = self.allocation[aid].get('user', None)
1007	self.allocation[aid]['experiment'] = ename
1008	self.allocation[aid]['log'] = [ ]
1009	# Create a logger that logs to the experiment's state object as
1010	# well as to the main log file.
1011	alloc_log = logging.getLogger('fedd.access.%s' % ename)
1012	h = logging.StreamHandler(
1013	list_log.list_log(self.allocation[aid]['log']))
1014	# XXX: there should be a global one of these rather than
1015	# repeating the code.
1016	h.setFormatter(logging.Formatter(
1017	"%(asctime)s %(name)s %(message)s",
1018	'%d %b %y %H:%M:%S'))
1019	alloc_log.addHandler(h)
1020	self.write_state()
1021	self.state_lock.release()
1022
1023	if not proj:
1024	raise service_error(service_error.internal,
1025	"Can't find project for %s" %aid)
1026
1027	if not user:
1028	raise service_error(service_error.internal,
1029	"Can't find creation user for %s" %aid)
1030
1031	return (ename, proj, user, pubkey_base, secretkey_base, alloc_log)
1032
1033	def finalize_experiment(self, starter, topo, aid, alloc_id):
1034	"""
1035	Store key bits of experiment state in the global repository, including
1036	the response that may need to be replayed, and return the response.
1037	"""
1038	# Copy the assigned names into the return topology
1039	embedding = [ ]
1040	for n in starter.node:
1041	embedding.append({
1042	'toponame': n,
1043	'physname': ["%s%s" % (starter.node[n], self.domain)],
1044	})
1045	# Grab the log (this is some anal locking, but better safe than
1046	# sorry)
1047	self.state_lock.acquire()
1048	logv = "".join(self.allocation[aid]['log'])
1049	# It's possible that the StartSegment call gets retried (!).
1050	# if the 'started' key is in the allocation, we'll return it rather
1051	# than redo the setup.
1052	self.allocation[aid]['started'] = {
1053	'allocID': alloc_id,
1054	'allocationLog': logv,
1055	'segmentdescription': {
1056	'topdldescription': topo.clone().to_dict()
1057	},
1058	'embedding': embedding
1059	}
1060	retval = copy.copy(self.allocation[aid]['started'])
1061	self.write_state()
1062	self.state_lock.release()
1063	return retval
1064
1065	# End of StartSegment support routines
1066
1067	def StartSegment(self, req, fid):
1068	err = None # Any service_error generated after tmpdir is created
1069	rv = None # Return value from segment creation
1070
1071	try:
1072	req = req['StartSegmentRequestBody']
1073	auth_attr = req['allocID']['fedid']
1074	topref = req['segmentdescription']['topdldescription']
1075	except KeyError:
1076	raise service_error(server_error.req, "Badly formed request")
1077
1078	connInfo = req.get('connection', [])
1079	services = req.get('service', [])
1080	aid = "%s" % auth_attr
1081	attrs = req.get('fedAttr', [])
1082	if not self.auth.check_attribute(fid, auth_attr):
1083	raise service_error(service_error.access, "Access denied")
1084	else:
1085	# See if this is a replay of an earlier succeeded StartSegment -
1086	# sometimes SSL kills 'em. If so, replay the response rather than
1087	# redoing the allocation.
1088	self.state_lock.acquire()
1089	retval = self.allocation[aid].get('started', None)
1090	self.state_lock.release()
1091	if retval:
1092	self.log.warning("Duplicate StartSegment for %s: " % aid + \
1093	"replaying response")
1094	return retval
1095
1096	# A new request. Do it.
1097
1098	if topref: topo = topdl.Topology(**topref)
1099	else:
1100	raise service_error(service_error.req,
1101	"Request missing segmentdescription'")
1102
1103	certfile = "%s/%s.pem" % (self.certdir, auth_attr)
1104	try:
1105	tmpdir = tempfile.mkdtemp(prefix="access-")
1106	softdir = "%s/software" % tmpdir
1107	os.mkdir(softdir)
1108	except EnvironmentError:
1109	raise service_error(service_error.internal, "Cannot create tmp dir")
1110
1111	# Try block alllows us to clean up temporary files.
1112	try:
1113	self.retrieve_software(topo, certfile, softdir)
1114	ename, proj, user, pubkey_base, secretkey_base, alloc_log = \
1115	self.initialize_experiment_info(attrs, aid,
1116	certfile, tmpdir)
1117
1118	# Set up userconf and seer if needed
1119	self.configure_userconf(services, tmpdir)
1120	self.configure_seer_services(services, topo, softdir)
1121	# Get and send synch store variables
1122	self.export_store_info(certfile, proj, ename, connInfo)
1123	self.import_store_info(certfile, connInfo)
1124
1125	expfile = "%s/experiment.tcl" % tmpdir
1126
1127	self.generate_portal_configs(topo, pubkey_base,
1128	secretkey_base, tmpdir, proj, ename, connInfo, services)
1129	self.generate_ns2(topo, expfile,
1130	"/proj/%s/software/%s/" % (proj, ename), connInfo)
1131
1132	starter = self.start_segment(keyfile=self.ssh_privkey_file,
1133	debug=self.create_debug, log=alloc_log, boss=self.boss,
1134	cert=self.xmlrpc_cert)
1135	rv = starter(self, ename, proj, user, expfile, tmpdir)
1136	except service_error, e:
1137	err = e
1138	except:
1139	t, v, st = sys.exc_info()
1140	err = service_error(service_error.internal, "%s: %s" % \
1141	(v, traceback.extract_tb(st)))
1142
1143	# Walk up tmpdir, deleting as we go
1144	if self.cleanup: self.remove_dirs(tmpdir)
1145	else: self.log.debug("[StartSegment]: not removing %s" % tmpdir)
1146
1147	if rv:
1148	return self.finalize_experiment(starter, topo, aid, req['allocID'])
1149	elif err:
1150	raise service_error(service_error.federant,
1151	"Swapin failed: %s" % err)
1152	else:
1153	raise service_error(service_error.federant, "Swapin failed")
1154
1155	def TerminateSegment(self, req, fid):
1156	try:
1157	req = req['TerminateSegmentRequestBody']
1158	except KeyError:
1159	raise service_error(server_error.req, "Badly formed request")
1160
1161	auth_attr = req['allocID']['fedid']
1162	aid = "%s" % auth_attr
1163	attrs = req.get('fedAttr', [])
1164	if not self.auth.check_attribute(fid, auth_attr):
1165	raise service_error(service_error.access, "Access denied")
1166
1167	self.state_lock.acquire()
1168	if aid in self.allocation:
1169	proj = self.allocation[aid].get('project', None)
1170	if not proj:
1171	proj = self.allocation[aid].get('sproject', None)
1172	user = self.allocation[aid].get('user', None)
1173	ename = self.allocation[aid].get('experiment', None)
1174	else:
1175	proj = None
1176	user = None
1177	ename = None
1178	self.state_lock.release()
1179
1180	if not proj:
1181	raise service_error(service_error.internal,
1182	"Can't find project for %s" % aid)
1183
1184	if not user:
1185	raise service_error(service_error.internal,
1186	"Can't find creation user for %s" % aid)
1187	if not ename:
1188	raise service_error(service_error.internal,
1189	"Can't find experiment name for %s" % aid)
1190	stopper = self.stop_segment(keyfile=self.ssh_privkey_file,
1191	debug=self.create_debug, boss=self.boss, cert=self.xmlrpc_cert)
1192	stopper(self, user, proj, ename)
1193	return { 'allocID': req['allocID'] }

Note: See TracBrowser for help on using the repository browser.

Download in other formats: