Context Navigation

source: fedd/federation/emulab_access.py @ 9216970

compt_changes

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

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

Download in other formats: