diff options
| -rw-r--r-- | nova/api/openstack/servers.py | 9 | ||||
| -rw-r--r-- | nova/compute/api.py | 115 | ||||
| -rw-r--r-- | nova/compute/manager.py | 33 | ||||
| -rw-r--r-- | nova/exception.py | 4 | ||||
| -rw-r--r-- | nova/tests/test_compute.py | 7 | ||||
| -rw-r--r-- | nova/tests/test_xenapi.py | 27 | ||||
| -rw-r--r-- | nova/virt/fake.py | 21 | ||||
| -rw-r--r-- | nova/virt/xenapi/vmops.py | 199 | ||||
| -rw-r--r-- | nova/virt/xenapi_conn.py | 7 | ||||
| -rwxr-xr-x | plugins/xenserver/xenapi/etc/xapi.d/plugins/agent | 126 |
10 files changed, 431 insertions, 117 deletions
diff --git a/nova/api/openstack/servers.py b/nova/api/openstack/servers.py index 29af82533..8cbcebed2 100644 --- a/nova/api/openstack/servers.py +++ b/nova/api/openstack/servers.py @@ -165,15 +165,18 @@ class Controller(wsgi.Controller): if not inst_dict: return faults.Fault(exc.HTTPUnprocessableEntity()) + ctxt = req.environ['nova.context'] update_dict = {} if 'adminPass' in inst_dict['server']: update_dict['admin_pass'] = inst_dict['server']['adminPass'] + try: + self.compute_api.set_admin_password(ctxt, id) + except exception.TimeoutException, e: + return exc.HTTPRequestTimeout() if 'name' in inst_dict['server']: update_dict['display_name'] = inst_dict['server']['name'] - try: - self.compute_api.update(req.environ['nova.context'], id, - **update_dict) + self.compute_api.update(ctxt, id, **update_dict) except exception.NotFound: return faults.Fault(exc.HTTPNotFound()) return exc.HTTPNoContent() diff --git a/nova/compute/api.py b/nova/compute/api.py index bf921aa00..90273da36 100644 --- a/nova/compute/api.py +++ b/nova/compute/api.py @@ -280,7 +280,7 @@ class API(base.Base): return self.db.instance_update(context, instance_id, kwargs) def delete(self, context, instance_id): - LOG.debug(_("Going to try and terminate %s"), instance_id) + LOG.debug(_("Going to try to terminate %s"), instance_id) try: instance = self.get(context, instance_id) except exception.NotFound as e: @@ -301,10 +301,8 @@ class API(base.Base): host = instance['host'] if host: - rpc.cast(context, - self.db.queue_get_for(context, FLAGS.compute_topic, host), - {"method": "terminate_instance", - "args": {"instance_id": instance_id}}) + self._cast_compute_message('terminate_instance', context, + instance_id, host) else: self.db.instance_destroy(context, instance_id) @@ -332,50 +330,34 @@ class API(base.Base): project_id) return self.db.instance_get_all(context) + def _cast_compute_message(self, method, context, instance_id, host=None): + """Generic handler for RPC calls to compute.""" + if not host: + instance = self.get(context, instance_id) + host = instance['host'] + queue = self.db.queue_get_for(context, FLAGS.compute_topic, host) + kwargs = {'method': method, 'args': {'instance_id': instance_id}} + rpc.cast(context, queue, kwargs) + def snapshot(self, context, instance_id, name): """Snapshot the given instance.""" - instance = self.get(context, instance_id) - host = instance['host'] - rpc.cast(context, - self.db.queue_get_for(context, FLAGS.compute_topic, host), - {"method": "snapshot_instance", - "args": {"instance_id": instance_id, "name": name}}) + self._cast_compute_message('snapshot_instance', context, instance_id) def reboot(self, context, instance_id): """Reboot the given instance.""" - instance = self.get(context, instance_id) - host = instance['host'] - rpc.cast(context, - self.db.queue_get_for(context, FLAGS.compute_topic, host), - {"method": "reboot_instance", - "args": {"instance_id": instance_id}}) + self._cast_compute_message('reboot_instance', context, instance_id) def pause(self, context, instance_id): """Pause the given instance.""" - instance = self.get(context, instance_id) - host = instance['host'] - rpc.cast(context, - self.db.queue_get_for(context, FLAGS.compute_topic, host), - {"method": "pause_instance", - "args": {"instance_id": instance_id}}) + self._cast_compute_message('pause_instance', context, instance_id) def unpause(self, context, instance_id): """Unpause the given instance.""" - instance = self.get(context, instance_id) - host = instance['host'] - rpc.cast(context, - self.db.queue_get_for(context, FLAGS.compute_topic, host), - {"method": "unpause_instance", - "args": {"instance_id": instance_id}}) + self._cast_compute_message('unpause_instance', context, instance_id) def get_diagnostics(self, context, instance_id): """Retrieve diagnostics for the given instance.""" - instance = self.get(context, instance_id) - host = instance["host"] - return rpc.call(context, - self.db.queue_get_for(context, FLAGS.compute_topic, host), - {"method": "get_diagnostics", - "args": {"instance_id": instance_id}}) + self._cast_compute_message('get_diagnostics', context, instance_id) def get_actions(self, context, instance_id): """Retrieve actions for the given instance.""" @@ -383,39 +365,23 @@ class API(base.Base): def suspend(self, context, instance_id): """suspend the instance with instance_id""" - instance = self.get(context, instance_id) - host = instance['host'] - rpc.cast(context, - self.db.queue_get_for(context, FLAGS.compute_topic, host), - {"method": "suspend_instance", - "args": {"instance_id": instance_id}}) + self._cast_compute_message('suspend_instance', context, instance_id) def resume(self, context, instance_id): """resume the instance with instance_id""" - instance = self.get(context, instance_id) - host = instance['host'] - rpc.cast(context, - self.db.queue_get_for(context, FLAGS.compute_topic, host), - {"method": "resume_instance", - "args": {"instance_id": instance_id}}) + self._cast_compute_message('resume_instance', context, instance_id) def rescue(self, context, instance_id): """Rescue the given instance.""" - instance = self.get(context, instance_id) - host = instance['host'] - rpc.cast(context, - self.db.queue_get_for(context, FLAGS.compute_topic, host), - {"method": "rescue_instance", - "args": {"instance_id": instance_id}}) + self._cast_compute_message('rescue_instance', context, instance_id) def unrescue(self, context, instance_id): """Unrescue the given instance.""" - instance = self.get(context, instance_id) - host = instance['host'] - rpc.cast(context, - self.db.queue_get_for(context, FLAGS.compute_topic, host), - {"method": "unrescue_instance", - "args": {"instance_id": instance['id']}}) + self._cast_compute_message('unrescue_instance', context, instance_id) + + def set_admin_password(self, context, instance_id): + """Set the root/admin password for the given instance.""" + self._cast_compute_message('set_admin_password', context, instance_id) def get_ajax_console(self, context, instance_id): """Get a url to an AJAX Console""" @@ -437,35 +403,16 @@ class API(base.Base): output['token'])} def lock(self, context, instance_id): - """ - lock the instance with instance_id - - """ - instance = self.get_instance(context, instance_id) - host = instance['host'] - rpc.cast(context, - self.db.queue_get_for(context, FLAGS.compute_topic, host), - {"method": "lock_instance", - "args": {"instance_id": instance['id']}}) + """lock the instance with instance_id""" + self._cast_compute_message('lock_instance', context, instance_id) def unlock(self, context, instance_id): - """ - unlock the instance with instance_id - - """ - instance = self.get_instance(context, instance_id) - host = instance['host'] - rpc.cast(context, - self.db.queue_get_for(context, FLAGS.compute_topic, host), - {"method": "unlock_instance", - "args": {"instance_id": instance['id']}}) + """unlock the instance with instance_id""" + self._cast_compute_message('unlock_instance', context, instance_id) def get_lock(self, context, instance_id): - """ - return the boolean state of (instance with instance_id)'s lock - - """ - instance = self.get_instance(context, instance_id) + """return the boolean state of (instance with instance_id)'s lock""" + instance = self.get(context, instance_id) return instance['locked'] def attach_volume(self, context, instance_id, volume_id, device): diff --git a/nova/compute/manager.py b/nova/compute/manager.py index 6b2fc4adb..613ee45f6 100644 --- a/nova/compute/manager.py +++ b/nova/compute/manager.py @@ -35,6 +35,8 @@ terminating it. """ import datetime +import random +import string import logging import socket import functools @@ -54,6 +56,8 @@ flags.DEFINE_string('compute_driver', 'nova.virt.connection.get_connection', 'Driver to use for controlling virtualization') flags.DEFINE_string('stub_network', False, 'Stub network related code') +flags.DEFINE_integer('password_length', 12, + 'Length of generated admin passwords') flags.DEFINE_string('console_host', socket.gethostname(), 'Console proxy host to use to connect to instances on' 'this host.') @@ -311,6 +315,35 @@ class ComputeManager(manager.Manager): @exception.wrap_exception @checks_instance_lock + def set_admin_password(self, context, instance_id, new_pass=None): + """Set the root/admin password for an instance on this server.""" + context = context.elevated() + instance_ref = self.db.instance_get(context, instance_id) + if instance_ref['state'] != power_state.RUNNING: + logging.warn('trying to reset the password on a non-running ' + 'instance: %s (state: %s expected: %s)', + instance_ref['id'], + instance_ref['state'], + power_state.RUNNING) + + logging.debug('instance %s: setting admin password', + instance_ref['name']) + if new_pass is None: + # Generate a random password + new_pass = self._generate_password(FLAGS.password_length) + + self.driver.set_admin_password(instance_ref, new_pass) + self._update_state(context, instance_id) + + def _generate_password(self, length=20): + """Generate a random sequence of letters and digits + to be used as a password. + """ + chrs = string.letters + string.digits + return "".join([random.choice(chrs) for i in xrange(length)]) + + @exception.wrap_exception + @checks_instance_lock def rescue_instance(self, context, instance_id): """Rescue an instance on this server.""" context = context.elevated() diff --git a/nova/exception.py b/nova/exception.py index 7680e534a..ecd814e5d 100644 --- a/nova/exception.py +++ b/nova/exception.py @@ -76,6 +76,10 @@ class InvalidInputException(Error): pass +class TimeoutException(Error): + pass + + def wrap_exception(f): def _wrap(*args, **kw): try: diff --git a/nova/tests/test_compute.py b/nova/tests/test_compute.py index 52660ee74..a7d47961c 100644 --- a/nova/tests/test_compute.py +++ b/nova/tests/test_compute.py @@ -151,6 +151,13 @@ class ComputeTestCase(test.TestCase): self.compute.reboot_instance(self.context, instance_id) self.compute.terminate_instance(self.context, instance_id) + def test_set_admin_password(self): + """Ensure instance can have its admin password set""" + instance_id = self._create_instance() + self.compute.run_instance(self.context, instance_id) + self.compute.set_admin_password(self.context, instance_id) + self.compute.terminate_instance(self.context, instance_id) + def test_snapshot(self): """Ensure instance can be snapshotted""" instance_id = self._create_instance() diff --git a/nova/tests/test_xenapi.py b/nova/tests/test_xenapi.py index ec9462ada..261ee0fde 100644 --- a/nova/tests/test_xenapi.py +++ b/nova/tests/test_xenapi.py @@ -31,6 +31,7 @@ from nova.compute import power_state from nova.virt import xenapi_conn from nova.virt.xenapi import fake as xenapi_fake from nova.virt.xenapi import volume_utils +from nova.virt.xenapi.vmops import SimpleDH from nova.tests.db import fakes as db_fakes from nova.tests.xenapi import stubs @@ -262,3 +263,29 @@ class XenAPIVMTestCase(test.TestCase): instance = db.instance_create(values) self.conn.spawn(instance) return instance + + +class XenAPIDiffieHellmanTestCase(test.TestCase): + """ + Unit tests for Diffie-Hellman code + """ + def setUp(self): + super(XenAPIDiffieHellmanTestCase, self).setUp() + self.alice = SimpleDH() + self.bob = SimpleDH() + + def test_shared(self): + alice_pub = self.alice.get_public() + bob_pub = self.bob.get_public() + alice_shared = self.alice.compute_shared(bob_pub) + bob_shared = self.bob.compute_shared(alice_pub) + self.assertEquals(alice_shared, bob_shared) + + def test_encryption(self): + msg = "This is a top-secret message" + enc = self.alice.encrypt(msg) + dec = self.bob.decrypt(enc) + self.assertEquals(dec, msg) + + def tearDown(self): + super(XenAPIDiffieHellmanTestCase, self).tearDown() diff --git a/nova/virt/fake.py b/nova/virt/fake.py index 9186d885e..a57a8f43b 100644 --- a/nova/virt/fake.py +++ b/nova/virt/fake.py @@ -98,7 +98,7 @@ class FakeConnection(object): the new instance. The work will be done asynchronously. This function returns a - Deferred that allows the caller to detect when it is complete. + task that allows the caller to detect when it is complete. Once this successfully completes, the instance should be running (power_state.RUNNING). @@ -122,7 +122,7 @@ class FakeConnection(object): The second parameter is the name of the snapshot. The work will be done asynchronously. This function returns a - Deferred that allows the caller to detect when it is complete. + task that allows the caller to detect when it is complete. """ pass @@ -134,7 +134,20 @@ class FakeConnection(object): and so the instance is being specified as instance.name. The work will be done asynchronously. This function returns a - Deferred that allows the caller to detect when it is complete. + task that allows the caller to detect when it is complete. + """ + pass + + def set_admin_password(self, instance, new_pass): + """ + Set the root password on the specified instance. + + The first parameter is an instance of nova.compute.service.Instance, + and so the instance is being specified as instance.name. The second + parameter is the value of the new password. + + The work will be done asynchronously. This function returns a + task that allows the caller to detect when it is complete. """ pass @@ -182,7 +195,7 @@ class FakeConnection(object): and so the instance is being specified as instance.name. The work will be done asynchronously. This function returns a - Deferred that allows the caller to detect when it is complete. + task that allows the caller to detect when it is complete. """ del self.instances[instance.name] diff --git a/nova/virt/xenapi/vmops.py b/nova/virt/xenapi/vmops.py index 7aebb502f..6e359ef82 100644 --- a/nova/virt/xenapi/vmops.py +++ b/nova/virt/xenapi/vmops.py @@ -20,6 +20,11 @@ Management class for VM-related functions (spawn, reboot, etc). """ import json +import M2Crypto +import os +import subprocess +import tempfile +import uuid from nova import db from nova import context @@ -127,12 +132,31 @@ class VMOps(object): """Refactored out the common code of many methods that receive either a vm name or a vm instance, and want a vm instance in return. """ + vm = None try: - instance_name = instance_or_vm.name - vm = VMHelper.lookup(self._session, instance_name) - except AttributeError: - # A vm opaque ref was passed - vm = instance_or_vm + if instance_or_vm.startswith("OpaqueRef:"): + # Got passed an opaque ref; return it + return instance_or_vm + else: + # Must be the instance name + instance_name = instance_or_vm + except (AttributeError, KeyError): + # Note the the KeyError will only happen with fakes.py + # Not a string; must be an ID or a vm instance + if isinstance(instance_or_vm, (int, long)): + ctx = context.get_admin_context() + try: + instance_obj = db.instance_get_by_id(ctx, instance_or_vm) + instance_name = instance_obj.name + except exception.NotFound: + # The unit tests screw this up, as they use an integer for + # the vm name. I'd fix that up, but that's a matter for + # another bug report. So for now, just try with the passed + # value + instance_name = instance_or_vm + else: + instance_name = instance_or_vm.name + vm = VMHelper.lookup(self._session, instance_name) if vm is None: raise Exception(_('Instance not present %s') % instance_name) return vm @@ -189,6 +213,44 @@ class VMOps(object): task = self._session.call_xenapi('Async.VM.clean_reboot', vm) self._session.wait_for_task(instance.id, task) + def set_admin_password(self, instance, new_pass): + """Set the root/admin password on the VM instance. This is done via + an agent running on the VM. Communication between nova and the agent + is done via writing xenstore records. Since communication is done over + the XenAPI RPC calls, we need to encrypt the password. We're using a + simple Diffie-Hellman class instead of the more advanced one in + M2Crypto for compatibility with the agent code. + """ + # Need to uniquely identify this request. + transaction_id = str(uuid.uuid4()) + # The simple Diffie-Hellman class is used to manage key exchange. + dh = SimpleDH() + args = {'id': transaction_id, 'pub': str(dh.get_public())} + resp = self._make_agent_call('key_init', instance, '', args) + if resp is None: + # No response from the agent + return + resp_dict = json.loads(resp) + # Successful return code from key_init is 'D0' + if resp_dict['returncode'] != 'D0': + # There was some sort of error; the message will contain + # a description of the error. + raise RuntimeError(resp_dict['message']) + agent_pub = int(resp_dict['message']) + dh.compute_shared(agent_pub) + enc_pass = dh.encrypt(new_pass) + # Send the encrypted password + args['enc_pass'] = enc_pass + resp = self._make_agent_call('password', instance, '', args) + if resp is None: + # No response from the agent + return + resp_dict = json.loads(resp) + # Successful return code from password is '0' + if resp_dict['returncode'] != '0': + raise RuntimeError(resp_dict['message']) + return resp_dict['message'] + def destroy(self, instance): """Destroy VM instance""" vm = VMHelper.lookup(self._session, instance.name) @@ -246,30 +308,19 @@ class VMOps(object): def suspend(self, instance, callback): """suspend the specified instance""" - instance_name = instance.name - vm = VMHelper.lookup(self._session, instance_name) - if vm is None: - raise Exception(_("suspend: instance not present %s") % - instance_name) + vm = self._get_vm_opaque_ref(instance) task = self._session.call_xenapi('Async.VM.suspend', vm) self._wait_with_callback(instance.id, task, callback) def resume(self, instance, callback): """resume the specified instance""" - instance_name = instance.name - vm = VMHelper.lookup(self._session, instance_name) - if vm is None: - raise Exception(_("resume: instance not present %s") % - instance_name) + vm = self._get_vm_opaque_ref(instance) task = self._session.call_xenapi('Async.VM.resume', vm, False, True) self._wait_with_callback(instance.id, task, callback) - def get_info(self, instance_id): + def get_info(self, instance): """Return data about VM instance""" - vm = VMHelper.lookup(self._session, instance_id) - if vm is None: - raise exception.NotFound(_('Instance not' - ' found %s') % instance_id) + vm = self._get_vm_opaque_ref(instance) rec = self._session.get_xenapi().VM.get_record(vm) return VMHelper.compile_info(rec) @@ -333,22 +384,34 @@ class VMOps(object): return self._make_plugin_call('xenstore.py', method=method, vm=vm, path=path, addl_args=addl_args) + def _make_agent_call(self, method, vm, path, addl_args={}): + """Abstracts out the interaction with the agent xenapi plugin.""" + return self._make_plugin_call('agent', method=method, vm=vm, + path=path, addl_args=addl_args) + def _make_plugin_call(self, plugin, method, vm, path, addl_args={}): """Abstracts out the process of calling a method of a xenapi plugin. Any errors raised by the plugin will in turn raise a RuntimeError here. """ + instance_id = vm.id vm = self._get_vm_opaque_ref(vm) rec = self._session.get_xenapi().VM.get_record(vm) args = {'dom_id': rec['domid'], 'path': path} args.update(addl_args) - # If the 'testing_mode' attribute is set, add that to the args. - if getattr(self, 'testing_mode', False): - args['testing_mode'] = 'true' try: task = self._session.async_call_plugin(plugin, method, args) - ret = self._session.wait_for_task(0, task) + ret = self._session.wait_for_task(instance_id, task) except self.XenAPI.Failure, e: - raise RuntimeError("%s" % e.details[-1]) + ret = None + err_trace = e.details[-1] + err_msg = err_trace.splitlines()[-1] + strargs = str(args) + if 'TIMEOUT:' in err_msg: + LOG.error(_('TIMEOUT: The call to %(method)s timed out. ' + 'VM id=%(instance_id)s; args=%(strargs)s') % locals()) + else: + LOG.error(_('The call to %(method)s returned an error: %(e)s. ' + 'VM id=%(instance_id)s; args=%(strargs)s') % locals()) return ret def add_to_xenstore(self, vm, path, key, value): @@ -460,3 +523,89 @@ class VMOps(object): """Removes all data from the xenstore parameter record for this VM.""" self.write_to_param_xenstore(instance_or_vm, {}) ######################################################################## + + +def _runproc(cmd): + pipe = subprocess.PIPE + return subprocess.Popen([cmd], shell=True, stdin=pipe, stdout=pipe, + stderr=pipe, close_fds=True) + + +class SimpleDH(object): + """This class wraps all the functionality needed to implement + basic Diffie-Hellman-Merkle key exchange in Python. It features + intelligent defaults for the prime and base numbers needed for the + calculation, while allowing you to supply your own. It requires that + the openssl binary be installed on the system on which this is run, + as it uses that to handle the encryption and decryption. If openssl + is not available, a RuntimeError will be raised. + """ + def __init__(self, prime=None, base=None, secret=None): + """You can specify the values for prime and base if you wish; + otherwise, reasonable default values will be used. + """ + if prime is None: + self._prime = 162259276829213363391578010288127 + else: + self._prime = prime + if base is None: + self._base = 5 + else: + self._base = base + self._shared = self._public = None + + self._dh = M2Crypto.DH.set_params( + self.dec_to_mpi(self._prime), + self.dec_to_mpi(self._base)) + self._dh.gen_key() + self._public = self.mpi_to_dec(self._dh.pub) + + def get_public(self): + return self._public + + def compute_shared(self, other): + self._shared = self.bin_to_dec( + self._dh.compute_key(self.dec_to_mpi(other))) + return self._shared + + def mpi_to_dec(self, mpi): + bn = M2Crypto.m2.mpi_to_bn(mpi) + hexval = M2Crypto.m2.bn_to_hex(bn) + dec = int(hexval, 16) + return dec + + def bin_to_dec(self, binval): + bn = M2Crypto.m2.bin_to_bn(binval) + hexval = M2Crypto.m2.bn_to_hex(bn) + dec = int(hexval, 16) + return dec + + def dec_to_mpi(self, dec): + bn = M2Crypto.m2.dec_to_bn('%s' % dec) + mpi = M2Crypto.m2.bn_to_mpi(bn) + return mpi + + def _run_ssl(self, text, which): + base_cmd = ('cat %(tmpfile)s | openssl enc -aes-128-cbc ' + '-a -pass pass:%(shared)s -nosalt %(dec_flag)s') + if which.lower()[0] == 'd': + dec_flag = ' -d' + else: + dec_flag = '' + fd, tmpfile = tempfile.mkstemp() + os.close(fd) + file(tmpfile, 'w').write(text) + shared = self._shared + cmd = base_cmd % locals() + proc = _runproc(cmd) + proc.wait() + err = proc.stderr.read() + if err: + raise RuntimeError(_('OpenSSL error: %s') % err) + return proc.stdout.read() + + def encrypt(self, text): + return self._run_ssl(text, 'enc') + + def decrypt(self, text): + return self._run_ssl(text, 'dec') diff --git a/nova/virt/xenapi_conn.py b/nova/virt/xenapi_conn.py index 45d0738a5..689844f34 100644 --- a/nova/virt/xenapi_conn.py +++ b/nova/virt/xenapi_conn.py @@ -149,6 +149,10 @@ class XenAPIConnection(object): """Reboot VM instance""" self._vmops.reboot(instance) + def set_admin_password(self, instance, new_pass): + """Set the root/admin password on the VM instance""" + self._vmops.set_admin_password(instance, new_pass) + def destroy(self, instance): """Destroy VM instance""" self._vmops.destroy(instance) @@ -266,7 +270,8 @@ class XenAPISession(object): def _poll_task(self, id, task, done): """Poll the given XenAPI task, and fire the given action if we - get a result.""" + get a result. + """ try: name = self._session.xenapi.task.get_name_label(task) status = self._session.xenapi.task.get_status(task) diff --git a/plugins/xenserver/xenapi/etc/xapi.d/plugins/agent b/plugins/xenserver/xenapi/etc/xapi.d/plugins/agent new file mode 100755 index 000000000..12c3a19c8 --- /dev/null +++ b/plugins/xenserver/xenapi/etc/xapi.d/plugins/agent @@ -0,0 +1,126 @@ +#!/usr/bin/env python + +# Copyright (c) 2011 Citrix Systems, Inc. +# Copyright 2011 OpenStack LLC. +# Copyright 2011 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +# +# XenAPI plugin for reading/writing information to xenstore +# + +try: + import json +except ImportError: + import simplejson as json +import os +import random +import subprocess +import tempfile +import time + +import XenAPIPlugin + +from pluginlib_nova import * +configure_logging("xenstore") +import xenstore + +AGENT_TIMEOUT = 30 + + +def jsonify(fnc): + def wrapper(*args, **kwargs): + return json.dumps(fnc(*args, **kwargs)) + return wrapper + + +class TimeoutError(StandardError): + pass + + +@jsonify +def key_init(self, arg_dict): + """Handles the Diffie-Hellman key exchange with the agent to + establish the shared secret key used to encrypt/decrypt sensitive + info to be passed, such as passwords. Returns the shared + secret key value. + """ + pub = int(arg_dict["pub"]) + arg_dict["value"] = json.dumps({"name": "keyinit", "value": pub}) + request_id = arg_dict["id"] + arg_dict["path"] = "data/host/%s" % request_id + xenstore.write_record(self, arg_dict) + try: + resp = _wait_for_agent(self, request_id, arg_dict) + except TimeoutError, e: + raise PluginError("%s" % e) + return resp + + +@jsonify +def password(self, arg_dict): + """Writes a request to xenstore that tells the agent to set + the root password for the given VM. The password should be + encrypted using the shared secret key that was returned by a + previous call to key_init. The encrypted password value should + be passed as the value for the 'enc_pass' key in arg_dict. + """ + pub = int(arg_dict["pub"]) + enc_pass = arg_dict["enc_pass"] + arg_dict["value"] = json.dumps({"name": "password", "value": enc_pass}) + request_id = arg_dict["id"] + arg_dict["path"] = "data/host/%s" % request_id + xenstore.write_record(self, arg_dict) + try: + resp = _wait_for_agent(self, request_id, arg_dict) + except TimeoutError, e: + raise PluginError("%s" % e) + return resp + + +def _wait_for_agent(self, request_id, arg_dict): + """Periodically checks xenstore for a response from the agent. + The request is always written to 'data/host/{id}', and + the agent's response for that request will be in 'data/guest/{id}'. + If no value appears from the agent within the time specified by + AGENT_TIMEOUT, the original request is deleted and a TimeoutError + is returned. + """ + arg_dict["path"] = "data/guest/%s" % request_id + arg_dict["ignore_missing_path"] = True + start = time.time() + while True: + if time.time() - start > AGENT_TIMEOUT: + # No response within the timeout period; bail out + # First, delete the request record + arg_dict["path"] = "data/host/%s" % request_id + xenstore.delete_record(self, arg_dict) + raise TimeoutError("TIMEOUT: No response from agent within %s seconds." % + AGENT_TIMEOUT) + ret = xenstore.read_record(self, arg_dict) + # Note: the response for None with be a string that includes + # double quotes. + if ret != '"None"': + # The agent responded + return ret + else: + time.sleep(3) + + +if __name__ == "__main__": + XenAPIPlugin.dispatch( + {"key_init": key_init, + "password": password}) |