Context Navigation

source: fedd/federation/protogeni_access.py @ 310d419

axis_examplecompt_changesinfo-opsversion-3.01version-3.02 GEC7

Last change on this file since 310d419 was 310d419, checked in by Ted Faber <faber@…>, 14 years ago
Renewal interval in seconds
Property mode set to `100644`
File size: 39.9 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
13	from threading import *
14	from M2Crypto.SSL import SSLError
15
16	from util import *
17	from access_project import access_project
18	from fedid import fedid, generate_fedid
19	from authorizer import authorizer
20	from service_error import service_error
21	from remote_service import xmlrpc_handler, soap_handler, service_caller
22
23	import httplib
24	import tempfile
25	from urlparse import urlparse
26
27	import topdl
28	import list_log
29	import proxy_protogeni_segment
30
31
32	# Make log messages disappear if noone configures a fedd logger
33	class nullHandler(logging.Handler):
34	def emit(self, record): pass
35
36	fl = logging.getLogger("fedd.access")
37	fl.addHandler(nullHandler())
38
39	class access:
40	"""
41	The implementation of access control based on mapping users to projects.
42
43	Users can be mapped to existing projects or have projects created
44	dynamically. This implements both direct requests and proxies.
45	"""
46
47	class parse_error(RuntimeError): pass
48
49
50	proxy_RequestAccess= service_caller('RequestAccess')
51	proxy_ReleaseAccess= service_caller('ReleaseAccess')
52
53	def __init__(self, config=None, auth=None):
54	"""
55	Initializer. Pulls parameters out of the ConfigParser's access section.
56	"""
57
58	def software_list(v):
59	l = [ ]
60	if v:
61	ps = v.split(" ")
62	while len(ps):
63	loc, file = ps[0:2]
64	del ps[0:2]
65	l.append((loc, file))
66	return l
67
68	# Make sure that the configuration is in place
69	if not config:
70	raise RunTimeError("No config to fedd.access")
71
72	self.project_priority = config.getboolean("access", "project_priority")
73	self.allow_proxy = config.getboolean("access", "allow_proxy")
74
75	self.domain = config.get("access", "domain")
76	self.certdir = config.get("access","certdir")
77	self.userconfdir = config.get("access","userconfdir")
78	self.userconfcmd = config.get("access","userconfcmd")
79	self.userconfurl = config.get("access","userconfurl")
80	self.federation_software = config.get("access", "federation_software")
81	self.portal_software = config.get("access", "portal_software")
82	self.ssh_port = config.get("access","ssh_port") or "22"
83	self.sshd = config.get("access","sshd")
84	self.sshd_config = config.get("access", "sshd_config")
85	self.create_debug = config.getboolean("access", "create_debug")
86	self.cleanup = not config.getboolean("access", "leave_tmpfiles")
87	self.access_type = config.get("access", "type")
88	self.staging_dir = config.get("access", "staging_dir") or "/tmp"
89	self.staging_host = config.get("access", "staging_host") \
90	or "ops.emulab.net"
91
92	self.federation_software = software_list(self.federation_software)
93	self.portal_software = software_list(self.portal_software)
94
95	self.renewal_interval = config.get("access", "renewal") or (3 * 60 )
96	self.renewal_interval = int(self.renewal_interval) * 60
97
98	self.ch_url = config.get("access", "ch_url")
99	self.sa_url = config.get("access", "sa_url")
100	self.cm_url = config.get("access", "cm_url")
101
102	self.attrs = { }
103	self.access = { }
104	self.restricted = [ ]
105	self.projects = { }
106	self.keys = { }
107	self.types = { }
108	self.allocation = { }
109	self.state = {
110	'projects': self.projects,
111	'allocation' : self.allocation,
112	'keys' : self.keys,
113	'types': self.types
114	}
115	self.log = logging.getLogger("fedd.access")
116	set_log_level(config, "access", self.log)
117	self.state_lock = Lock()
118	# XXX: Configurable
119	self.exports = set(('SMB', 'seer', 'tmcd', 'userconfig'))
120	self.imports = set(('SMB', 'seer', 'userconfig'))
121
122	if auth: self.auth = auth
123	else:
124	self.log.error(\
125	"[access]: No authorizer initialized, creating local one.")
126	auth = authorizer()
127
128	tb = config.get('access', 'testbed')
129	if tb: self.testbed = [ t.strip() for t in tb.split(',') ]
130	else: self.testbed = [ ]
131
132	if config.has_option("access", "accessdb"):
133	self.read_access(config.get("access", "accessdb"))
134
135	self.state_filename = config.get("access", "access_state")
136	print "Calling read_state %s" % self.state_filename
137	self.read_state()
138
139	# Keep cert_file and cert_pwd coming from the same place
140	self.cert_file = config.get("access", "cert_file")
141	if self.cert_file:
142	self.sert_pwd = config.get("access", "cert_pw")
143	else:
144	self.cert_file = config.get("globals", "cert_file")
145	self.sert_pwd = config.get("globals", "cert_pw")
146
147	self.trusted_certs = config.get("access", "trusted_certs") or \
148	config.get("globals", "trusted_certs")
149
150	self.start_segment = proxy_protogeni_segment.start_segment
151	self.stop_segment = proxy_protogeni_segment.stop_segment
152	self.renew_segment = proxy_protogeni_segment.renew_segment
153
154	self.call_SetValue = service_caller('SetValue')
155	self.call_GetValue = service_caller('GetValue')
156
157	self.RenewSlices()
158
159	self.soap_services = {\
160	'RequestAccess': soap_handler("RequestAccess", self.RequestAccess),
161	'ReleaseAccess': soap_handler("ReleaseAccess", self.ReleaseAccess),
162	'StartSegment': soap_handler("StartSegment", self.StartSegment),
163	'TerminateSegment': soap_handler("TerminateSegment",
164	self.TerminateSegment),
165	}
166	self.xmlrpc_services = {\
167	'RequestAccess': xmlrpc_handler('RequestAccess',
168	self.RequestAccess),
169	'ReleaseAccess': xmlrpc_handler('ReleaseAccess',
170	self.ReleaseAccess),
171	'StartSegment': xmlrpc_handler("StartSegment", self.StartSegment),
172	'TerminateSegment': xmlrpc_handler('TerminateSegment',
173	self.TerminateSegment),
174	}
175
176	def read_access(self, config):
177	"""
178	Read a configuration file and set internal parameters.
179
180	There are access lines of the
181	form (tb, proj, user) -> user that map the first tuple of
182	names to the user for for access purposes. Names in the key (left side)
183	can include "<NONE> or <ANY>" to act as wildcards or to require the
184	fields to be empty. Similarly aproj or auser can be <SAME> or
185	<DYNAMIC> indicating that either the matching key is to be used or a
186	dynamic user or project will be created. These names can also be
187	federated IDs (fedid's) if prefixed with fedid:. The user is the
188	ProtoGENI identity certificate.
189	Testbed attributes outside the forms above can be given using the
190	format attribute: name value: value. The name is a single word and the
191	value continues to the end of the line. Empty lines and lines startin
192	with a # are ignored.
193
194	Parsing errors result in a self.parse_error exception being raised.
195	"""
196	lineno=0
197	name_expr = "["+string.ascii_letters + string.digits + "\/\.\-_]+"
198	fedid_expr = "fedid:[" + string.hexdigits + "]+"
199	key_name = "(<ANY>\|<NONE>\|"+fedid_expr + "\|"+ name_expr + ")"
200
201	attr_re = re.compile('attribute:\s([\._\-a-z0-9]+)\s+value:\s(.*)',
202	re.IGNORECASE)
203	access_str = '\('+key_name+'\s,\s'+key_name+'\s,\s'+ \
204	key_name+'\s\)\s->\s\(('+name_expr +')\s,\s*('\
205	+ name_expr + ')\s,\s('+name_expr+')\s,?\s(' + \
206	name_expr+ ')?\)'
207	access_re = re.compile(access_str, re.IGNORECASE)
208
209
210	def parse_name(n):
211	if n.startswith('fedid:'): return fedid(hexstr=n[len('fedid:'):])
212	else: return n
213
214	def auth_name(n):
215	if isinstance(n, basestring):
216	if n =='<any>' or n =='<none>': return None
217	else: return unicode(n)
218	else:
219	return n
220
221	f = open(config, "r");
222	for line in f:
223	lineno += 1
224	line = line.strip();
225	if len(line) == 0 or line.startswith('#'):
226	continue
227
228	# Extended (attribute: x value: y) attribute line
229	m = attr_re.match(line)
230	if m != None:
231	attr, val = m.group(1,2)
232	self.attrs[attr] = val
233	continue
234
235	# Access line (t, p, u) -> (a, pw) line
236	m = access_re.match(line)
237	if m != None:
238	access_key = tuple([ parse_name(x) for x in m.group(1,2,3)])
239	auth_key = tuple([ auth_name(x) for x in access_key])
240	cert = auth_name(parse_name(m.group(4)))
241	user_name = auth_name(parse_name(m.group(5)))
242	ssh_key = unicode(m.group(6))
243	if m.group(6): pw = unicode(m.group(7))
244	else: pw = None
245
246	self.access[access_key] = (cert, user_name, ssh_key, pw)
247	self.auth.set_attribute(auth_key, "access")
248	continue
249
250	# Nothing matched to here: unknown line - raise exception
251	f.close()
252	raise self.parse_error("Unknown statement at line %d of %s" % \
253	(lineno, config))
254	f.close()
255
256	# need this ?
257	def get_users(self, obj):
258	"""
259	Return a list of the IDs of the users in dict
260	"""
261	if obj.has_key('user'):
262	return [ unpack_id(u['userID']) \
263	for u in obj['user'] if u.has_key('userID') ]
264	else:
265	return None
266	# need this ?
267
268	def write_state(self):
269	if self.state_filename:
270	try:
271	f = open(self.state_filename, 'w')
272	pickle.dump(self.state, f)
273	except IOError, e:
274	self.log.error("Can't write file %s: %s" % \
275	(self.state_filename, e))
276	except pickle.PicklingError, e:
277	self.log.error("Pickling problem: %s" % e)
278	except TypeError, e:
279	self.log.error("Pickling problem (TypeError): %s" % e)
280
281
282	def read_state(self):
283	"""
284	Read a new copy of access state. Old state is overwritten.
285
286	State format is a simple pickling of the state dictionary.
287	"""
288	if self.state_filename:
289	try:
290	f = open(self.state_filename, "r")
291	self.state = pickle.load(f)
292	self.log.debug("[read_state]: Read state from %s" % \
293	self.state_filename)
294	except IOError, e:
295	self.log.warning(("[read_state]: No saved state: " +\
296	"Can't open %s: %s") % (self.state_filename, e))
297	except EOFError, e:
298	self.log.warning(("[read_state]: " +\
299	"Empty or damaged state file: %s:") % \
300	self.state_filename)
301	except pickle.UnpicklingError, e:
302	self.log.warning(("[read_state]: No saved state: " + \
303	"Unpickling failed: %s") % e)
304
305	# Add the ownership attributes to the authorizer. Note that the
306	# indices of the allocation dict are strings, but the attributes are
307	# fedids, so there is a conversion.
308	for k in self.state.get('allocation', {}).keys():
309	for o in self.state['allocation'][k].get('owners', []):
310	self.auth.set_attribute(o, fedid(hexstr=k))
311	self.auth.set_attribute(fedid(hexstr=k),fedid(hexstr=k))
312
313	if self.allocation != self.state['allocation']:
314	self.allocation = self.state['allocation']
315
316	def permute_wildcards(self, a, p):
317	"""Return a copy of a with various fields wildcarded.
318
319	The bits of p control the wildcards. A set bit is a wildcard
320	replacement with the lowest bit being user then project then testbed.
321	"""
322	if p & 1: user = ["<any>"]
323	else: user = a[2]
324	if p & 2: proj = "<any>"
325	else: proj = a[1]
326	if p & 4: tb = "<any>"
327	else: tb = a[0]
328
329	return (tb, proj, user)
330
331
332	def find_access(self, search):
333	"""
334	Search the access DB for a match on this tuple. Return the matching
335	user (protoGENI cert).
336
337	NB, if the initial tuple fails to match we start inserting wildcards in
338	an order determined by self.project_priority. Try the list of users in
339	order (when wildcarded, there's only one user in the list).
340	"""
341	if self.project_priority: perm = (0, 1, 2, 3, 4, 5, 6, 7)
342	else: perm = (0, 2, 1, 3, 4, 6, 5, 7)
343
344	for p in perm:
345	s = self.permute_wildcards(search, p)
346	# s[2] is None on an anonymous, unwildcarded request
347	if s[2] != None:
348	for u in s[2]:
349	if self.access.has_key((s[0], s[1], u)):
350	return self.access[(s[0], s[1], u)]
351	else:
352	if self.access.has_key(s):
353	return self.access[s]
354	return None
355
356	def lookup_access(self, req, fid):
357	"""
358	Determine the allowed access for this request. Return the access and
359	which fields are dynamic.
360
361	The fedid is needed to construct the request
362	"""
363	# Search keys
364	tb = None
365	project = None
366	user = None
367	# Return values
368	rp = access_project(None, ())
369	ru = None
370	user_re = re.compile("user:\s(.*)")
371	project_re = re.compile("project:\s(.*)")
372
373	user = [ user_re.findall(x)[0] for x in req.get('credential', []) \
374	if user_re.match(x)]
375	project = [ project_re.findall(x)[0] \
376	for x in req.get('credential', []) \
377	if project_re.match(x)]
378
379	if len(project) == 1: project = project[0]
380	elif len(project) == 0: project = None
381	else:
382	raise service_error(service_error.req,
383	"More than one project credential")
384
385
386	user_fedids = [ u for u in user if isinstance(u, fedid)]
387
388	# Determine how the caller is representing itself. If its fedid shows
389	# up as a project or a singleton user, let that stand. If neither the
390	# usernames nor the project name is a fedid, the caller is a testbed.
391	if project and isinstance(project, fedid):
392	if project == fid:
393	# The caller is the project (which is already in the tuple
394	# passed in to the authorizer)
395	owners = user_fedids
396	owners.append(project)
397	else:
398	raise service_error(service_error.req,
399	"Project asserting different fedid")
400	else:
401	if fid not in user_fedids:
402	tb = fid
403	owners = user_fedids
404	owners.append(fid)
405	else:
406	if len(fedids) > 1:
407	raise service_error(service_error.req,
408	"User asserting different fedid")
409	else:
410	# Which is a singleton
411	owners = user_fedids
412	# Confirm authorization
413
414	for u in user:
415	self.log.debug("[lookup_access] Checking access for %s" % \
416	((tb, project, u),))
417	if self.auth.check_attribute((tb, project, u), 'access'):
418	self.log.debug("[lookup_access] Access granted")
419	break
420	else:
421	self.log.debug("[lookup_access] Access Denied")
422	else:
423	raise service_error(service_error.access, "Access denied")
424
425	# This maps a valid user to the ProtoGENI credentials to use
426	found = self.find_access((tb, project, user))
427
428	if found == None:
429	raise service_error(service_error.access,
430	"Access denied - cannot map access")
431	return found, owners
432
433	def get_handler(self, path, fid):
434	self.log.info("Get handler %s %s" % (path, fid))
435	if self.auth.check_attribute(fid, path) and self.userconfdir:
436	return ("%s/%s" % (self.userconfdir, path), "application/binary")
437	else:
438	return (None, None)
439
440	def export_userconf(self, project):
441	dev_null = None
442	confid, confcert = generate_fedid("test", dir=self.userconfdir,
443	log=self.log)
444	conffilename = "%s/%s" % (self.userconfdir, str(confid))
445	cf = None
446	try:
447	cf = open(conffilename, "w")
448	os.chmod(conffilename, stat.S_IRUSR \| stat.S_IWUSR)
449	except IOError, e:
450	raise service_error(service_error.internal,
451	"Cannot create user configuration data")
452
453	try:
454	dev_null = open("/dev/null", "a")
455	except IOError, e:
456	self.log.error("export_userconf: can't open /dev/null: %s" % e)
457
458	cmd = "%s %s" % (self.userconfcmd, project)
459	conf = subprocess.call(cmd.split(" "),
460	stdout=cf, stderr=dev_null, close_fds=True)
461
462	self.auth.set_attribute(confid, "/%s" % str(confid))
463
464	return confid, confcert
465
466
467	def export_services(self, sreq, project, user):
468	exp = [ ]
469	state = { }
470	# XXX: Filthy shortcut here using http: so urlparse will give the right
471	# answers.
472	for s in sreq:
473	sname = s.get('name', '')
474	svis = s.get('visibility', '')
475	if svis == 'export':
476	if sname in self.exports:
477	outs = s.copy()
478	if sname == 'SMB':
479	outs = s.copy()
480	outs['server'] = "http://fs:139"
481	outs['fedAttr'] = [
482	{ 'attribute': 'SMBSHARE', 'value': 'USERS' },
483	{ 'attribute': 'SMBUSER', 'value': user },
484	{ 'attribute': 'SMBPROJ', 'value': project },
485	]
486	elif sname == 'seer':
487	outs['server'] = "http://control:16606"
488	elif sname == 'tmcd':
489	outs['server'] = "http://boss:7777"
490	elif sname == 'userconfig':
491	if self.userconfdir and self.userconfcmd \
492	and self.userconfurl:
493	cid, cert = self.export_userconf(project)
494	outs['server'] = "%s/%s" % \
495	(self.userconfurl, str(cid))
496	outs['fedAttr'] = [
497	{ 'attribute': 'cert', 'value': cert },
498	]
499	state['userconfig'] = unicode(cid)
500	exp.append(outs)
501	return (exp, state)
502
503	def build_response(self, alloc_id, ap, services):
504	"""
505	Create the SOAP response.
506
507	Build the dictionary description of the response and use
508	fedd_utils.pack_soap to create the soap message. ap is the allocate
509	project message returned from a remote project allocation (even if that
510	allocation was done locally).
511	"""
512	# Because alloc_id is already a fedd_services_types.IDType_Holder,
513	# there's no need to repack it
514	msg = {
515	'allocID': alloc_id,
516	'fedAttr': [
517	{ 'attribute': 'domain', 'value': self.domain } ,
518	{ 'attribute': 'project', 'value':
519	ap['project'].get('name', {}).get('localname', "???") },
520	]
521	}
522	if len(self.attrs) > 0:
523	msg['fedAttr'].extend(
524	[ { 'attribute': x, 'value' : y } \
525	for x,y in self.attrs.iteritems()])
526
527	if services:
528	msg['service'] = services
529	return msg
530
531	def RequestAccess(self, req, fid):
532	"""
533	Handle the access request. Proxy if not for us.
534
535	Parse out the fields and make the allocations or rejections if for us,
536	otherwise, assuming we're willing to proxy, proxy the request out.
537	"""
538
539	# The dance to get into the request body
540	if req.has_key('RequestAccessRequestBody'):
541	req = req['RequestAccessRequestBody']
542	else:
543	raise service_error(service_error.req, "No request!?")
544
545	if req.has_key('destinationTestbed'):
546	dt = unpack_id(req['destinationTestbed'])
547
548	if dt == None or dt in self.testbed:
549	# Request for this fedd
550	found, owners = self.lookup_access(req, fid)
551	# keep track of what's been added
552	allocID, alloc_cert = generate_fedid(subj="alloc", log=self.log)
553	aid = unicode(allocID)
554
555	self.state_lock.acquire()
556	self.allocation[aid] = { }
557	# The protoGENI certificate
558	self.allocation[aid]['credentials'] = found
559	# The list of owner FIDs
560	self.allocation[aid]['owners'] = owners
561	self.write_state()
562	self.state_lock.release()
563	for o in owners:
564	self.auth.set_attribute(o, allocID)
565	self.auth.set_attribute(allocID, allocID)
566
567	try:
568	f = open("%s/%s.pem" % (self.certdir, aid), "w")
569	print >>f, alloc_cert
570	f.close()
571	except IOError, e:
572	raise service_error(service_error.internal,
573	"Can't open %s/%s : %s" % (self.certdir, aid, e))
574	return { 'allocID': { 'fedid': allocID } }
575	else:
576	if self.allow_proxy:
577	resp = self.proxy_RequestAccess.call_service(dt, req,
578	self.cert_file, self.cert_pwd,
579	self.trusted_certs)
580	if resp.has_key('RequestAccessResponseBody'):
581	return resp['RequestAccessResponseBody']
582	else:
583	return None
584	else:
585	raise service_error(service_error.access,
586	"Access proxying denied")
587
588
589	def ReleaseAccess(self, req, fid):
590	# The dance to get into the request body
591	if req.has_key('ReleaseAccessRequestBody'):
592	req = req['ReleaseAccessRequestBody']
593	else:
594	raise service_error(service_error.req, "No request!?")
595
596	if req.has_key('destinationTestbed'):
597	dt = unpack_id(req['destinationTestbed'])
598	else:
599	dt = None
600
601	if dt == None or dt in self.testbed:
602	# Local request
603	try:
604	if req['allocID'].has_key('localname'):
605	auth_attr = aid = req['allocID']['localname']
606	elif req['allocID'].has_key('fedid'):
607	aid = unicode(req['allocID']['fedid'])
608	auth_attr = req['allocID']['fedid']
609	else:
610	raise service_error(service_error.req,
611	"Only localnames and fedids are understood")
612	except KeyError:
613	raise service_error(service_error.req, "Badly formed request")
614
615	self.log.debug("[access] deallocation requested for %s", aid)
616	if not self.auth.check_attribute(fid, auth_attr):
617	self.log.debug("[access] deallocation denied for %s", aid)
618	raise service_error(service_error.access, "Access Denied")
619
620	self.state_lock.acquire()
621	if self.allocation.has_key(aid):
622	self.log.debug("Found allocation for %s" %aid)
623	del self.allocation[aid]
624	self.write_state()
625	self.state_lock.release()
626	# And remove the access cert
627	cf = "%s/%s.pem" % (self.certdir, aid)
628	self.log.debug("Removing %s" % cf)
629	os.remove(cf)
630	return { 'allocID': req['allocID'] }
631	else:
632	self.state_lock.release()
633	raise service_error(service_error.req, "No such allocation")
634
635	else:
636	if self.allow_proxy:
637	resp = self.proxy_ReleaseAccess.call_service(dt, req,
638	self.cert_file, self.cert_pwd,
639	self.trusted_certs)
640	if resp.has_key('ReleaseAccessResponseBody'):
641	return resp['ReleaseAccessResponseBody']
642	else:
643	return None
644	else:
645	raise service_error(service_error.access,
646	"Access proxying denied")
647
648	def import_store_info(self, cf, connInfo):
649	"""
650	Pull any import parameters in connInfo in. We translate them either
651	into known member names or fedAddrs.
652	"""
653
654	for c in connInfo:
655	for p in [ p for p in c.get('parameter', []) \
656	if p.get('type', '') == 'input']:
657	name = p.get('name', None)
658	key = p.get('key', None)
659	store = p.get('store', None)
660
661	if name and key and store :
662	req = { 'name': key, 'wait': True }
663	r = self.call_GetValue(store, req, cf)
664	r = r.get('GetValueResponseBody', None)
665	if r :
666	if r.get('name', '') == key:
667	v = r.get('value', None)
668	if v is not None:
669	if name == 'peer':
670	c['peer'] = v
671	else:
672	if c.has_key('fedAttr'):
673	c['fedAttr'].append({
674	'attribute': name, 'value': v})
675	else:
676	c['fedAttr']= [{
677	'attribute': name, 'value': v}]
678	else:
679	raise service_error(service_error.internal,
680	'None value exported for %s' % key)
681	else:
682	raise service_error(service_error.internal,
683	'Different name returned for %s: %s' \
684	% (key, r.get('name','')))
685	else:
686	raise service_error(service_error.internal,
687	'Badly formatted response: no GetValueResponseBody')
688	else:
689	raise service_error(service_error.internal,
690	'Bad Services missing info for import %s' % c)
691
692	def generate_portal_configs(self, topo, pubkey_base, secretkey_base,
693	tmpdir, master, leid, connInfo, services):
694
695	def conninfo_to_dict(key, info):
696	"""
697	Make a cpoy of the connection information about key, and flatten it
698	into a single dict by parsing out any feddAttrs.
699	"""
700
701	rv = None
702	for i in info:
703	if key == i.get('portal', "") or \
704	key in [e.get('element', "") \
705	for e in i.get('member', [])]:
706	rv = i.copy()
707	break
708
709	else:
710	return rv
711
712	if 'fedAttr' in rv:
713	for a in rv['fedAttr']:
714	attr = a.get('attribute', "")
715	val = a.get('value', "")
716	if attr and attr not in rv:
717	rv[attr] = val
718	del rv['fedAttr']
719	return rv
720
721	# XXX: un hardcode this
722	def client_null(f, s):
723	print >>f, "Service: %s" % s['name']
724
725	def client_smb(f, s):
726	print >>f, "Service: %s" % s['name']
727	smbshare = None
728	smbuser = None
729	smbproj = None
730	for a in s.get('fedAttr', []):
731	if a.get('attribute', '') == 'SMBSHARE':
732	smbshare = a.get('value', None)
733	elif a.get('attribute', '') == 'SMBUSER':
734	smbuser = a.get('value', None)
735	elif a.get('attribute', '') == 'SMBPROJ':
736	smbproj = a.get('value', None)
737
738	if all((smbshare, smbuser, smbproj)):
739	print >>f, "SMBshare: %s" % smbshare
740	print >>f, "ProjectUser: %s" % smbuser
741	print >>f, "ProjectName: %s" % smbproj
742
743	client_service_out = {
744	'SMB': client_smb,
745	'tmcd': client_null,
746	'seer': client_null,
747	'userconfig': client_null,
748	}
749	# XXX: end un hardcode this
750
751
752	seer_out = False
753	client_out = False
754	for e in [ e for e in topo.elements \
755	if isinstance(e, topdl.Computer) and e.get_attribute('portal')]:
756	myname = e.name[0]
757	type = e.get_attribute('portal_type')
758
759	info = conninfo_to_dict(myname, connInfo)
760
761	if not info:
762	raise service_error(service_error.req,
763	"No connectivity info for %s" % myname)
764
765	peer = info.get('peer', "")
766	ldomain = self.domain;
767
768	mexp = info.get('masterexperiment',"")
769	mproj, meid = mexp.split("/", 1)
770	mdomain = info.get('masterdomain',"")
771	muser = info.get('masteruser','root')
772	smbshare = info.get('smbshare', 'USERS')
773	ssh_port = info.get('ssh_port', '22')
774
775	active = info.get('active', 'False')
776
777	cfn = "%s/%s.gw.conf" % (tmpdir, myname.lower())
778	tunnelconfig = self.attrs.has_key('TunnelCfg')
779	try:
780	f = open(cfn, "w")
781	if active == 'True':
782	print >>f, "active: True"
783	print >>f, "ssh_port: %s" % ssh_port
784	if type in ('control', 'both'):
785	for s in [s for s in services \
786	if s.get('name', "") in self.imports]:
787	p = urlparse(s.get('server', 'http://localhost'))
788	print >>f, 'port: remote:%s:%s:%s' % \
789	(p.port, p.hostname, p.port)
790
791	if tunnelconfig:
792	print >>f, "tunnelip: %s" % tunnelconfig
793	# XXX: send this an fedattr
794	#print >>f, "seercontrol: control.%s.%s%s" % \
795	#(meid.lower(), mproj.lower(), mdomain)
796	print >>f, "peer: %s" % peer.lower()
797	print >>f, "ssh_pubkey: /usr/local/federation/etc/%s" % \
798	pubkey_base
799	print >>f, "ssh_privkey: /usr/local/federation/etc/%s" % \
800	secretkey_base
801	f.close()
802	except IOError, e:
803	raise service_error(service_error.internal,
804	"Can't write protal config %s: %s" % (cfn, e))
805
806	# XXX: This little seer config file needs to go away.
807	if not seer_out:
808	try:
809	seerfn = "%s/seer.conf" % tmpdir
810	f = open(seerfn, "w")
811	if not master:
812	print >>f, "ControlNode: control.%s.%s%s" % \
813	(meid.lower(), mproj.lower(), mdomain)
814	print >>f, "ExperimentID: %s" % mexp
815	f.close()
816	except IOError, e:
817	raise service_error(service_error.internal,
818	"Can't write seer.conf: %s" %e)
819	seer_out = True
820
821	if not client_out and type in ('control', 'both'):
822	try:
823	f = open("%s/client.conf" % tmpdir, "w")
824	print >>f, "ControlGateway: %s%s" % \
825	(myname.lower(), ldomain.lower())
826	for s in services:
827	if s.get('name',"") in self.imports and \
828	s.get('visibility','') == 'import':
829	client_service_out[s['name']](f, s)
830	# Does seer need this?
831	# print >>f, "ExperimentID: %s/%s" % (mproj, meid)
832	f.close()
833	except IOError, e:
834	raise service_error(service_error.internal,
835	"Cannot write client.conf: %s" %s)
836	client_out = True
837
838
839	def generate_rspec(self, topo, softdir, master, connInfo):
840	t = topo.clone()
841
842	starts = { }
843
844	# The startcmds for master and slave testbeds
845	if master:
846	gate_cmd = self.attrs.get('MasterConnectorStartCmd', '/bin/true')
847	node_cmd = self.attrs.get('MasterNodeStartCmd', 'bin/true')
848	else:
849	gate_cmd = self.attrs.get('SlaveConnectorStartCmd', '/bin/true')
850	node_cmd = self.attrs.get('SlaveNodeStartCmd', 'bin/true')
851
852	# Weed out the things we aren't going to instantiate: Segments, portal
853	# substrates, and portal interfaces. (The copy in the for loop allows
854	# us to delete from e.elements in side the for loop). While we're
855	# touching all the elements, we also adjust paths from the original
856	# testbed to local testbed paths and put the federation commands and
857	# startcommands into a dict so we can start them manually later.
858	# ProtoGENI requires setup before the federation commands run, so we
859	# run them by hand after we've seeded configurations.
860	for e in [e for e in t.elements]:
861	if isinstance(e, topdl.Segment):
862	t.elements.remove(e)
863	# Fix software paths
864	for s in getattr(e, 'software', []):
865	s.location = re.sub("^.*/", softdir, s.location)
866	if isinstance(e, topdl.Computer):
867	if e.get_attribute('portal') and gate_cmd:
868	# Portals never have a user-specified start command
869	starts[e.name[0]] = gate_cmd
870	elif node_cmd:
871	if e.get_attribute('startup'):
872	starts[e.name[0]] = "%s \\$USER '%s'" % \
873	(node_cmd, e.get_attribute('startup'))
874	e.remove_attribute('startup')
875	else:
876	starts[e.name[0]] = node_cmd
877
878	# Remove portal interfaces
879	e.interface = [i for i in e.interface \
880	if not i.get_attribute('portal')]
881
882	t.substrates = [ s.clone() for s in t.substrates ]
883	t.incorporate_elements()
884
885	# Customize the ns2 output for local portal commands and images
886	filters = []
887
888	# NB: these are extra commands issued for the node, not the startcmds
889	if master: cmd = self.attrs.get('MasterConnectorCmd', '')
890	else: cmd = self.attrs.get('SlaveConnectorCmd', '')
891
892	if cmd:
893	filters.append(topdl.generate_portal_command_filter(cmd))
894
895	# Convert to rspec and return it
896	exp_rspec = topdl.topology_to_rspec(t, filters)
897
898	return exp_rspec
899
900	def StartSegment(self, req, fid):
901	def get_url(url, cf, destdir, fn=None):
902	po = urlparse(url)
903	if not fn:
904	fn = po.path.rpartition('/')[2]
905	retries = 0
906	ok = False
907	while not ok and retries < 5:
908	try:
909	conn = httplib.HTTPSConnection(po.hostname, port=po.port,
910	cert_file=cf, key_file=cf)
911	conn.putrequest('GET', po.path)
912	conn.endheaders()
913	response = conn.getresponse()
914
915	lf = open("%s/%s" % (destdir, fn), "w")
916	buf = response.read(4096)
917	while buf:
918	lf.write(buf)
919	buf = response.read(4096)
920	lf.close()
921	ok = True
922	except IOError, e:
923	print e
924	raise service_error(service_error.internal,
925	"Error writing tempfile: %s" %e)
926	except httplib.HTTPException, e:
927	print e
928	raise service_error(service_error.internal,
929	"Error retrieving data: %s" % e)
930	except SSLError, e:
931	print "SSL error %s" %e
932	retries += 1
933
934	if retries > 5:
935	raise service_error(service_error.internal,
936	"Repeated SSL failures")
937
938
939	configs = set(('hosts', 'ssh_pubkey', 'ssh_secretkey'))
940
941	err = None # Any service_error generated after tmpdir is created
942	rv = None # Return value from segment creation
943
944	try:
945	req = req['StartSegmentRequestBody']
946	except KeyError:
947	raise service_error(service_error.req, "Badly formed request")
948
949	connInfo = req.get('connection', [])
950	services = req.get('service', [])
951	auth_attr = req['allocID']['fedid']
952	aid = "%s" % auth_attr
953	attrs = req.get('fedAttr', [])
954	if not self.auth.check_attribute(fid, auth_attr):
955	raise service_error(service_error.access, "Access denied")
956
957
958	if req.has_key('segmentdescription') and \
959	req['segmentdescription'].has_key('topdldescription'):
960	topo = \
961	topdl.Topology(**req['segmentdescription']['topdldescription'])
962	else:
963	raise service_error(service_error.req,
964	"Request missing segmentdescription'")
965
966	master = req.get('master', False)
967
968	certfile = "%s/%s.pem" % (self.certdir, auth_attr)
969	try:
970	tmpdir = tempfile.mkdtemp(prefix="access-")
971	softdir = "%s/software" % tmpdir
972	except IOError:
973	raise service_error(service_error.internal, "Cannot create tmp dir")
974
975	# Try block alllows us to clean up temporary files.
976	try:
977	sw = set()
978	for e in topo.elements:
979	for s in getattr(e, 'software', []):
980	sw.add(s.location)
981	if len(sw) > 0:
982	os.mkdir(softdir)
983	for s in sw:
984	self.log.debug("Retrieving %s" % s)
985	get_url(s, certfile, softdir)
986
987	# Copy local portal node software to the tempdir
988	for s in (self.portal_software, self.federation_software):
989	for l, f in s:
990	base = os.path.basename(f)
991	copy_file(f, "%s/%s" % (softdir, base))
992
993	# Ick. Put this python rpm in a place that it will get moved into
994	# the staging area. It's a hack to install a modern (in a Roman
995	# sense of modern) python on ProtoGENI
996	python_rpm ="python2.4-2.4-1pydotorg.i586.rpm"
997	if os.access("./%s" % python_rpm, os.R_OK):
998	copy_file("./%s" % python_rpm, "%s/%s" % (softdir, python_rpm))
999
1000	for a in attrs:
1001	if a['attribute'] in configs:
1002	get_url(a['value'], certfile, tmpdir)
1003	if a['attribute'] == 'ssh_pubkey':
1004	pubkey_base = a['value'].rpartition('/')[2]
1005	if a['attribute'] == 'ssh_secretkey':
1006	secretkey_base = a['value'].rpartition('/')[2]
1007	if a['attribute'] == 'experiment_name':
1008	ename = a['value']
1009
1010	# If the userconf service was imported, collect the configuration
1011	# data.
1012	for s in services:
1013	if s.get("name", "") == 'userconfig' \
1014	and s.get('visibility',"") == 'import':
1015
1016	# Collect ther server and certificate info.
1017	u = s.get('server', None)
1018	for a in s.get('fedAttr', []):
1019	if a.get('attribute',"") == 'cert':
1020	cert = a.get('value', None)
1021	break
1022	else:
1023	cert = None
1024
1025	if cert:
1026	# Make a temporary certificate file for get_url. The
1027	# finally clause removes it whether something goes
1028	# wrong (including an exception from get_url) or not.
1029	try:
1030	tfos, tn = tempfile.mkstemp(suffix=".pem")
1031	tf = os.fdopen(tfos, 'w')
1032	print >>tf, cert
1033	tf.close()
1034	get_url(u, tn, tmpdir, "userconf")
1035	except IOError, e:
1036	raise service_error(service.error.internal,
1037	"Cannot create temp file for " +
1038	"userconfig certificates: %s e")
1039	finally:
1040	if tn: os.remove(tn)
1041	else:
1042	raise service_error(service_error.req,
1043	"No certificate for retreiving userconfig")
1044	break
1045
1046
1047
1048	self.state_lock.acquire()
1049	if self.allocation.has_key(aid):
1050	cf, user, ssh_key, cpw = self.allocation[aid]['credentials']
1051	self.allocation[aid]['experiment'] = ename
1052	self.allocation[aid]['log'] = [ ]
1053	# Create a logger that logs to the experiment's state object as
1054	# well as to the main log file.
1055	alloc_log = logging.getLogger('fedd.access.%s' % ename)
1056	h = logging.StreamHandler(
1057	list_log.list_log(self.allocation[aid]['log']))
1058	# XXX: there should be a global one of these rather than
1059	# repeating the code.
1060	h.setFormatter(logging.Formatter(
1061	"%(asctime)s %(name)s %(message)s",
1062	'%d %b %y %H:%M:%S'))
1063	alloc_log.addHandler(h)
1064	self.write_state()
1065	else:
1066	self.log.error("No allocation for %s!?" % aid)
1067	self.state_lock.release()
1068
1069	# XXX: we really need to put the import and connection info
1070	# generation off longer.
1071	self.import_store_info(certfile, connInfo)
1072	#self.generate_portal_configs(topo, pubkey_base,
1073	#secretkey_base, tmpdir, master, ename, connInfo,
1074	#services)
1075	rspec = self.generate_rspec(topo, "%s/%s/" \
1076	% (self.staging_dir, ename), master, connInfo)
1077
1078	starter = self.start_segment(keyfile=ssh_key,
1079	debug=self.create_debug, log=alloc_log,
1080	ch_url = self.ch_url, sa_url=self.sa_url,
1081	cm_url=self.cm_url)
1082	rv = starter(self, aid, user, rspec, pubkey_base, secretkey_base,
1083	master, ename,
1084	"%s/%s" % (self.staging_dir, ename), tmpdir, cf, cpw,
1085	certfile, topo, connInfo, services)
1086	except service_error, e:
1087	err = e
1088	except e:
1089	err = service_error(service_error.internal, str(e))
1090
1091	# Walk up tmpdir, deleting as we go
1092	if self.cleanup:
1093	self.log.debug("[StartSegment]: removing %s" % tmpdir)
1094	for path, dirs, files in os.walk(tmpdir, topdown=False):
1095	for f in files:
1096	os.remove(os.path.join(path, f))
1097	for d in dirs:
1098	os.rmdir(os.path.join(path, d))
1099	os.rmdir(tmpdir)
1100	else:
1101	self.log.debug("[StartSegment]: not removing %s" % tmpdir)
1102
1103	if rv:
1104	# Grab the log (this is some anal locking, but better safe than
1105	# sorry)
1106	self.state_lock.acquire()
1107	logv = "".join(self.allocation[aid]['log'])
1108	self.state_lock.release()
1109
1110	return { 'allocID': req['allocID'], 'allocationLog': logv }
1111	elif err:
1112	raise service_error(service_error.federant,
1113	"Swapin failed: %s" % err)
1114	else:
1115	raise service_error(service_error.federant, "Swapin failed")
1116
1117	def TerminateSegment(self, req, fid):
1118	try:
1119	req = req['TerminateSegmentRequestBody']
1120	except KeyError:
1121	raise service_error(service_error.req, "Badly formed request")
1122
1123	auth_attr = req['allocID']['fedid']
1124	aid = "%s" % auth_attr
1125	attrs = req.get('fedAttr', [])
1126	if not self.auth.check_attribute(fid, auth_attr):
1127	raise service_error(service_error.access, "Access denied")
1128
1129	self.state_lock.acquire()
1130	if self.allocation.has_key(aid):
1131	cf, user, ssh_key, cpw = self.allocation[aid]['credentials']
1132	slice_cred = self.allocation[aid].get('slice_credential', None)
1133	ename = self.allocation[aid].get('experiment', None)
1134	else:
1135	cf, user, ssh_key, cpw = (None, None, None, None)
1136	slice_cred = None
1137	ename = None
1138	self.state_lock.release()
1139
1140	if ename:
1141	staging = "%s/%s" % ( self.staging_dir, ename)
1142	else:
1143	self.log.warn("Can't find experiment name for %s" % aid)
1144	staging = None
1145
1146	stopper = self.stop_segment(keyfile=ssh_key, debug=self.create_debug,
1147	ch_url = self.ch_url, sa_url=self.sa_url, cm_url=self.cm_url)
1148	stopper(self, user, staging, slice_cred, cf, cpw)
1149	return { 'allocID': req['allocID'] }
1150
1151	def RenewSlices(self):
1152	self.log.info("Scanning for slices to renew")
1153	self.state_lock.acquire()
1154	aids = self.allocation.keys()
1155	self.state_lock.release()
1156
1157	for aid in aids:
1158	self.state_lock.acquire()
1159	if self.allocation.has_key(aid):
1160	name = self.allocation[aid].get('slice_name', None)
1161	scred = self.allocation[aid].get('slice_credential', None)
1162	cf, user, ssh_key, cpw = self.allocation[aid]['credentials']
1163	else:
1164	name = None
1165	scred = None
1166	self.state_lock.release()
1167
1168	# There's a ProtoGENI slice associated with the segment; renew it.
1169	if name and scred:
1170	renewer = self.renew_segment(log=self.log,
1171	debug=self.create_debug, keyfile=ssh_key,
1172	cm_url = self.cm_url, sa_url = self.sa_url,
1173	ch_url = self.ch_url)
1174	new_scred = renewer(name, scred, self.renewal_interval, cf, cpw)
1175	if new_scred:
1176	self.log.info("Slice %s renewed until %s GMT" % \
1177	(name, time.asctime(time.gmtime(\
1178	time.time()+self.renewal_interval))))
1179	self.state_lock.acquire()
1180	if self.allocation.has_key(aid):
1181	self.allocation[aid]['slice_credential'] = new_scred
1182	self.state_lock.release()
1183	else:
1184	self.log.info("Failed to renew slice %s " % name)
1185
1186	# Let's do this all again soon. (4 tries before the slices time out)
1187	t = Timer(self.renewal_interval/4, self.RenewSlices)
1188	t.start()

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

Download in other formats: