Context Navigation

source: fedd/federation/emulab_access.py @ b931822

compt_changes

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

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

Download in other formats: