Context Navigation

source: fedd/federation/emulab_access.py @ 5bf359d

axis_examplecompt_changesinfo-opsversion-3.01version-3.02

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

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

Download in other formats: