Context Navigation

source: fedd/federation/emulab_access.py @ b501f63

version-3.02

Last change on this file since b501f63 was 807b5ca, checked in by Ted Faber <faber@…>, 14 years ago
Stone typo in setting up interconnection testbeds.
Property mode set to `100644`
File size: 36.3 KB

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

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

Download in other formats: