Context Navigation

source: fedd/federation/emulab_access.py @ fe6e0be

compt_changes

Last change on this file since fe6e0be was 7653f01, checked in by Ted Faber <faber@…>, 13 years ago
Broke debugging.
Property mode set to `100644`
File size: 34.6 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
[7653f01]	758	i = 0 # For fake debugging instantiation
[06cc65b]	759	# Copy the assigned names into the return topology
[c6f867c]	760	for e in t.elements:
[29d5f7c]	761	if isinstance(e, topdl.Computer):
	762	if not self.create_debug:
[6e33086]	763	if e.name in info.node:
[1ae1aa2]	764	add_new(("%s%s" %
	765	(info.node[e.name].pname, self.domain),),
	766	e.localname)
	767	e.status = info.node[e.name].status
	768	os = info.node[e.name].getOS()
[cebcdce]	769	if os: merge_os(os, e)
[29d5f7c]	770	else:
	771	# Simple debugging assignment
[6e33086]	772	add_new(("node%d%s" % (i, self.domain),), e.localname)
[29d5f7c]	773	e.status = 'active'
[6e33086]	774	add_new(('testop1', 'testop2'), e.operation)
[29d5f7c]	775	i += 1
	776
[6e33086]	777
	778	def finalize_experiment(self, starter, topo, aid, alloc_id, proof):
	779	"""
	780	Store key bits of experiment state in the global repository, including
	781	the response that may need to be replayed, and return the response.
	782	"""
	783	i = 0
	784	t = topo.clone()
	785	self.decorate_topology(starter, t)
[06cc65b]	786	# Grab the log (this is some anal locking, but better safe than
	787	# sorry)
	788	self.state_lock.acquire()
	789	logv = "".join(self.allocation[aid]['log'])
	790	# It's possible that the StartSegment call gets retried (!).
	791	# if the 'started' key is in the allocation, we'll return it rather
	792	# than redo the setup.
	793	self.allocation[aid]['started'] = {
	794	'allocID': alloc_id,
	795	'allocationLog': logv,
	796	'segmentdescription': {
[b709861]	797	'topdldescription': t.to_dict()
[06cc65b]	798	},
[e83f2f2]	799	'proof': proof.to_dict(),
[06cc65b]	800	}
[b709861]	801	self.allocation[aid]['topo'] = t
[06cc65b]	802	retval = copy.copy(self.allocation[aid]['started'])
	803	self.write_state()
	804	self.state_lock.release()
	805	return retval
	806
	807	# End of StartSegment support routines
	808
	809	def StartSegment(self, req, fid):
[b770aa0]	810	err = None # Any service_error generated after tmpdir is created
	811	rv = None # Return value from segment creation
	812
[cc8d8e9]	813	try:
	814	req = req['StartSegmentRequestBody']
[06cc65b]	815	auth_attr = req['allocID']['fedid']
	816	topref = req['segmentdescription']['topdldescription']
[cc8d8e9]	817	except KeyError:
	818	raise service_error(server_error.req, "Badly formed request")
[ecca6eb]	819
[e02cd14]	820	connInfo = req.get('connection', [])
	821	services = req.get('service', [])
[ecca6eb]	822	aid = "%s" % auth_attr
[6c57fe9]	823	attrs = req.get('fedAttr', [])
[e83f2f2]	824
	825	access_ok, proof = self.auth.check_attribute(fid, auth_attr,
	826	with_proof=True)
	827	if not access_ok:
[ecca6eb]	828	raise service_error(service_error.access, "Access denied")
[cd06678]	829	else:
	830	# See if this is a replay of an earlier succeeded StartSegment -
	831	# sometimes SSL kills 'em. If so, replay the response rather than
	832	# redoing the allocation.
	833	self.state_lock.acquire()
	834	retval = self.allocation[aid].get('started', None)
	835	self.state_lock.release()
	836	if retval:
	837	self.log.warning("Duplicate StartSegment for %s: " % aid + \
	838	"replaying response")
	839	return retval
	840
	841	# A new request. Do it.
[6c57fe9]	842
[06cc65b]	843	if topref: topo = topdl.Topology(**topref)
[6c57fe9]	844	else:
	845	raise service_error(service_error.req,
	846	"Request missing segmentdescription'")
[2761484]	847
[6c57fe9]	848	certfile = "%s/%s.pem" % (self.certdir, auth_attr)
	849	try:
	850	tmpdir = tempfile.mkdtemp(prefix="access-")
[ecca6eb]	851	softdir = "%s/software" % tmpdir
[c23684d]	852	os.mkdir(softdir)
[d3c8759]	853	except EnvironmentError:
[6c57fe9]	854	raise service_error(service_error.internal, "Cannot create tmp dir")
	855
[b770aa0]	856	# Try block alllows us to clean up temporary files.
	857	try:
[06cc65b]	858	self.retrieve_software(topo, certfile, softdir)
[f77a256]	859	ename, proj, user, xmlrpc_cert, pubkey_base, secretkey_base, \
	860	alloc_log = self.initialize_experiment_info(attrs, aid,
	861	certfile, tmpdir)
[9b3627e]	862
[f3898f7]	863	if '/' in proj: proj, gid = proj.split('/')
	864	else: gid = None
	865
	866
[06cc65b]	867	# Set up userconf and seer if needed
[c200d36]	868	self.configure_userconf(services, tmpdir)
[49051fb]	869	self.configure_seer_services(services, topo, softdir)
[06cc65b]	870	# Get and send synch store variables
[2761484]	871	self.export_store_info(certfile, proj, ename, connInfo)
	872	self.import_store_info(certfile, connInfo)
	873
[b770aa0]	874	expfile = "%s/experiment.tcl" % tmpdir
	875
	876	self.generate_portal_configs(topo, pubkey_base,
[06cc65b]	877	secretkey_base, tmpdir, proj, ename, connInfo, services)
[b770aa0]	878	self.generate_ns2(topo, expfile,
[8cf2b90e]	879	"/proj/%s/software/%s/" % (proj, ename), connInfo)
[f07fa49]	880
[b770aa0]	881	starter = self.start_segment(keyfile=self.ssh_privkey_file,
[181aeb4]	882	debug=self.create_debug, log=alloc_log, boss=self.boss,
[f77a256]	883	ops=self.ops, cert=xmlrpc_cert)
[f3898f7]	884	rv = starter(self, ename, proj, user, expfile, tmpdir, gid=gid)
[b770aa0]	885	except service_error, e:
	886	err = e
[06cc65b]	887	except:
	888	t, v, st = sys.exc_info()
	889	err = service_error(service_error.internal, "%s: %s" % \
	890	(v, traceback.extract_tb(st)))
[b770aa0]	891
[574055e]	892	# Walk up tmpdir, deleting as we go
[06cc65b]	893	if self.cleanup: self.remove_dirs(tmpdir)
	894	else: self.log.debug("[StartSegment]: not removing %s" % tmpdir)
[574055e]	895
[fd556d1]	896	if rv:
[e83f2f2]	897	return self.finalize_experiment(starter, topo, aid, req['allocID'],
	898	proof)
[b770aa0]	899	elif err:
	900	raise service_error(service_error.federant,
	901	"Swapin failed: %s" % err)
[fd556d1]	902	else:
	903	raise service_error(service_error.federant, "Swapin failed")
[5ae3857]	904
	905	def TerminateSegment(self, req, fid):
	906	try:
	907	req = req['TerminateSegmentRequestBody']
	908	except KeyError:
	909	raise service_error(server_error.req, "Badly formed request")
	910
	911	auth_attr = req['allocID']['fedid']
	912	aid = "%s" % auth_attr
	913	attrs = req.get('fedAttr', [])
[e83f2f2]	914
	915	access_ok, proof = self.auth.check_attribute(fid, auth_attr,
	916	with_proof=True)
	917	if not access_ok:
[5ae3857]	918	raise service_error(service_error.access, "Access denied")
	919
	920	self.state_lock.acquire()
[06cc65b]	921	if aid in self.allocation:
[5ae3857]	922	proj = self.allocation[aid].get('project', None)
	923	user = self.allocation[aid].get('user', None)
[f77a256]	924	xmlrpc_cert = self.allocation[aid].get('cert', None)
[5ae3857]	925	ename = self.allocation[aid].get('experiment', None)
[05c41f5]	926	nonce = self.allocation[aid].get('nonce', False)
[1d913e13]	927	else:
	928	proj = None
	929	user = None
	930	ename = None
[05c41f5]	931	nonce = False
[f77a256]	932	xmlrpc_cert = None
[5ae3857]	933	self.state_lock.release()
	934
	935	if not proj:
	936	raise service_error(service_error.internal,
	937	"Can't find project for %s" % aid)
[f3898f7]	938	else:
	939	if '/' in proj: proj, gid = proj.split('/')
	940	else: gid = None
[5ae3857]	941
	942	if not user:
	943	raise service_error(service_error.internal,
	944	"Can't find creation user for %s" % aid)
	945	if not ename:
	946	raise service_error(service_error.internal,
	947	"Can't find experiment name for %s" % aid)
[fd556d1]	948	stopper = self.stop_segment(keyfile=self.ssh_privkey_file,
[c7141dc]	949	debug=self.create_debug, boss=self.boss, ops=self.ops,
[f77a256]	950	cert=xmlrpc_cert)
[05c41f5]	951	stopper(self, user, proj, ename, gid, nonce)
[e83f2f2]	952	return { 'allocID': req['allocID'], 'proof': proof.to_dict() }
[45e880d]	953
	954	def InfoSegment(self, req, fid):
	955	try:
	956	req = req['InfoSegmentRequestBody']
	957	except KeyError:
	958	raise service_error(server_error.req, "Badly formed request")
	959
	960	auth_attr = req['allocID']['fedid']
	961	aid = "%s" % auth_attr
	962
	963	access_ok, proof = self.auth.check_attribute(fid, auth_attr,
	964	with_proof=True)
	965	if not access_ok:
	966	raise service_error(service_error.access, "Access denied")
	967
	968	self.state_lock.acquire()
	969	if aid in self.allocation:
	970	topo = self.allocation[aid].get('topo', None)
	971	if topo: topo = topo.clone()
	972	proj = self.allocation[aid].get('project', None)
	973	user = self.allocation[aid].get('user', None)
[f77a256]	974	xmlrpc_cert = self.allocation[aid].get('cert', None)
[45e880d]	975	ename = self.allocation[aid].get('experiment', None)
	976	else:
	977	proj = None
	978	user = None
	979	ename = None
[b709861]	980	topo = None
[f77a256]	981	xmlrpc_cert = None
[45e880d]	982	self.state_lock.release()
	983
	984	if not proj:
	985	raise service_error(service_error.internal,
	986	"Can't find project for %s" % aid)
	987	else:
	988	if '/' in proj: proj, gid = proj.split('/')
	989	else: gid = None
	990
	991	if not user:
	992	raise service_error(service_error.internal,
	993	"Can't find creation user for %s" % aid)
	994	if not ename:
	995	raise service_error(service_error.internal,
	996	"Can't find experiment name for %s" % aid)
	997	info = self.info_segment(keyfile=self.ssh_privkey_file,
[c7141dc]	998	debug=self.create_debug, boss=self.boss, ops=self.ops,
[f77a256]	999	cert=xmlrpc_cert)
[6e33086]	1000	info(self, user, proj, ename)
	1001	self.decorate_topology(info, topo)
[1ae1aa2]	1002	self.state_lock.acquire()
	1003	if aid in self.allocation:
	1004	self.allocation[aid]['topo'] = topo
	1005	self.write_state()
	1006	self.state_lock.release()
	1007
[7f57435]	1008	rv = {
[45e880d]	1009	'allocID': req['allocID'],
	1010	'proof': proof.to_dict(),
	1011	}
[7f57435]	1012	if topo:
	1013	rv['segmentdescription'] = { 'topdldescription' : topo.to_dict() }
	1014	return rv
[b709861]	1015
	1016	def OperationSegment(self, req, fid):
	1017	def get_pname(e):
	1018	"""
	1019	Get the physical name of a node
	1020	"""
	1021	if e.localname:
	1022	return re.sub('\..*','', e.localname[0])
	1023	else:
	1024	return None
	1025
	1026	try:
	1027	req = req['OperationSegmentRequestBody']
	1028	except KeyError:
	1029	raise service_error(server_error.req, "Badly formed request")
	1030
	1031	auth_attr = req['allocID']['fedid']
	1032	aid = "%s" % auth_attr
	1033
	1034	access_ok, proof = self.auth.check_attribute(fid, auth_attr,
	1035	with_proof=True)
	1036	if not access_ok:
	1037	raise service_error(service_error.access, "Access denied")
	1038
	1039	op = req.get('operation', None)
	1040	targets = req.get('target', None)
	1041	params = req.get('parameter', None)
	1042
	1043	if op is None :
	1044	raise service_error(service_error.req, "missing operation")
	1045	elif targets is None:
	1046	raise service_error(service_error.req, "no targets")
	1047
[f77a256]	1048	self.state_lock.acquire()
[b709861]	1049	if aid in self.allocation:
	1050	topo = self.allocation[aid].get('topo', None)
	1051	if topo: topo = topo.clone()
[f77a256]	1052	xmlrpc_cert = self.allocation[aid].get('cert', None)
[b709861]	1053	else:
	1054	topo = None
[f77a256]	1055	xmlrpc_cert = None
	1056	self.state_lock.release()
[b709861]	1057
	1058	targets = copy.copy(targets)
	1059	ptargets = { }
	1060	for e in topo.elements:
	1061	if isinstance(e, topdl.Computer):
	1062	if e.name in targets:
	1063	targets.remove(e.name)
	1064	pn = get_pname(e)
	1065	if pn: ptargets[e.name] = pn
	1066
	1067	status = [ operation_status(t, operation_status.no_target) \
	1068	for t in targets]
	1069
	1070	ops = self.operation_segment(keyfile=self.ssh_privkey_file,
[c7141dc]	1071	debug=self.create_debug, boss=self.boss, ops=self.ops,
[f77a256]	1072	cert=xmlrpc_cert)
[1ae1aa2]	1073	ops(self, op, ptargets, params, topo)
[b709861]	1074
	1075	status.extend(ops.status)
	1076
	1077	return {
	1078	'allocID': req['allocID'],
	1079	'status': [s.to_dict() for s in status],
	1080	'proof': proof.to_dict(),
	1081	}

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

Download in other formats: