Context Navigation

source: fedd/federation/protogeni_access.py @ 37ed9a5

axis_examplecompt_changesinfo-opsversion-3.01version-3.02

Last change on this file since 37ed9a5 was 37ed9a5, checked in by Ted Faber <faber@…>, 14 years ago
Much more nicely factored now.
Property mode set to `100644`
File size: 47.4 KB

Line
1	#!/usr/local/bin/python
2
3	import os,sys
4	import stat # for chmod constants
5	import re
6	import time
7	import string
8	import copy
9	import pickle
10	import logging
11	import subprocess
12	import random
13	import traceback
14
15	from threading import Thread, Timer, Lock
16	from M2Crypto.SSL import SSLError
17
18	from util import *
19	from access_project import access_project
20	from fedid import fedid, generate_fedid
21	from authorizer import authorizer
22	from service_error import service_error
23	from remote_service import xmlrpc_handler, soap_handler, service_caller
24
25	import httplib
26	import tempfile
27	from urlparse import urlparse
28
29	from access import access_base
30	from proxy_segment import proxy_segment
31
32	import topdl
33	import list_log
34
35
36	# Make log messages disappear if noone configures a fedd logger
37	class nullHandler(logging.Handler):
38	def emit(self, record): pass
39
40	fl = logging.getLogger("fedd.access")
41	fl.addHandler(nullHandler())
42
43	class protogeni_proxy(proxy_segment):
44	class ProtoGENIError(Exception):
45	def __init__(self, op, code, output):
46	Exception.__init__(self, output)
47	self.op = op
48	self.code = code
49	self.output = output
50
51	def __init__(self, log=None, keyfile=None, debug=False,
52	ch_url=None, sa_url=None, cm_url=None):
53	proxy_segment.__init__(self, log=log, keyfile=keyfile, debug=debug)
54
55	self.ProtoGENIError = protogeni_proxy.ProtoGENIError
56	self.ch_url = ch_url
57	self.sa_url = sa_url
58	self.cm_url = cm_url
59
60	self.call_SetValue = service_caller('SetValue')
61
62	self.debug_fail = ['Resolve']
63	self.debug_response = {
64	'RedeemTicket': ("XML blob1", "XML blob2"),
65	'SliceStatus': { 'status': 'ready' },
66	}
67
68
69	def pg_call(self, url, method, params, context):
70	max_retries = 5
71	retries = 0
72
73	s = service_caller(method, request_body_name="", strict=False)
74	self.log.debug("Calling %s %s" % (url, method))
75	if not self.debug:
76	while retries < max_retries:
77	r = s.call_xmlrpc_service(url, params, context=context)
78	code = r.get('code', -1)
79	if code == 0:
80	# Success leaves the loop here
81	return r.get('value', None)
82	elif code == 14 and retries +1 < max_retries:
83	# Busy resource
84	retries+= 1
85	self.log.info("Resource busy, retrying in 30 secs")
86	time.sleep(30)
87	else:
88	# NB: out of retries falls through to here
89	raise self.ProtoGENIError(op=method,
90	code=r.get('code', 'unknown'),
91	output=r.get('output', 'No output'))
92	else:
93	if method in self.debug_fail:
94	raise self.ProtoGENIError(op=method, code='unknown',
95	output='No output')
96	elif self.debug_response.has_key(method):
97	return self.debug_response[method]
98	else:
99	return "%s XML blob" % method
100
101
102
103	class access(access_base):
104	"""
105	The implementation of access control based on mapping users to projects.
106
107	Users can be mapped to existing projects or have projects created
108	dynamically. This implements both direct requests and proxies.
109	"""
110
111	def __init__(self, config=None, auth=None):
112	"""
113	Initializer. Pulls parameters out of the ConfigParser's access section.
114	"""
115
116	access_base.__init__(self, config, auth)
117
118	self.domain = config.get("access", "domain")
119	self.userconfdir = config.get("access","userconfdir")
120	self.userconfcmd = config.get("access","userconfcmd")
121	self.userconfurl = config.get("access","userconfurl")
122	self.federation_software = config.get("access", "federation_software")
123	self.portal_software = config.get("access", "portal_software")
124	self.ssh_port = config.get("access","ssh_port") or "22"
125	self.sshd = config.get("access","sshd")
126	self.sshd_config = config.get("access", "sshd_config")
127	self.access_type = config.get("access", "type")
128	self.staging_dir = config.get("access", "staging_dir") or "/tmp"
129	self.staging_host = config.get("access", "staging_host") \
130	or "ops.emulab.net"
131	self.local_seer_software = config.get("access", "local_seer_software")
132	self.local_seer_image = config.get("access", "local_seer_image")
133	self.local_seer_start = config.get("access", "local_seer_start")
134
135	self.dragon_endpoint = config.get("access", "dragon")
136	self.dragon_vlans = config.get("access", "dragon_vlans")
137	self.deter_internal = config.get("access", "deter_internal")
138
139	self.tunnel_config = config.getboolean("access", "tunnel_config")
140	self.portal_command = config.get("access", "portal_command")
141	self.portal_image = config.get("access", "portal_image")
142	self.portal_type = config.get("access", "portal_type") or "pc"
143	self.portal_startcommand = config.get("access", "portal_startcommand")
144	self.node_startcommand = config.get("access", "node_startcommand")
145
146	self.federation_software = self.software_list(self.federation_software)
147	self.portal_software = self.software_list(self.portal_software)
148	self.local_seer_software = self.software_list(self.local_seer_software)
149
150	self.renewal_interval = config.get("access", "renewal") or (3 * 60 )
151	self.renewal_interval = int(self.renewal_interval) * 60
152
153	self.ch_url = config.get("access", "ch_url")
154	self.sa_url = config.get("access", "sa_url")
155	self.cm_url = config.get("access", "cm_url")
156
157	self.restricted = [ ]
158
159	# read_state in the base_class
160	self.state_lock.acquire()
161	for a in ('allocation', 'projects', 'keys', 'types'):
162	if a not in self.state:
163	self.state[a] = { }
164	self.allocation = self.state['allocation']
165	self.projects = self.state['projects']
166	self.keys = self.state['keys']
167	self.types = self.state['types']
168	# Add the ownership attributes to the authorizer. Note that the
169	# indices of the allocation dict are strings, but the attributes are
170	# fedids, so there is a conversion.
171	for k in self.state.get('allocation', {}).keys():
172	for o in self.state['allocation'][k].get('owners', []):
173	self.auth.set_attribute(o, fedid(hexstr=k))
174	self.auth.set_attribute(fedid(hexstr=k),fedid(hexstr=k))
175
176	self.state_lock.release()
177
178
179	self.log = logging.getLogger("fedd.access")
180	set_log_level(config, "access", self.log)
181
182	self.access = { }
183	if config.has_option("access", "accessdb"):
184	self.read_access(config.get("access", "accessdb"),
185	access_obj=self.make_access_info)
186
187	self.lookup_access = self.lookup_access_base
188
189	self.call_SetValue = service_caller('SetValue')
190	self.call_GetValue = service_caller('GetValue')
191	self.exports = {
192	'local_seer_control': self.export_local_seer,
193	'seer_master': self.export_seer_master,
194	'hide_hosts': self.export_hide_hosts,
195	}
196
197	if not self.local_seer_image or not self.local_seer_software or \
198	not self.local_seer_start:
199	if 'local_seer_control' in self.exports:
200	del self.exports['local_seer_control']
201
202	if not self.local_seer_image or not self.local_seer_software or \
203	not self.seer_master_start:
204	if 'seer_master' in self.exports:
205	del self.exports['seer_master']
206
207	self.RenewSlices()
208
209	self.soap_services = {\
210	'RequestAccess': soap_handler("RequestAccess", self.RequestAccess),
211	'ReleaseAccess': soap_handler("ReleaseAccess", self.ReleaseAccess),
212	'StartSegment': soap_handler("StartSegment", self.StartSegment),
213	'TerminateSegment': soap_handler("TerminateSegment",
214	self.TerminateSegment),
215	}
216	self.xmlrpc_services = {\
217	'RequestAccess': xmlrpc_handler('RequestAccess',
218	self.RequestAccess),
219	'ReleaseAccess': xmlrpc_handler('ReleaseAccess',
220	self.ReleaseAccess),
221	'StartSegment': xmlrpc_handler("StartSegment", self.StartSegment),
222	'TerminateSegment': xmlrpc_handler('TerminateSegment',
223	self.TerminateSegment),
224	}
225
226	@staticmethod
227	def make_access_info(s):
228	"""
229	Split a string of the form (id, id, id, id) ito its constituent tuples
230	and return them as a tuple. Use to import access info from the
231	access_db.
232	"""
233
234	ss = s.strip()
235	if ss.startswith('(') and ss.endswith(')'):
236	l = [ s.strip() for s in ss[1:-1].split(",")]
237	if len(l) == 4:
238	return tuple(l)
239	else:
240	raise self.parse_error(
241	"Exactly 4 elements in access info required")
242	else:
243	raise self.parse_error("Expecting parenthezied values")
244
245
246	def get_handler(self, path, fid):
247	self.log.info("Get handler %s %s" % (path, fid))
248	if self.auth.check_attribute(fid, path) and self.userconfdir:
249	return ("%s/%s" % (self.userconfdir, path), "application/binary")
250	else:
251	return (None, None)
252
253	def build_access_response(self, alloc_id, services):
254	"""
255	Create the SOAP response.
256
257	Build the dictionary description of the response and use
258	fedd_utils.pack_soap to create the soap message. ap is the allocate
259	project message returned from a remote project allocation (even if that
260	allocation was done locally).
261	"""
262	# Because alloc_id is already a fedd_services_types.IDType_Holder,
263	# there's no need to repack it
264	msg = {
265	'allocID': alloc_id,
266	'fedAttr': [
267	{ 'attribute': 'domain', 'value': self.domain } ,
268	]
269	}
270	if self.dragon_endpoint:
271	msg['fedAttr'].append({'attribute': 'dragon',
272	'value': self.dragon_endpoint})
273	if self.deter_internal:
274	msg['fedAttr'].append({'attribute': 'deter_internal',
275	'value': self.deter_internal})
276	#XXX: ??
277	if self.dragon_vlans:
278	msg['fedAttr'].append({'attribute': 'vlans',
279	'value': self.dragon_vlans})
280
281	if services:
282	msg['service'] = services
283	return msg
284
285	def RequestAccess(self, req, fid):
286	"""
287	Handle the access request.
288	"""
289
290	# The dance to get into the request body
291	if req.has_key('RequestAccessRequestBody'):
292	req = req['RequestAccessRequestBody']
293	else:
294	raise service_error(service_error.req, "No request!?")
295
296	if req.has_key('destinationTestbed'):
297	dt = unpack_id(req['destinationTestbed'])
298
299	# Request for this fedd
300	found, match = self.lookup_access(req, fid)
301	services, svc_state = self.export_services(req.get('service',[]),
302	None, None)
303	# keep track of what's been added
304	allocID, alloc_cert = generate_fedid(subj="alloc", log=self.log)
305	aid = unicode(allocID)
306
307	self.state_lock.acquire()
308	self.allocation[aid] = { }
309	# The protoGENI certificate
310	self.allocation[aid]['credentials'] = found
311	# The list of owner FIDs
312	self.allocation[aid]['owners'] = [ fid ]
313	self.write_state()
314	self.state_lock.release()
315	self.auth.set_attribute(fid, allocID)
316	self.auth.set_attribute(allocID, allocID)
317
318	try:
319	f = open("%s/%s.pem" % (self.certdir, aid), "w")
320	print >>f, alloc_cert
321	f.close()
322	except EnvironmentError, e:
323	raise service_error(service_error.internal,
324	"Can't open %s/%s : %s" % (self.certdir, aid, e))
325	return self.build_access_response({ 'fedid': allocID }, None)
326
327
328	def ReleaseAccess(self, req, fid):
329	# The dance to get into the request body
330	if req.has_key('ReleaseAccessRequestBody'):
331	req = req['ReleaseAccessRequestBody']
332	else:
333	raise service_error(service_error.req, "No request!?")
334
335	# Local request
336	try:
337	if req['allocID'].has_key('localname'):
338	auth_attr = aid = req['allocID']['localname']
339	elif req['allocID'].has_key('fedid'):
340	aid = unicode(req['allocID']['fedid'])
341	auth_attr = req['allocID']['fedid']
342	else:
343	raise service_error(service_error.req,
344	"Only localnames and fedids are understood")
345	except KeyError:
346	raise service_error(service_error.req, "Badly formed request")
347
348	self.log.debug("[access] deallocation requested for %s", aid)
349	if not self.auth.check_attribute(fid, auth_attr):
350	self.log.debug("[access] deallocation denied for %s", aid)
351	raise service_error(service_error.access, "Access Denied")
352
353	self.state_lock.acquire()
354	if self.allocation.has_key(aid):
355	self.log.debug("Found allocation for %s" %aid)
356	del self.allocation[aid]
357	self.write_state()
358	self.state_lock.release()
359	# And remove the access cert
360	cf = "%s/%s.pem" % (self.certdir, aid)
361	self.log.debug("Removing %s" % cf)
362	os.remove(cf)
363	return { 'allocID': req['allocID'] }
364	else:
365	self.state_lock.release()
366	raise service_error(service_error.req, "No such allocation")
367
368	def manifest_to_dict(self, manifest, ignore_debug=False):
369	"""
370	Turn the manifest into a dict were each virtual nodename (i.e. the
371	topdl name) has an entry with the allocated machine in hostname and the
372	interfaces in 'interfaces'. I love having XML parser code lying
373	around.
374	"""
375	if self.create_debug and not ignore_debug:
376	self.log.debug("Returning null manifest dict")
377	return { }
378
379	# The class allows us to keep a little state - the dict under
380	# consteruction and the current entry in that dict for the interface
381	# element code.
382	class manifest_parser:
383	def __init__(self):
384	self.d = { }
385	self.current_key=None
386
387	# If the element is a node, create a dict entry for it. If it's an
388	# interface inside a node, add an entry in the interfaces list with
389	# the virtual name and component id.
390	def start_element(self, name, attrs):
391	if name == 'node':
392	self.current_key = attrs.get('virtual_id',"")
393	if self.current_key:
394	self.d[self.current_key] = {
395	'hostname': attrs.get('hostname', None),
396	'interfaces': { }
397	}
398	elif name == 'interface' and self.current_key:
399	self.d[self.current_key]['interfaces']\
400	[attrs.get('virtual_id','')] = \
401	attrs.get('component_id', None)
402	# When a node is finished, clear current_key
403	def end_element(self, name):
404	if name == 'node': self.current_key = None
405
406	node = { }
407
408	mp = manifest_parser()
409	p = xml.parsers.expat.ParserCreate()
410	# These are bound to the class we just created
411	p.StartElementHandler = mp.start_element
412	p.EndElementHandler = mp.end_element
413
414	p.Parse(manifest)
415	# Make the node dict that the callers expect
416	for k in mp.d:
417	node[k] = mp.d.get('hostname', '')
418	return mp.d
419
420	def fake_manifest(self, topo):
421	"""
422	Fake the output of manifest_to_dict with a bunch of generic node an
423	interface names, for debugging.
424	"""
425	node = { }
426	for i, e in enumerate([ e for e in topo.elements \
427	if isinstance(e, topdl.Computer)]):
428	node[e.name] = {
429	'hostname': "node%03d" % i,
430	'interfaces': { }
431	}
432	for j, inf in enumerate(e.interface):
433	node[e.name]['interfaces'][inf.name] = 'eth%d' % j
434
435	return node
436
437
438	def generate_portal_configs(self, topo, pubkey_base,
439	secretkey_base, tmpdir, leid, connInfo, services, nodes):
440
441	def conninfo_to_dict(key, info):
442	"""
443	Make a cpoy of the connection information about key, and flatten it
444	into a single dict by parsing out any feddAttrs.
445	"""
446
447	rv = None
448	for i in info:
449	if key == i.get('portal', "") or \
450	key in [e.get('element', "") \
451	for e in i.get('member', [])]:
452	rv = i.copy()
453	break
454
455	else:
456	return rv
457
458	if 'fedAttr' in rv:
459	for a in rv['fedAttr']:
460	attr = a.get('attribute', "")
461	val = a.get('value', "")
462	if attr and attr not in rv:
463	rv[attr] = val
464	del rv['fedAttr']
465	return rv
466
467	# XXX: un hardcode this
468	def client_null(f, s):
469	print >>f, "Service: %s" % s['name']
470
471	def client_seer_master(f, s):
472	print >>f, 'PortalAlias: seer-master'
473
474	def client_smb(f, s):
475	print >>f, "Service: %s" % s['name']
476	smbshare = None
477	smbuser = None
478	smbproj = None
479	for a in s.get('fedAttr', []):
480	if a.get('attribute', '') == 'SMBSHARE':
481	smbshare = a.get('value', None)
482	elif a.get('attribute', '') == 'SMBUSER':
483	smbuser = a.get('value', None)
484	elif a.get('attribute', '') == 'SMBPROJ':
485	smbproj = a.get('value', None)
486
487	if all((smbshare, smbuser, smbproj)):
488	print >>f, "SMBshare: %s" % smbshare
489	print >>f, "ProjectUser: %s" % smbuser
490	print >>f, "ProjectName: %s" % smbproj
491
492	def client_hide_hosts(f, s):
493	for a in s.get('fedAttr', [ ]):
494	if a.get('attribute', "") == 'hosts':
495	print >>f, 'Hide: %s' % a.get('value', "")
496
497	client_service_out = {
498	'SMB': client_smb,
499	'tmcd': client_null,
500	'seer': client_null,
501	'userconfig': client_null,
502	'project_export': client_null,
503	'seer_master': client_seer_master,
504	'hide_hosts': client_hide_hosts,
505	}
506
507	def client_seer_master_export(f, s):
508	print >>f, "AddedNode: seer-master"
509
510	def client_seer_local_export(f, s):
511	print >>f, "AddedNode: control"
512
513	client_export_service_out = {
514	'seer_master': client_seer_master_export,
515	'local_seer_control': client_seer_local_export,
516	}
517
518	def server_port(f, s):
519	p = urlparse(s.get('server', 'http://localhost'))
520	print >>f, 'port: remote:%s:%s:%s' % (p.port, p.hostname, p.port)
521
522	def server_null(f,s): pass
523
524	def server_seer(f, s):
525	print >>f, 'seer: true'
526
527	server_service_out = {
528	'SMB': server_port,
529	'tmcd': server_port,
530	'userconfig': server_null,
531	'project_export': server_null,
532	'seer': server_seer,
533	'seer_master': server_port,
534	'hide_hosts': server_null,
535	}
536	# XXX: end un hardcode this
537
538
539	seer_out = False
540	client_out = False
541	for e in [ e for e in topo.elements \
542	if isinstance(e, topdl.Computer) and e.get_attribute('portal')]:
543	myname = e.name
544	type = e.get_attribute('portal_type')
545
546	info = conninfo_to_dict(myname, connInfo)
547
548	if not info:
549	raise service_error(service_error.req,
550	"No connectivity info for %s" % myname)
551
552	# Translate to physical name (ProtoGENI doesn't have DNS)
553	physname = nodes.get(myname, { }).get('hostname', None)
554	peer = info.get('peer', "")
555	ldomain = self.domain
556	ssh_port = info.get('ssh_port', 22)
557
558	# Collect this for the client.conf file
559	if 'masterexperiment' in info:
560	mproj, meid = info['masterexperiment'].split("/", 1)
561
562	active = info.get('active', 'False')
563
564	if type in ('control', 'both'):
565	testbed = e.get_attribute('testbed')
566	control_gw = myname
567
568	cfn = "%s/%s.gw.conf" % (tmpdir, myname.lower())
569	tunnelconfig = self.tunnel_config
570	try:
571	f = open(cfn, "w")
572	if active == 'True':
573	print >>f, "active: True"
574	print >>f, "ssh_port: %s" % ssh_port
575	if type in ('control', 'both'):
576	for s in [s for s in services \
577	if s.get('name', "") in self.imports]:
578	server_service_out[s['name']](f, s)
579
580	if tunnelconfig:
581	print >>f, "tunnelip: %s" % tunnelconfig
582	print >>f, "peer: %s" % peer.lower()
583	print >>f, "ssh_pubkey: /usr/local/federation/etc/%s" % \
584	pubkey_base
585	print >>f, "ssh_privkey: /usr/local/federation/etc/%s" % \
586	secretkey_base
587	f.close()
588	except EnvironmentError, e:
589	raise service_error(service_error.internal,
590	"Can't write protal config %s: %s" % (cfn, e))
591
592	# Done with portals, write the client config file.
593	try:
594	f = open("%s/client.conf" % tmpdir, "w")
595	if control_gw:
596	print >>f, "ControlGateway: %s" % physname.lower()
597	for s in services:
598	if s.get('name',"") in self.imports and \
599	s.get('visibility','') == 'import':
600	client_service_out[s['name']](f, s)
601	if s.get('name', '') in self.exports and \
602	s.get('visibility', '') == 'export' and \
603	s['name'] in client_export_service_out:
604	client_export_service_out[s['name']](f, s)
605	# Seer uses this.
606	if mproj and meid:
607	print >>f, "ExperimentID: %s/%s" % (mproj, meid)
608	f.close()
609	except EnvironmentError, e:
610	raise service_error(service_error.internal,
611	"Cannot write client.conf: %s" %s)
612
613
614
615	def export_store_info(self, cf, nodes, ssh_port, connInfo):
616	"""
617	For the export requests in the connection info, install the peer names
618	at the experiment controller via SetValue calls.
619	"""
620
621	for c in connInfo:
622	for p in [ p for p in c.get('parameter', []) \
623	if p.get('type', '') == 'output']:
624
625	if p.get('name', '') == 'peer':
626	k = p.get('key', None)
627	surl = p.get('store', None)
628	if surl and k and k.index('/') != -1:
629	if self.create_debug:
630	req = { 'name': k, 'value': 'debug' }
631	self.call_SetValue(surl, req, cf)
632	else:
633	n = nodes.get(k[k.index('/')+1:], { })
634	value = n.get('hostname', None)
635	if value:
636	req = { 'name': k, 'value': value }
637	self.call_SetValue(surl, req, cf)
638	else:
639	self.log.error("No hostname for %s" % \
640	k[k.index('/'):])
641	else:
642	self.log.error("Bad export request: %s" % p)
643	elif p.get('name', '') == 'ssh_port':
644	k = p.get('key', None)
645	surl = p.get('store', None)
646	if surl and k:
647	req = { 'name': k, 'value': ssh_port }
648	self.call_SetValue(surl, req, cf)
649	else:
650	self.log.error("Bad export request: %s" % p)
651	else:
652
653	self.log.error("Unknown export parameter: %s" % \
654	p.get('name'))
655	continue
656
657	def write_node_config_script(self, elem, node, user, pubkey,
658	secretkey, stagingdir, tmpdir):
659	"""
660	Write out the configuration script that is to run on the node
661	represented by elem in the topology. This is called
662	once per node to configure.
663	"""
664	# These little functions/functors just make things more readable. Each
665	# one encapsulates a small task of copying software files or installing
666	# them.
667	class stage_file_type:
668	"""
669	Write code copying file sfrom the staging host to the host on which
670	this will run.
671	"""
672	def __init__(self, user, host, stagingdir):
673	self.user = user
674	self.host = host
675	self.stagingdir = stagingdir
676	self.scp = "/usr/bin/scp -i .ssh/id_rsa -o " + \
677	"'ForwardX11 no' -o 'StrictHostKeyChecking no' "
678
679	def __call__(self, script, file, dest="."):
680	# If the file is a full pathname, do not use stagingdir
681	if file.find('/') == -1:
682	file = "%s/%s" % (self.stagingdir, file)
683	print >>script, "%s %s@%s:%s %s" % \
684	(self.scp, self.user, self.host, file, dest)
685
686	def install_tar(script, loc, base):
687	"""
688	Print code to script to install a tarfile in loc.
689	"""
690	tar = "/bin/tar"
691	mkdir="/bin/mkdir"
692
693	print >>script, "%s -p %s" % (mkdir, loc)
694	print >>script, "%s -C %s -xzf %s" % (tar, loc, base)
695
696	def install_rpm(script, base):
697	"""
698	Print code to script to install an rpm
699	"""
700	rpm = "/bin/rpm"
701	print >>script, "%s --install %s" % (rpm, base)
702
703	ifconfig = "/sbin/ifconfig"
704	stage_file = stage_file_type(user, self.staging_host, stagingdir)
705	pname = node.get('hostname', None)
706	fed_dir = "/usr/local/federation"
707	fed_etc_dir = "%s/etc" % fed_dir
708	fed_bin_dir = "%s/bin" % fed_dir
709	fed_lib_dir = "%s/lib" % fed_dir
710
711	if pname:
712	sfile = "%s/%s.startup" % (tmpdir, pname)
713	script = open(sfile, "w")
714	# Reset the interfaces to the ones in the topo file
715	for i in [ i for i in elem.interface \
716	if not i.get_attribute('portal')]:
717	pinf = node['interfaces'].get(i.name, None)
718	addr = i.get_attribute('ip4_address')
719	netmask = i.get_attribute('ip4_netmask') or '255.255.255.0'
720	if pinf and addr:
721	print >>script, \
722	"%s %s %s netmask %s" % \
723	(ifconfig, pinf, addr, netmask)
724	else:
725	self.log.error("Missing interface or address for %s" \
726	% i.name)
727
728	for l, f in self.federation_software:
729	base = os.path.basename(f)
730	stage_file(script, base)
731	if l: install_tar(script, l, base)
732	else: install_rpm(script, base)
733
734	for s in elem.software:
735	s_base = s.location.rpartition('/')[2]
736	stage_file(script, s_base)
737	if s.install: install_tar(script, s.install, s_base)
738	else: install_rpm(script, s_base)
739
740	for f in ('hosts', pubkey, secretkey, 'client.conf',
741	'userconf'):
742	stage_file(script, f, fed_etc_dir)
743	if self.sshd:
744	stage_file(script, self.sshd, fed_bin_dir)
745	if self.sshd_config:
746	stage_file(script, self.sshd_config, fed_etc_dir)
747
748	# Look in tmpdir to get the names. They've all been copied
749	# into the (remote) staging dir
750	if os.access("%s/%s.gw.conf" % (tmpdir, elem.name), os.R_OK):
751	stage_file(script, "%s.gw.conf" % elem.name, fed_etc_dir)
752
753	# Hackery dackery dock: the ProtoGENI python is really ancient.
754	# A modern version (though packaged for Mandrake (remember
755	# Mandrake? good times, good times)) should be in the
756	# federation_software list, but we need to move rename is for
757	# SEER.
758	print >>script, "rm /usr/bin/python"
759	print >>script, "ln /usr/bin/python2.4 /usr/bin/python"
760	# Back to less hacky stuff
761
762	# Start commands
763	if elem.get_attribute('portal') and self.portal_startcommand:
764	# Install portal software
765	for l, f in self.portal_software:
766	base = os.path.basename(f)
767	stage_file(script, base)
768	if l: install_tar(script, l, base)
769	else: install_rpm(script, base)
770
771	# Portals never have a user-specified start command
772	print >>script, self.portal_startcommand
773	elif self.node_startcommand:
774	# XXX: debug
775	print >>script, "sudo perl -I%s %simport_key.pl /users/%s/.ssh/authorized_keys /root/.ssh/authorized_keys" % (fed_lib_dir, fed_bin_dir, user)
776	# XXX: debug
777	if elem.get_attribute('startup'):
778	print >>script, "%s \\$USER '%s'" % \
779	(self.node_startcommand,
780	elem.get_attribute('startup'))
781	else:
782	print >>script, self.node_startcommand
783	script.close()
784	return sfile, pname
785	else:
786	return None, None
787
788
789	def configure_nodes(self, segment_commands, topo, nodes, user,
790	pubkey, secretkey, stagingdir, tmpdir):
791	"""
792	For each node in the topology, generate a script file that copies
793	software onto it and installs it in the proper places and then runs the
794	startup command (including the federation commands.
795	"""
796
797
798
799	for e in [ e for e in topo.elements if isinstance(e, topdl.Computer)]:
800	vname = e.name
801	sfile, pname = self.write_node_config_script(e,
802	nodes.get(vname, { }),
803	user, pubkey, secretkey, stagingdir, tmpdir)
804	if sfile:
805	if not segment_commands.scp_file(sfile, user, pname):
806	self.log.error("Could not copy script to %s" % pname)
807	else:
808	self.log.error("Unmapped node: %s" % vname)
809
810	def start_node(self, user, host, node, segment_commands):
811	"""
812	Copy an identity to a node for the configuration script to be able to
813	import data and then run the startup script remotely.
814	"""
815	# Place an identity on the node so that the copying can succeed
816	segment_commands.ssh_cmd(user, host, "scp .ssh/id_rsa %s:.ssh" % node)
817	segment_commands.ssh_cmd(user, node,
818	"sudo /bin/sh ./%s.startup &" % node)
819
820	def start_nodes(self, user, host, nodes, segment_commands):
821	"""
822	Start a thread to initialize each node and wait for them to complete.
823	Each thread runs start_node.
824	"""
825	threads = [ ]
826	for n in nodes:
827	t = Thread(target=self.start_node, args=(user, host, n,
828	segment_commands))
829	t.start()
830	threads.append(t)
831
832	done = [not t.isAlive() for t in threads]
833	while not all(done):
834	self.log.info("Waiting for threads %s" % done)
835	time.sleep(10)
836	done = [not t.isAlive() for t in threads]
837
838	def set_up_staging_filespace(self, segment_commands, user, host,
839	stagingdir):
840	"""
841	Set up teh staging area on the staging machine. To reduce the number
842	of ssh commands, we compose a script and execute it remotely.
843	"""
844
845	self.log.info("[start_segment]: creating script file")
846	try:
847	sf, scriptname = tempfile.mkstemp()
848	scriptfile = os.fdopen(sf, 'w')
849	except EnvironmentError:
850	return False
851
852	scriptbase = os.path.basename(scriptname)
853
854	# Script the filesystem changes
855	print >>scriptfile, "/bin/rm -rf %s" % stagingdir
856	print >>scriptfile, 'mkdir -p %s' % stagingdir
857	print >>scriptfile, "rm -f %s" % scriptbase
858	scriptfile.close()
859
860	# Move the script to the remote machine
861	# XXX: could collide tempfile names on the remote host
862	if segment_commands.scp_file(scriptname, user, host, scriptbase):
863	os.remove(scriptname)
864	else:
865	return False
866
867	# Execute the script (and the script's last line deletes it)
868	if not segment_commands.ssh_cmd(user, host, "sh -x %s" % scriptbase):
869	return False
870
871	def initialize_protogeni_context(self, segment_commands, certfile, certpw):
872	"""
873	Protogeni interactions take a context and a protogeni certificate.
874	This establishes both for later calls and returns them.
875	"""
876	if os.access(certfile, os.R_OK):
877	ctxt = fedd_ssl_context(my_cert=certfile, password=certpw)
878	else:
879	self.log.error("[start_segment]: Cannot read certfile: %s" % \
880	certfile)
881	return None, None
882
883	try:
884	gcred = segment_commands.pg_call(self.sa_url,
885	'GetCredential', {}, ctxt)
886	except self.ProtoGENIError, e:
887	raise service_error(service_error.federant,
888	"ProtoGENI: %s" % e)
889
890	return ctxt, gcred
891
892	def get_free_slicename(self, segment_commands, user, gcred, ctxt):
893	"""
894	Find a usable slice name by trying random ones until there's no
895	collision.
896	"""
897
898	def random_slicename(user):
899	"""
900	Return a random slicename by appending 5 letters to the username.
901	"""
902	slicename = user
903	for i in range(0,5):
904	slicename += random.choice(string.ascii_letters)
905	return slicename
906
907	while True:
908	slicename = random_slicename(user)
909	try:
910	param = {
911	'credential': gcred,
912	'hrn': slicename,
913	'type': 'Slice'
914	}
915	segment_commands.pg_call(self.sa_url, 'Resolve', param, ctxt)
916	except segment_commands.ProtoGENIError, e:
917	print e
918	break
919
920	return slicename
921
922	def allocate_slice(self, segment_commands, slicename, rspec, gcred, ctxt):
923	"""
924	Create the slice and allocate resources. If any of this stuff fails,
925	the allocations will time out on PG in short order, so we just raise
926	the service_error. Return the slice and sliver credentials as well as
927	the manifest.
928	"""
929	try:
930	param = {
931	'credential': gcred,
932	'hrn': slicename,
933	'type': 'Slice'
934	}
935	slice_cred = segment_commands.pg_call(self.sa_url, 'Register',
936	param, ctxt)
937	f = open("./slice_cred", "w")
938	print >>f, slice_cred
939	f.close()
940	# Populate the ssh keys (let PG format them)
941	param = {
942	'credential': gcred,
943	}
944	keys = segment_commands.pg_call(self.sa_url, 'GetKeys', param,
945	ctxt)
946	# Grab and redeem a ticket
947	param = {
948	'credential': slice_cred,
949	'rspec': rspec,
950	}
951	ticket = segment_commands.pg_call(self.cm_url, 'GetTicket', param,
952	ctxt)
953	f = open("./ticket", "w")
954	print >>f, ticket
955	f.close()
956	param = {
957	'credential': slice_cred,
958	'keys': keys,
959	'ticket': ticket,
960	}
961	sliver_cred, manifest = segment_commands.pg_call(self.cm_url,
962	'RedeemTicket', param, ctxt)
963	f = open("./sliver_cred", "w")
964	print >>f, sliver_cred
965	f.close()
966	f = open("./manifest", "w")
967	print >>f, manifest
968	f.close()
969	# start 'em up
970	param = {
971	'credential': sliver_cred,
972	}
973	segment_commands.pg_call(self.cm_url, 'StartSliver', param, ctxt)
974	except segment_commands.ProtoGENIError, e:
975	raise service_error(service_error.federant,
976	"ProtoGENI: %s %s" % (e.code, e))
977
978	return (slice_cred, sliver_cred, manifest)
979
980	def wait_for_slice(self, segment_commands, slice_cred, ctxt):
981	"""
982	Wait for the given slice to finish its startup. Return the final
983	status.
984	"""
985	status = 'notready'
986	try:
987	while status == 'notready':
988	param = {
989	'credential': slice_cred
990	}
991	r = segment_commands.pg_call(self.cm_url,
992	'SliceStatus', param, ctxt)
993	status = r.get('status', 'notready')
994	if status == 'notready':
995	time.sleep(30)
996	except segment_commands.ProtoGENIError, e:
997	raise service_error(service_error.federant,
998	"ProtoGENI: %s %s" % (e.code, e))
999
1000	return status
1001
1002	def delete_slice(self, segment_commands, slice_cred, ctxt):
1003	"""
1004	Delete the slice resources. An error from the service is ignores,
1005	because the soft state will go away anyway.
1006	"""
1007	try:
1008	param = { 'credential': slice_cred }
1009	segment_commands.pg_call(self.cm_url, 'DeleteSliver',
1010	param, ctxt)
1011	except segment_commands.ProtoGENIError, e:
1012	self.log.warn("ProtoGENI: %s" % e)
1013
1014
1015
1016	def start_segment(self, segment_commands, aid, user, rspec, pubkey,
1017	secretkey, ename, stagingdir, tmpdir, certfile, certpw,
1018	export_certfile, topo, connInfo, services, timeout=0):
1019	"""
1020	Start a sub-experiment on a federant.
1021
1022	Get the current state, modify or create as appropriate, ship data
1023	and configs and start the experiment. There are small ordering
1024	differences based on the initial state of the sub-experiment.
1025	"""
1026
1027	# Local software dir
1028	lsoftdir = "%s/software" % tmpdir
1029	host = self.staging_host
1030
1031	ctxt, gcred = self.initialize_protogeni_context(segment_commands,
1032	certfile, certpw)
1033
1034	if not ctxt: return False
1035
1036	self.set_up_staging_filespace(segment_commands, user, host, stagingdir)
1037	slicename = self.get_free_slicename(segment_commands, user, gcred, ctxt)
1038	self.log.info("Creating %s" % slicename)
1039	slice_cred, sliver_cred, manifest = self.allocate_slice(
1040	segment_commands, slicename, rspec, gcred, ctxt)
1041
1042	# With manifest in hand, we can export the portal node names.
1043	if self.create_debug: nodes = self.fake_manifest(topo)
1044	else: nodes = self.manifest_to_dict(manifest)
1045
1046	self.export_store_info(export_certfile, nodes, self.ssh_port,
1047	connInfo)
1048	self.generate_portal_configs(topo, pubkey, secretkey, tmpdir,
1049	ename, connInfo, services, nodes)
1050
1051	# Copy software to the staging machine (done after generation to copy
1052	# those, too)
1053	for d in (tmpdir, lsoftdir):
1054	if os.path.isdir(d):
1055	for f in os.listdir(d):
1056	if not os.path.isdir("%s/%s" % (d, f)):
1057	if not segment_commands.scp_file("%s/%s" % (d, f),
1058	user, host, "%s/%s" % (stagingdir, f)):
1059	self.log.error("Scp failed")
1060	return False
1061
1062
1063	# Now we wait for the nodes to start on PG
1064	status = self.wait_for_slice(segment_commands, slice_cred, ctxt)
1065	if status == 'failed':
1066	self.log.error('Sliver failed to start on ProtoGENI')
1067	self.delete_slice(segment_commands, slice_cred, ctxt)
1068	return False
1069	else:
1070	# All good: save ProtoGENI info in shared state
1071	self.state_lock.acquire()
1072	self.allocation[aid]['slice_name'] = slicename
1073	self.allocation[aid]['slice_credential'] = slice_cred
1074	self.allocation[aid]['sliver_credential'] = sliver_cred
1075	self.allocation[aid]['manifest'] = manifest
1076	self.allocation[aid]['certfile'] = certfile
1077	self.allocation[aid]['certpw'] = certpw
1078	self.write_state()
1079	self.state_lock.release()
1080
1081	# Now we have configuration to do for ProtoGENI
1082	self.configure_nodes(segment_commands, topo, nodes, user, pubkey,
1083	secretkey, stagingdir, tmpdir)
1084
1085	self.start_nodes(user, self.staging_host,
1086	[ n.get('hostname', None) for n in nodes.values()],
1087	segment_commands)
1088
1089	# Everything has gone OK.
1090	return True, dict([(k, n.get('hostname', None)) \
1091	for k, n in nodes.items()])
1092
1093	def generate_rspec(self, topo, softdir, connInfo):
1094	t = topo.clone()
1095
1096	starts = { }
1097	# Weed out the things we aren't going to instantiate: Segments, portal
1098	# substrates, and portal interfaces. (The copy in the for loop allows
1099	# us to delete from e.elements in side the for loop). While we're
1100	# touching all the elements, we also adjust paths from the original
1101	# testbed to local testbed paths and put the federation commands and
1102	# startcommands into a dict so we can start them manually later.
1103	# ProtoGENI requires setup before the federation commands run, so we
1104	# run them by hand after we've seeded configurations.
1105	for e in [e for e in t.elements]:
1106	if isinstance(e, topdl.Segment):
1107	t.elements.remove(e)
1108	# Fix software paths
1109	for s in getattr(e, 'software', []):
1110	s.location = re.sub("^.*/", softdir, s.location)
1111	if isinstance(e, topdl.Computer):
1112	if e.get_attribute('portal') and self.portal_startcommand:
1113	# Portals never have a user-specified start command
1114	starts[e.name] = self.portal_startcommand
1115	elif self.node_startcommand:
1116	if e.get_attribute('startup'):
1117	starts[e.name] = "%s \\$USER '%s'" % \
1118	(self.node_startcommand,
1119	e.get_attribute('startup'))
1120	e.remove_attribute('startup')
1121	else:
1122	starts[e.name] = self.node_startcommand
1123
1124	# Remove portal interfaces
1125	e.interface = [i for i in e.interface \
1126	if not i.get_attribute('portal')]
1127
1128	t.substrates = [ s.clone() for s in t.substrates ]
1129	t.incorporate_elements()
1130
1131	# Customize the ns2 output for local portal commands and images
1132	filters = []
1133
1134	# NB: these are extra commands issued for the node, not the startcmds
1135	if self.portal_command:
1136	filters.append(topdl.generate_portal_command_filter(
1137	self.portal_command))
1138
1139	# Convert to rspec and return it
1140	exp_rspec = topdl.topology_to_rspec(t, filters)
1141
1142	return exp_rspec
1143
1144	def retrieve_software(self, topo, certfile, softdir):
1145	"""
1146	Collect the software that nodes in the topology need loaded and stage
1147	it locally. This implies retrieving it from the experiment_controller
1148	and placing it into softdir. Certfile is used to prove that this node
1149	has access to that data (it's the allocation/segment fedid). Finally
1150	local portal and federation software is also copied to the same staging
1151	directory for simplicity - all software needed for experiment creation
1152	is in softdir.
1153	"""
1154	sw = set()
1155	for e in topo.elements:
1156	for s in getattr(e, 'software', []):
1157	sw.add(s.location)
1158	os.mkdir(softdir)
1159	for s in sw:
1160	self.log.debug("Retrieving %s" % s)
1161	try:
1162	get_url(s, certfile, softdir)
1163	except:
1164	t, v, st = sys.exc_info()
1165	raise service_error(service_error.internal,
1166	"Error retrieving %s: %s" % (s, v))
1167
1168	# Copy local portal node software to the tempdir
1169	for s in (self.portal_software, self.federation_software):
1170	for l, f in s:
1171	base = os.path.basename(f)
1172	copy_file(f, "%s/%s" % (softdir, base))
1173
1174	# Ick. Put this python rpm in a place that it will get moved into
1175	# the staging area. It's a hack to install a modern (in a Roman
1176	# sense of modern) python on ProtoGENI
1177	python_rpm ="python2.4-2.4-1pydotorg.i586.rpm"
1178	if os.access("./%s" % python_rpm, os.R_OK):
1179	copy_file("./%s" % python_rpm, "%s/%s" % (softdir, python_rpm))
1180
1181
1182	def initialize_experiment_info(self, attrs, aid, certfile, tmpdir):
1183	"""
1184	Gather common configuration files, retrieve or create an experiment
1185	name and project name, and return the ssh_key filenames. Create an
1186	allocation log bound to the state log variable as well.
1187	"""
1188	configs = set(('hosts', 'ssh_pubkey', 'ssh_secretkey'))
1189	ename = None
1190	pubkey_base = None
1191	secretkey_base = None
1192	alloc_log = None
1193
1194	for a in attrs:
1195	if a['attribute'] in configs:
1196	try:
1197	self.log.debug("Retrieving %s" % a['value'])
1198	get_url(a['value'], certfile, tmpdir)
1199	except:
1200	t, v, st = sys.exc_info()
1201	raise service_error(service_error.internal,
1202	"Error retrieving %s: %s" % (a.get('value', ""), v))
1203	if a['attribute'] == 'ssh_pubkey':
1204	pubkey_base = a['value'].rpartition('/')[2]
1205	if a['attribute'] == 'ssh_secretkey':
1206	secretkey_base = a['value'].rpartition('/')[2]
1207	if a['attribute'] == 'experiment_name':
1208	ename = a['value']
1209
1210	if not ename:
1211	ename = ""
1212	for i in range(0,5):
1213	ename += random.choice(string.ascii_letters)
1214	self.log.warn("No experiment name: picked one randomly: %s" \
1215	% ename)
1216
1217	self.state_lock.acquire()
1218	if self.allocation.has_key(aid):
1219	cf, user, ssh_key, cpw = self.allocation[aid]['credentials']
1220	self.allocation[aid]['experiment'] = ename
1221	self.allocation[aid]['log'] = [ ]
1222	# Create a logger that logs to the experiment's state object as
1223	# well as to the main log file.
1224	alloc_log = logging.getLogger('fedd.access.%s' % ename)
1225	h = logging.StreamHandler(
1226	list_log.list_log(self.allocation[aid]['log']))
1227	# XXX: there should be a global one of these rather than
1228	# repeating the code.
1229	h.setFormatter(logging.Formatter(
1230	"%(asctime)s %(name)s %(message)s",
1231	'%d %b %y %H:%M:%S'))
1232	alloc_log.addHandler(h)
1233	self.write_state()
1234	else:
1235	self.log.error("No allocation for %s!?" % aid)
1236	self.state_lock.release()
1237
1238	return (ename, pubkey_base, secretkey_base, cf, user, ssh_key,
1239	cpw, alloc_log)
1240
1241	def finalize_experiment(self, topo, nodes, aid, alloc_id):
1242	# Copy the assigned names into the return topology
1243	rvtopo = topo.clone()
1244	embedding = [ ]
1245	for k, n in nodes.items():
1246	embedding.append({
1247	'toponame': k,
1248	'physname': ["%s%s" % (n, self.domain)],
1249	})
1250	# Grab the log (this is some anal locking, but better safe than
1251	# sorry)
1252	self.state_lock.acquire()
1253	logv = "".join(self.allocation[aid]['log'])
1254	# It's possible that the StartSegment call gets retried (!).
1255	# if the 'started' key is in the allocation, we'll return it rather
1256	# than redo the setup.
1257	self.allocation[aid]['started'] = {
1258	'allocID': alloc_id,
1259	'allocationLog': logv,
1260	'segmentdescription': {
1261	'topdldescription': rvtopo.to_dict() },
1262	'embedding': embedding,
1263	}
1264	retval = copy.deepcopy(self.allocation[aid]['started'])
1265	self.write_state()
1266	self.state_lock.release()
1267
1268	return retval
1269
1270	def StartSegment(self, req, fid):
1271	err = None # Any service_error generated after tmpdir is created
1272	rv = None # Return value from segment creation
1273
1274	try:
1275	req = req['StartSegmentRequestBody']
1276	topref = req['segmentdescription']['topdldescription']
1277	except KeyError:
1278	raise service_error(service_error.req, "Badly formed request")
1279
1280	connInfo = req.get('connection', [])
1281	services = req.get('service', [])
1282	auth_attr = req['allocID']['fedid']
1283	aid = "%s" % auth_attr
1284	attrs = req.get('fedAttr', [])
1285	if not self.auth.check_attribute(fid, auth_attr):
1286	raise service_error(service_error.access, "Access denied")
1287	else:
1288	# See if this is a replay of an earlier succeeded StartSegment -
1289	# sometimes SSL kills 'em. If so, replay the response rather than
1290	# redoing the allocation.
1291	self.state_lock.acquire()
1292	retval = self.allocation[aid].get('started', None)
1293	self.state_lock.release()
1294	if retval:
1295	self.log.warning("Duplicate StartSegment for %s: " % aid + \
1296	"replaying response")
1297	return retval
1298
1299	if topref:
1300	topo = topdl.Topology(**topref)
1301	else:
1302	raise service_error(service_error.req,
1303	"Request missing segmentdescription'")
1304
1305	certfile = "%s/%s.pem" % (self.certdir, auth_attr)
1306	try:
1307	tmpdir = tempfile.mkdtemp(prefix="access-")
1308	softdir = "%s/software" % tmpdir
1309	except EnvironmentError:
1310	raise service_error(service_error.internal, "Cannot create tmp dir")
1311
1312	# Try block alllows us to clean up temporary files.
1313	try:
1314	self.retrieve_software(topo, certfile, softdir)
1315	self.configure_userconf(services, tmpdir)
1316	ename, pubkey_base, secretkey_base, cf, user, ssh_key, \
1317	cpw, alloc_log = self.initialize_experiment_info(attrs,
1318	aid, certfile, tmpdir)
1319	self.import_store_info(certfile, connInfo)
1320	rspec = self.generate_rspec(topo, "%s/%s/" \
1321	% (self.staging_dir, ename), connInfo)
1322
1323	segment_commands = protogeni_proxy(keyfile=ssh_key,
1324	debug=self.create_debug, log=alloc_log,
1325	ch_url = self.ch_url, sa_url=self.sa_url,
1326	cm_url=self.cm_url)
1327	rv, nodes = self.start_segment(segment_commands, aid, user, rspec,
1328	pubkey_base,
1329	secretkey_base, ename,
1330	"%s/%s" % (self.staging_dir, ename), tmpdir, cf, cpw,
1331	certfile, topo, connInfo, services)
1332	except EnvironmentError, e:
1333	err = service_error(service_error.internal, "%s" % e)
1334	except service_error, e:
1335	err = e
1336	except:
1337	t, v, st = sys.exc_info()
1338	err = service_error(service_error.internal, "%s: %s" % \
1339	(v, traceback.extract_tb(st)))
1340
1341	# Walk up tmpdir, deleting as we go
1342	if self.cleanup: self.remove_dirs(tmpdir)
1343	else: self.log.debug("[StartSegment]: not removing %s" % tmpdir)
1344
1345	if rv:
1346	return self.finalize_experiment(topo, nodes, aid, req['allocID'])
1347	elif err:
1348	raise service_error(service_error.federant,
1349	"Swapin failed: %s" % err)
1350	else:
1351	raise service_error(service_error.federant, "Swapin failed")
1352
1353	def stop_segment(self, segment_commands, user, stagingdir, slice_cred,
1354	certfile, certpw):
1355	"""
1356	Stop a sub experiment by calling swapexp on the federant
1357	"""
1358	host = self.staging_host
1359	rv = False
1360	try:
1361	# Clean out tar files: we've gone over quota in the past
1362	if stagingdir:
1363	segment_commands.ssh_cmd(user, host, "rm -rf %s" % stagingdir)
1364	if slice_cred:
1365	self.log.error('Removing Sliver on ProtoGENI')
1366	ctxt = fedd_ssl_context(my_cert=certfile, password=certpw)
1367	self.delete_slice(segment_commands, slice_cred, ctxt)
1368	return True
1369	except self.ssh_cmd_timeout:
1370	rv = False
1371	return rv
1372
1373	def TerminateSegment(self, req, fid):
1374	try:
1375	req = req['TerminateSegmentRequestBody']
1376	except KeyError:
1377	raise service_error(service_error.req, "Badly formed request")
1378
1379	auth_attr = req['allocID']['fedid']
1380	aid = "%s" % auth_attr
1381	attrs = req.get('fedAttr', [])
1382	if not self.auth.check_attribute(fid, auth_attr):
1383	raise service_error(service_error.access, "Access denied")
1384
1385	self.state_lock.acquire()
1386	if self.allocation.has_key(aid):
1387	cf, user, ssh_key, cpw = self.allocation[aid]['credentials']
1388	slice_cred = self.allocation[aid].get('slice_credential', None)
1389	ename = self.allocation[aid].get('experiment', None)
1390	else:
1391	cf, user, ssh_key, cpw = (None, None, None, None)
1392	slice_cred = None
1393	ename = None
1394	self.state_lock.release()
1395
1396	if ename:
1397	staging = "%s/%s" % ( self.staging_dir, ename)
1398	else:
1399	self.log.warn("Can't find experiment name for %s" % aid)
1400	staging = None
1401
1402	segment_commands = protogeni_proxy(keyfile=ssh_key,
1403	debug=self.create_debug, ch_url = self.ch_url,
1404	sa_url=self.sa_url, cm_url=self.cm_url)
1405	self.stop_segment(segment_commands, user, staging, slice_cred, cf, cpw)
1406	return { 'allocID': req['allocID'] }
1407
1408	def renew_segment(self, segment_commands, name, scred, interval,
1409	certfile, certpw):
1410	"""
1411	Linear code through the segment renewal calls.
1412	"""
1413	ctxt = fedd_ssl_context(my_cert=certfile, password=certpw)
1414	try:
1415	expiration = time.strftime("%Y%m%dT%H:%M:%S",
1416	time.gmtime(time.time() + interval))
1417	cred = segment_commands.pg_call(self.sa_url, 'GetCredential',
1418	{}, ctxt)
1419
1420	param = {
1421	'credential': scred,
1422	'expiration': expiration
1423	}
1424	r = segment_commands.pg_call(self.sa_url, 'RenewSlice', param, ctxt)
1425	param = {
1426	'credential': cred,
1427	'hrn': name,
1428	'type': 'Slice',
1429	}
1430	slice = segment_commands.pg_call(self.sa_url, 'Resolve',
1431	param, ctxt)
1432	uuid = slice.get('uuid', None)
1433	if uuid == None:
1434	sys.exit('No uuid for %s' % slicename)
1435
1436	print 'Calling GetCredential (uuid)'
1437	param = {
1438	'credential': cred,
1439	'uuid': uuid,
1440	'type': 'Slice',
1441	}
1442	new_scred = segment_commands.pg_call(self.sa_url, 'GetCredential',
1443	param, ctxt)
1444	f = open('./new_slice_cred', 'w')
1445	print >>f, new_scred
1446	f.close()
1447
1448	except segment_commands.ProtoGENIError, e:
1449	self.log.error("Failed to extend slice %s: %s" % (name, e))
1450	return None
1451	try:
1452	print 'Calling RenewSlice (CM)'
1453	param = {
1454	'credential': new_scred,
1455	}
1456	r = segment_commands.pg_call(self.cm_url, 'RenewSlice', param, ctxt)
1457	except segment_commands.ProtoGENIError, e:
1458	self.log.warn("Failed to renew sliver for %s: %s" % (name, e))
1459
1460	return new_scred
1461
1462
1463	def RenewSlices(self):
1464	self.log.info("Scanning for slices to renew")
1465	self.state_lock.acquire()
1466	aids = self.allocation.keys()
1467	self.state_lock.release()
1468
1469	for aid in aids:
1470	self.state_lock.acquire()
1471	if self.allocation.has_key(aid):
1472	name = self.allocation[aid].get('slice_name', None)
1473	scred = self.allocation[aid].get('slice_credential', None)
1474	cf, user, ssh_key, cpw = self.allocation[aid]['credentials']
1475	else:
1476	name = None
1477	scred = None
1478	self.state_lock.release()
1479
1480	if not os.access(cf, os.R_OK):
1481	self.log.error(
1482	"[RenewSlices] cred.file %s unreadable, ignoring" % cf)
1483	continue
1484
1485	# There's a ProtoGENI slice associated with the segment; renew it.
1486	if name and scred:
1487	segment_commands = protogeni_proxy(log=self.log,
1488	debug=self.create_debug, keyfile=ssh_key,
1489	cm_url = self.cm_url, sa_url = self.sa_url,
1490	ch_url = self.ch_url)
1491	new_scred = self.renew_segment(segment_commands, name, scred,
1492	self.renewal_interval, cf, cpw)
1493	if new_scred:
1494	self.log.info("Slice %s renewed until %s GMT" % \
1495	(name, time.asctime(time.gmtime(\
1496	time.time()+self.renewal_interval))))
1497	self.state_lock.acquire()
1498	if self.allocation.has_key(aid):
1499	self.allocation[aid]['slice_credential'] = new_scred
1500	self.state_lock.release()
1501	else:
1502	self.log.info("Failed to renew slice %s " % name)
1503
1504	# Let's do this all again soon. (4 tries before the slices time out)
1505	t = Timer(self.renewal_interval/4, self.RenewSlices)
1506	t.start()

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

Download in other formats: