path: root/nova/tests/api/ec2/test_api.py

# vim: tabstop=4 shiftwidth=4 softtabstop=4

# Copyright 2010 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.

"""Unit tests for the API endpoint."""

import random
import re
import StringIO

import boto
import boto.connection
from boto.ec2 import regioninfo
from boto import exception as boto_exc
# newer versions of boto use their own wrapper on top of httplib.HTTPResponse
if hasattr(boto.connection, 'HTTPResponse'):
    httplib = boto.connection
else:
    import httplib
import fixtures
import webob

from nova.api import auth
from nova.api import ec2
from nova.api.ec2 import apirequest
from nova.api.ec2 import ec2utils
from nova import block_device
from nova import context
from nova import exception
from nova.openstack.common import timeutils
from nova.openstack.common import versionutils
from nova import test
from nova.tests import matchers


class FakeHttplibSocket(object):
    """a fake socket implementation for httplib.HTTPResponse, trivial."""
    def __init__(self, response_string):
        self.response_string = response_string
        self._buffer = StringIO.StringIO(response_string)

    def makefile(self, _mode, _other):
        """Returns the socket's internal buffer."""
        return self._buffer


class FakeHttplibConnection(object):
    """A fake httplib.HTTPConnection for boto to use

    requests made via this connection actually get translated and routed into
    our WSGI app, we then wait for the response and turn it back into
    the HTTPResponse that boto expects.
    """
    def __init__(self, app, host, is_secure=False):
        self.app = app
        self.host = host

    def request(self, method, path, data, headers):
        req = webob.Request.blank(path)
        req.method = method
        req.body = data
        req.headers = headers
        req.headers['Accept'] = 'text/html'
        req.host = self.host
        # Call the WSGI app, get the HTTP response
        resp = str(req.get_response(self.app))
        # For some reason, the response doesn't have "HTTP/1.0 " prepended; I
        # guess that's a function the web server usually provides.
        resp = "HTTP/1.0 %s" % resp
        self.sock = FakeHttplibSocket(resp)
        self.http_response = httplib.HTTPResponse(self.sock)
        # NOTE(vish): boto is accessing private variables for some reason
        self._HTTPConnection__response = self.http_response
        self.http_response.begin()

    def getresponse(self):
        return self.http_response

    def getresponsebody(self):
        return self.sock.response_string

    def close(self):
        """Required for compatibility with boto/tornado."""
        pass


class XmlConversionTestCase(test.TestCase):
    """Unit test api xml conversion."""
    def test_number_conversion(self):
        conv = ec2utils._try_convert
        self.assertIsNone(conv('None'))
        self.assertEqual(conv('True'), True)
        self.assertEqual(conv('TRUE'), True)
        self.assertEqual(conv('true'), True)
        self.assertEqual(conv('False'), False)
        self.assertEqual(conv('FALSE'), False)
        self.assertEqual(conv('false'), False)
        self.assertEqual(conv('0'), 0)
        self.assertEqual(conv('42'), 42)
        self.assertEqual(conv('3.14'), 3.14)
        self.assertEqual(conv('-57.12'), -57.12)
        self.assertEqual(conv('0x57'), 0x57)
        self.assertEqual(conv('-0x57'), -0x57)
        self.assertEqual(conv('-'), '-')
        self.assertEqual(conv('-0'), 0)
        self.assertEqual(conv('0.0'), 0.0)
        self.assertEqual(conv('1e-8'), 0.0)
        self.assertEqual(conv('-1e-8'), 0.0)
        self.assertEqual(conv('0xDD8G'), '0xDD8G')
        self.assertEqual(conv('0XDD8G'), '0XDD8G')
        self.assertEqual(conv('-stringy'), '-stringy')
        self.assertEqual(conv('stringy'), 'stringy')
        self.assertEqual(conv('add'), 'add')
        self.assertEqual(conv('remove'), 'remove')
        self.assertEqual(conv(''), '')


class Ec2utilsTestCase(test.TestCase):
    def test_ec2_id_to_id(self):
        self.assertEqual(ec2utils.ec2_id_to_id('i-0000001e'), 30)
        self.assertEqual(ec2utils.ec2_id_to_id('ami-1d'), 29)
        self.assertEqual(ec2utils.ec2_id_to_id('snap-0000001c'), 28)
        self.assertEqual(ec2utils.ec2_id_to_id('vol-0000001b'), 27)

    def test_bad_ec2_id(self):
        self.assertRaises(exception.InvalidEc2Id,
                          ec2utils.ec2_id_to_id,
                          'badone')

    def test_id_to_ec2_id(self):
        self.assertEqual(ec2utils.id_to_ec2_id(30), 'i-0000001e')
        self.assertEqual(ec2utils.id_to_ec2_id(29, 'ami-%08x'), 'ami-0000001d')
        self.assertEqual(ec2utils.id_to_ec2_snap_id(28), 'snap-0000001c')
        self.assertEqual(ec2utils.id_to_ec2_vol_id(27), 'vol-0000001b')

    def test_dict_from_dotted_str(self):
        in_str = [('BlockDeviceMapping.1.DeviceName', '/dev/sda1'),
                  ('BlockDeviceMapping.1.Ebs.SnapshotId', 'snap-0000001c'),
                  ('BlockDeviceMapping.1.Ebs.VolumeSize', '80'),
                  ('BlockDeviceMapping.1.Ebs.DeleteOnTermination', 'false'),
                  ('BlockDeviceMapping.2.DeviceName', '/dev/sdc'),
                  ('BlockDeviceMapping.2.VirtualName', 'ephemeral0')]
        expected_dict = {
            'block_device_mapping': {
            '1': {'device_name': '/dev/sda1',
                  'ebs': {'snapshot_id': 'snap-0000001c',
                          'volume_size': 80,
                          'delete_on_termination': False}},
            '2': {'device_name': '/dev/sdc',
                  'virtual_name': 'ephemeral0'}}}
        out_dict = ec2utils.dict_from_dotted_str(in_str)

        self.assertThat(out_dict, matchers.DictMatches(expected_dict))

    def test_properties_root_defice_name(self):
        mappings = [{"device": "/dev/sda1", "virtual": "root"}]
        properties0 = {'mappings': mappings}
        properties1 = {'root_device_name': '/dev/sdb', 'mappings': mappings}

        root_device_name = block_device.properties_root_device_name(
            properties0)
        self.assertEqual(root_device_name, '/dev/sda1')

        root_device_name = block_device.properties_root_device_name(
            properties1)
        self.assertEqual(root_device_name, '/dev/sdb')

    def test_regex_from_ec2_regex(self):
        def _test_re(ec2_regex, expected, literal, match=True):
            regex = ec2utils.regex_from_ec2_regex(ec2_regex)
            self.assertEqual(regex, expected)
            if match:
                self.assertIsNotNone(re.match(regex, literal))
            else:
                self.assertIsNone(re.match(regex, literal))

        # wildcards
        _test_re('foo', '\Afoo\Z(?s)', 'foo')
        _test_re('foo', '\Afoo\Z(?s)', 'baz', match=False)
        _test_re('foo?bar', '\Afoo.bar\Z(?s)', 'foo bar')
        _test_re('foo?bar', '\Afoo.bar\Z(?s)', 'foo   bar', match=False)
        _test_re('foo*bar', '\Afoo.*bar\Z(?s)', 'foo QUUX bar')

        # backslashes and escaped wildcards
        _test_re('foo\\', '\Afoo\\\\\Z(?s)', 'foo\\')
        _test_re('foo*bar', '\Afoo.*bar\Z(?s)', 'zork QUUX bar', match=False)
        _test_re('foo\\?bar', '\Afoo[?]bar\Z(?s)', 'foo?bar')
        _test_re('foo\\?bar', '\Afoo[?]bar\Z(?s)', 'foo bar', match=False)
        _test_re('foo\\*bar', '\Afoo[*]bar\Z(?s)', 'foo*bar')
        _test_re('foo\\*bar', '\Afoo[*]bar\Z(?s)', 'foo bar', match=False)

        # analog to the example given in the EC2 API docs
        ec2_regex = '\*nova\?\\end'
        expected = r'\A[*]nova[?]\\end\Z(?s)'
        literal = r'*nova?\end'
        _test_re(ec2_regex, expected, literal)

    def test_mapping_prepend_dev(self):
        mappings = [
            {'virtual': 'ami',
             'device': 'sda1'},
            {'virtual': 'root',
             'device': '/dev/sda1'},

            {'virtual': 'swap',
             'device': 'sdb1'},
            {'virtual': 'swap',
             'device': '/dev/sdb2'},

            {'virtual': 'ephemeral0',
            'device': 'sdc1'},
            {'virtual': 'ephemeral1',
             'device': '/dev/sdc1'}]
        expected_result = [
            {'virtual': 'ami',
             'device': 'sda1'},
            {'virtual': 'root',
             'device': '/dev/sda1'},

            {'virtual': 'swap',
             'device': '/dev/sdb1'},
            {'virtual': 'swap',
             'device': '/dev/sdb2'},

            {'virtual': 'ephemeral0',
             'device': '/dev/sdc1'},
            {'virtual': 'ephemeral1',
             'device': '/dev/sdc1'}]
        self.assertThat(block_device.mappings_prepend_dev(mappings),
                        matchers.DictListMatches(expected_result))


class ApiEc2TestCase(test.TestCase):
    """Unit test for the cloud controller on an EC2 API."""
    def setUp(self):
        super(ApiEc2TestCase, self).setUp()
        self.host = '127.0.0.1'
        # NOTE(vish): skipping the Authorizer
        roles = ['sysadmin', 'netadmin']
        ctxt = context.RequestContext('fake', 'fake', roles=roles)
        self.app = auth.InjectContext(ctxt, ec2.FaultWrapper(
                ec2.RequestLogging(ec2.Requestify(ec2.Authorizer(ec2.Executor()
                               ), 'nova.api.ec2.cloud.CloudController'))))
        self.useFixture(fixtures.FakeLogger('boto'))

    def expect_http(self, host=None, is_secure=False, api_version=None):
        """Returns a new EC2 connection."""
        self.ec2 = boto.connect_ec2(
                aws_access_key_id='fake',
                aws_secret_access_key='fake',
                is_secure=False,
                region=regioninfo.RegionInfo(None, 'test', self.host),
                port=8773,
                path='/services/Cloud')
        if api_version:
            self.ec2.APIVersion = api_version

        self.mox.StubOutWithMock(self.ec2, 'new_http_connection')
        self.http = FakeHttplibConnection(
                self.app, '%s:8773' % (self.host), False)
        # pylint: disable=E1103
        if versionutils.is_compatible('2.14', boto.Version, same_major=False):
            self.ec2.new_http_connection(host or self.host, 8773,
                is_secure).AndReturn(self.http)
        elif versionutils.is_compatible('2', boto.Version, same_major=False):
            self.ec2.new_http_connection(host or '%s:8773' % (self.host),
                is_secure).AndReturn(self.http)
        else:
            self.ec2.new_http_connection(host, is_secure).AndReturn(self.http)
        return self.http

    def test_return_valid_isoformat(self):
        """
            Ensure that the ec2 api returns datetime in xs:dateTime
            (which apparently isn't datetime.isoformat())
            NOTE(ken-pepple): https://bugs.launchpad.net/nova/+bug/721297
        """
        conv = apirequest._database_to_isoformat
        # sqlite database representation with microseconds
        time_to_convert = timeutils.parse_strtime("2011-02-21 20:14:10.634276",
                                                  "%Y-%m-%d %H:%M:%S.%f")
        self.assertEqual(conv(time_to_convert), '2011-02-21T20:14:10.634Z')
        # mysqlite database representation
        time_to_convert = timeutils.parse_strtime("2011-02-21 19:56:18",
                                                  "%Y-%m-%d %H:%M:%S")
        self.assertEqual(conv(time_to_convert), '2011-02-21T19:56:18.000Z')

    def test_xmlns_version_matches_request_version(self):
        self.expect_http(api_version='2010-10-30')
        self.mox.ReplayAll()

        # Any request should be fine
        self.ec2.get_all_instances()
        self.assertTrue(self.ec2.APIVersion in self.http.getresponsebody(),
                        'The version in the xmlns of the response does '
                        'not match the API version given in the request.')

    def test_describe_instances(self):
        """Test that, after creating a user and a project, the describe
        instances call to the API works properly.
        """
        self.expect_http()
        self.mox.ReplayAll()
        self.assertEqual(self.ec2.get_all_instances(), [])

    def test_terminate_invalid_instance(self):
        # Attempt to terminate an invalid instance.
        self.expect_http()
        self.mox.ReplayAll()
        self.assertRaises(boto_exc.EC2ResponseError,
                self.ec2.terminate_instances, "i-00000005")

    def test_get_all_key_pairs(self):
        """Test that, after creating a user and project and generating
         a key pair, that the API call to list key pairs works properly.
         """
        keyname = "".join(random.choice("sdiuisudfsdcnpaqwertasd")
                          for x in range(random.randint(4, 8)))
        self.expect_http()
        self.mox.ReplayAll()
        self.ec2.create_key_pair(keyname)
        rv = self.ec2.get_all_key_pairs()
        results = [k for k in rv if k.name == keyname]
        self.assertEqual(len(results), 1)

    def test_create_duplicate_key_pair(self):
        """Test that, after successfully generating a keypair,
        requesting a second keypair with the same name fails sanely.
        """
        self.expect_http()
        self.mox.ReplayAll()
        self.ec2.create_key_pair('test')

        try:
            self.ec2.create_key_pair('test')
        except boto_exc.EC2ResponseError as e:
            if e.code == 'InvalidKeyPair.Duplicate':
                pass
            else:
                self.assertEqual('InvalidKeyPair.Duplicate', e.code)
        else:
            self.fail('Exception not raised.')

    def test_get_all_security_groups(self):
        # Test that we can retrieve security groups.
        self.expect_http()
        self.mox.ReplayAll()

        rv = self.ec2.get_all_security_groups()

        self.assertEqual(len(rv), 1)
        self.assertEqual(rv[0].name, 'default')

    def test_create_delete_security_group(self):
        # Test that we can create a security group.
        self.expect_http()
        self.mox.ReplayAll()

        security_group_name = "".join(random.choice("sdiuisudfsdcnpaqwertasd")
                                      for x in range(random.randint(4, 8)))

        self.ec2.create_security_group(security_group_name, 'test group')

        self.expect_http()
        self.mox.ReplayAll()

        rv = self.ec2.get_all_security_groups()
        self.assertEqual(len(rv), 2)
        self.assertIn(security_group_name, [group.name for group in rv])

        self.expect_http()
        self.mox.ReplayAll()

        self.ec2.delete_security_group(security_group_name)

    def test_group_name_valid_chars_security_group(self):
        """Test that we sanely handle invalid security group names.

         EC2 API Spec states we should only accept alphanumeric characters,
         spaces, dashes, and underscores. Amazon implementation
         accepts more characters - so, [:print:] is ok.
         """
        bad_strict_ec2 = "aa \t\x01\x02\x7f"
        bad_amazon_ec2 = "aa #^% -=99"
        test_raise = [
            (True, bad_amazon_ec2, "test desc"),
            (True, "test name", bad_amazon_ec2),
            (False, bad_strict_ec2, "test desc"),
        ]
        for test in test_raise:
            self.expect_http()
            self.mox.ReplayAll()
            self.flags(ec2_strict_validation=test[0])
            self.assertRaises(boto_exc.EC2ResponseError,
                              self.ec2.create_security_group,
                              test[1],
                              test[2])
        test_accept = [
            (False, bad_amazon_ec2, "test desc"),
            (False, "test name", bad_amazon_ec2),
        ]
        for test in test_accept:
            self.expect_http()
            self.mox.ReplayAll()
            self.flags(ec2_strict_validation=test[0])
            self.ec2.create_security_group(test[1], test[2])
            self.expect_http()
            self.mox.ReplayAll()
            self.ec2.delete_security_group(test[1])

    def test_group_name_valid_length_security_group(self):
        """Test that we sanely handle invalid security group names.

         API Spec states that the length should not exceed 255 char.
        """
        self.expect_http()
        self.mox.ReplayAll()

        # Test block group_name > 255 chars
        security_group_name = "".join(random.choice("poiuytrewqasdfghjklmnbvc")
                                      for x in range(random.randint(256, 266)))

        self.assertRaises(boto_exc.EC2ResponseError,
                self.ec2.create_security_group,
                security_group_name,
                'test group')

    def test_authorize_revoke_security_group_cidr(self):
        """
        Test that we can add and remove CIDR based rules
        to a security group
        """
        self.expect_http()
        self.mox.ReplayAll()

        security_group_name = "".join(random.choice("sdiuisudfsdcnpaqwertasd")
                                      for x in range(random.randint(4, 8)))

        group = self.ec2.create_security_group(security_group_name,
                                               'test group')

        self.expect_http()
        self.mox.ReplayAll()
        group.connection = self.ec2

        group.authorize('tcp', 80, 81, '0.0.0.0/0')
        group.authorize('icmp', -1, -1, '0.0.0.0/0')
        group.authorize('udp', 80, 81, '0.0.0.0/0')
        group.authorize('tcp', 1, 65535, '0.0.0.0/0')
        group.authorize('udp', 1, 65535, '0.0.0.0/0')
        group.authorize('icmp', 1, 0, '0.0.0.0/0')
        group.authorize('icmp', 0, 1, '0.0.0.0/0')
        group.authorize('icmp', 0, 0, '0.0.0.0/0')

        def _assert(message, *args):
            try:
                group.authorize(*args)
            except boto_exc.EC2ResponseError as e:
                self.assertEqual(e.status, 400, 'Expected status to be 400')
                self.assertIn(message, e.error_message)
            else:
                raise self.failureException, 'EC2ResponseError not raised'

        # Invalid CIDR address
        _assert('Invalid CIDR', 'tcp', 80, 81, '0.0.0.0/0444')
        # Missing ports
        _assert('Not enough parameters', 'tcp', '0.0.0.0/0')
        # from port cannot be greater than to port
        _assert('Invalid port range', 'tcp', 100, 1, '0.0.0.0/0')
        # For tcp, negative values are not allowed
        _assert('Invalid port range', 'tcp', -1, 1, '0.0.0.0/0')
        # For tcp, valid port range 1-65535
        _assert('Invalid port range', 'tcp', 1, 65599, '0.0.0.0/0')
        # Invalid Cidr for ICMP type
        _assert('Invalid CIDR', 'icmp', -1, -1, '0.0.444.0/4')
        # Invalid protocol
        _assert('Invalid IP protocol', 'xyz', 1, 14, '0.0.0.0/0')
        # Invalid port
        _assert('Invalid input received: To and From ports must be integers',
                'tcp', " ", "81", '0.0.0.0/0')
        # Invalid icmp port
        _assert('Invalid input received: '
                'Type and Code must be integers for ICMP protocol type',
                'icmp', " ", "81", '0.0.0.0/0')
        # Invalid CIDR Address
        _assert('Invalid CIDR', 'icmp', -1, -1, '0.0.0.0')
        # Invalid CIDR Address
        _assert('Invalid CIDR', 'icmp', -1, -1, '0.0.0.0/')
        # Invalid Cidr ports
        _assert('Invalid port range', 'icmp', 1, 256, '0.0.0.0/0')

        self.expect_http()
        self.mox.ReplayAll()

        rv = self.ec2.get_all_security_groups()

        group = [grp for grp in rv if grp.name == security_group_name][0]

        self.assertEqual(len(group.rules), 8)
        self.assertEqual(int(group.rules[0].from_port), 80)
        self.assertEqual(int(group.rules[0].to_port), 81)
        self.assertEqual(len(group.rules[0].grants), 1)
        self.assertEqual(str(group.rules[0].grants[0]), '0.0.0.0/0')

        self.expect_http()
        self.mox.ReplayAll()
        group.connection = self.ec2

        group.revoke('tcp', 80, 81, '0.0.0.0/0')
        group.revoke('icmp', -1, -1, '0.0.0.0/0')
        group.revoke('udp', 80, 81, '0.0.0.0/0')
        group.revoke('tcp', 1, 65535, '0.0.0.0/0')
        group.revoke('udp', 1, 65535, '0.0.0.0/0')
        group.revoke('icmp', 1, 0, '0.0.0.0/0')
        group.revoke('icmp', 0, 1, '0.0.0.0/0')
        group.revoke('icmp', 0, 0, '0.0.0.0/0')

        self.expect_http()
        self.mox.ReplayAll()

        self.ec2.delete_security_group(security_group_name)

        self.expect_http()
        self.mox.ReplayAll()
        group.connection = self.ec2

        rv = self.ec2.get_all_security_groups()

        self.assertEqual(len(rv), 1)
        self.assertEqual(rv[0].name, 'default')

    def test_authorize_revoke_security_group_cidr_v6(self):
        """
        Test that we can add and remove CIDR based rules
        to a security group for IPv6
        """
        self.expect_http()
        self.mox.ReplayAll()

        security_group_name = "".join(random.choice("sdiuisudfsdcnpaqwertasd")
                                      for x in range(random.randint(4, 8)))

        group = self.ec2.create_security_group(security_group_name,
                                               'test group')

        self.expect_http()
        self.mox.ReplayAll()
        group.connection = self.ec2

        group.authorize('tcp', 80, 81, '::/0')

        self.expect_http()
        self.mox.ReplayAll()

        rv = self.ec2.get_all_security_groups()

        group = [grp for grp in rv if grp.name == security_group_name][0]
        self.assertEqual(len(group.rules), 1)
        self.assertEqual(int(group.rules[0].from_port), 80)
        self.assertEqual(int(group.rules[0].to_port), 81)
        self.assertEqual(len(group.rules[0].grants), 1)
        self.assertEqual(str(group.rules[0].grants[0]), '::/0')

        self.expect_http()
        self.mox.ReplayAll()
        group.connection = self.ec2

        group.revoke('tcp', 80, 81, '::/0')

        self.expect_http()
        self.mox.ReplayAll()

        self.ec2.delete_security_group(security_group_name)

        self.expect_http()
        self.mox.ReplayAll()
        group.connection = self.ec2

        rv = self.ec2.get_all_security_groups()

        self.assertEqual(len(rv), 1)
        self.assertEqual(rv[0].name, 'default')

    def test_authorize_revoke_security_group_foreign_group(self):
        """
        Test that we can grant and revoke another security group access
        to a security group
        """
        self.expect_http()
        self.mox.ReplayAll()

        rand_string = 'sdiuisudfsdcnpaqwertasd'
        security_group_name = "".join(random.choice(rand_string)
                                      for x in range(random.randint(4, 8)))
        other_security_group_name = "".join(random.choice(rand_string)
                                      for x in range(random.randint(4, 8)))

        group = self.ec2.create_security_group(security_group_name,
                                               'test group')

        self.expect_http()
        self.mox.ReplayAll()

        other_group = self.ec2.create_security_group(other_security_group_name,
                                                     'some other group')

        self.expect_http()
        self.mox.ReplayAll()
        group.connection = self.ec2

        group.authorize(src_group=other_group)

        self.expect_http()
        self.mox.ReplayAll()

        rv = self.ec2.get_all_security_groups()

        # I don't bother checkng that we actually find it here,
        # because the create/delete unit test further up should
        # be good enough for that.
        for group in rv:
            if group.name == security_group_name:
                self.assertEqual(len(group.rules), 3)
                self.assertEqual(len(group.rules[0].grants), 1)
                self.assertEqual(str(group.rules[0].grants[0]),
                                 '%s-%s' % (other_security_group_name, 'fake'))

        self.expect_http()
        self.mox.ReplayAll()

        rv = self.ec2.get_all_security_groups()

        for group in rv:
            if group.name == security_group_name:
                self.expect_http()
                self.mox.ReplayAll()
                group.connection = self.ec2
                group.revoke(src_group=other_group)

        self.expect_http()
        self.mox.ReplayAll()

        self.ec2.delete_security_group(security_group_name)
        self.ec2.delete_security_group(other_security_group_name)