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


import json

from oslo_config import cfg
from oslo_utils import importutils
import requests
import six

from heat.api.aws import ec2token
from heat.api.aws import exception
from heat.common import wsgi
from heat.tests import common


class Ec2TokenTest(common.HeatTestCase):
    '''
    Tests the Ec2Token middleware
    '''

    def setUp(self):
        super(Ec2TokenTest, self).setUp()
        self.m.StubOutWithMock(requests, 'post')

    def _dummy_GET_request(self, params=None, environ=None):
        # Mangle the params dict into a query string
        params = params or {}
        environ = environ or {}
        qs = "&".join(["=".join([k, str(params[k])]) for k in params])
        environ.update({'REQUEST_METHOD': 'GET', 'QUERY_STRING': qs})
        req = wsgi.Request(environ)
        return req

    def test_conf_get_paste(self):
        dummy_conf = {'auth_uri': 'http://192.0.2.9/v2.0'}
        ec2 = ec2token.EC2Token(app=None, conf=dummy_conf)
        self.assertEqual('http://192.0.2.9/v2.0', ec2._conf_get('auth_uri'))
        self.assertEqual(
            'http://192.0.2.9/v2.0/ec2tokens',
            ec2._conf_get_keystone_ec2_uri('http://192.0.2.9/v2.0'))

    def test_conf_get_opts(self):
        cfg.CONF.set_default('auth_uri', 'http://192.0.2.9/v2.0/',
                             group='ec2authtoken')
        ec2 = ec2token.EC2Token(app=None, conf={})
        self.assertEqual('http://192.0.2.9/v2.0/', ec2._conf_get('auth_uri'))
        self.assertEqual(
            'http://192.0.2.9/v2.0/ec2tokens',
            ec2._conf_get_keystone_ec2_uri('http://192.0.2.9/v2.0/'))

    def test_conf_get_ssl_default_options(self):
        ec2 = ec2token.EC2Token(app=None, conf={})
        self.assertTrue(ec2.ssl_options['verify'],
                        "SSL verify should be True by default")
        self.assertIsNone(ec2.ssl_options['cert'],
                          "SSL client cert should be None by default")

    def test_conf_ssl_insecure_option(self):
        ec2 = ec2token.EC2Token(app=None, conf={})
        cfg.CONF.set_default('insecure', 'True', group='ec2authtoken')
        cfg.CONF.set_default('ca_file', None, group='ec2authtoken')
        self.assertFalse(ec2.ssl_options['verify'])

    def test_conf_get_ssl_opts(self):
        cfg.CONF.set_default('auth_uri', 'https://192.0.2.9/v2.0/',
                             group='ec2authtoken')
        cfg.CONF.set_default('ca_file', '/home/user/cacert.pem',
                             group='ec2authtoken')
        cfg.CONF.set_default('insecure', 'false', group='ec2authtoken')
        cfg.CONF.set_default('cert_file', '/home/user/mycert',
                             group='ec2authtoken')
        cfg.CONF.set_default('key_file', '/home/user/mykey',
                             group='ec2authtoken')
        ec2 = ec2token.EC2Token(app=None, conf={})
        self.assertEqual('/home/user/cacert.pem', ec2.ssl_options['verify'])
        self.assertEqual(('/home/user/mycert', '/home/user/mykey'),
                         ec2.ssl_options['cert'])

    def test_get_signature_param_old(self):
        params = {'Signature': 'foo'}
        dummy_req = self._dummy_GET_request(params)
        ec2 = ec2token.EC2Token(app=None, conf={})
        self.assertEqual('foo', ec2._get_signature(dummy_req))

    def test_get_signature_param_new(self):
        params = {'X-Amz-Signature': 'foo'}
        dummy_req = self._dummy_GET_request(params)
        ec2 = ec2token.EC2Token(app=None, conf={})
        self.assertEqual('foo', ec2._get_signature(dummy_req))

    def test_get_signature_header_space(self):
        req_env = {'HTTP_AUTHORIZATION':
                   ('Authorization: foo  Credential=foo/bar, '
                    'SignedHeaders=content-type;host;x-amz-date, '
                    'Signature=xyz')}
        dummy_req = self._dummy_GET_request(environ=req_env)
        ec2 = ec2token.EC2Token(app=None, conf={})
        self.assertEqual('xyz', ec2._get_signature(dummy_req))

    def test_get_signature_header_notlast(self):
        req_env = {'HTTP_AUTHORIZATION':
                   ('Authorization: foo  Credential=foo/bar, '
                    'Signature=xyz,'
                    'SignedHeaders=content-type;host;x-amz-date ')}
        dummy_req = self._dummy_GET_request(environ=req_env)
        ec2 = ec2token.EC2Token(app=None, conf={})
        self.assertEqual('xyz', ec2._get_signature(dummy_req))

    def test_get_signature_header_nospace(self):
        req_env = {'HTTP_AUTHORIZATION':
                   ('Authorization: foo  Credential=foo/bar,'
                    'SignedHeaders=content-type;host;x-amz-date,'
                    'Signature=xyz')}
        dummy_req = self._dummy_GET_request(environ=req_env)
        ec2 = ec2token.EC2Token(app=None, conf={})
        self.assertEqual('xyz', ec2._get_signature(dummy_req))

    def test_get_access_param_old(self):
        params = {'AWSAccessKeyId': 'foo'}
        dummy_req = self._dummy_GET_request(params)
        ec2 = ec2token.EC2Token(app=None, conf={})
        self.assertEqual('foo', ec2._get_access(dummy_req))

    def test_get_access_param_new(self):
        params = {'X-Amz-Credential': 'foo/bar'}
        dummy_req = self._dummy_GET_request(params)
        ec2 = ec2token.EC2Token(app=None, conf={})
        self.assertEqual('foo', ec2._get_access(dummy_req))

    def test_get_access_header_space(self):
        req_env = {'HTTP_AUTHORIZATION':
                   ('Authorization: foo  Credential=foo/bar, '
                    'SignedHeaders=content-type;host;x-amz-date, '
                    'Signature=xyz')}
        dummy_req = self._dummy_GET_request(environ=req_env)
        ec2 = ec2token.EC2Token(app=None, conf={})
        self.assertEqual('foo', ec2._get_access(dummy_req))

    def test_get_access_header_nospace(self):
        req_env = {'HTTP_AUTHORIZATION':
                   ('Authorization: foo  Credential=foo/bar,'
                    'SignedHeaders=content-type;host;x-amz-date,'
                    'Signature=xyz')}
        dummy_req = self._dummy_GET_request(environ=req_env)
        ec2 = ec2token.EC2Token(app=None, conf={})
        self.assertEqual('foo', ec2._get_access(dummy_req))

    def test_get_access_header_last(self):
        req_env = {'HTTP_AUTHORIZATION':
                   ('Authorization: foo '
                    'SignedHeaders=content-type;host;x-amz-date,'
                    'Signature=xyz,Credential=foo/bar')}
        dummy_req = self._dummy_GET_request(environ=req_env)
        ec2 = ec2token.EC2Token(app=None, conf={})
        self.assertEqual('foo', ec2._get_access(dummy_req))

    def test_call_x_auth_user(self):
        req_env = {'HTTP_X_AUTH_USER': 'foo'}
        dummy_req = self._dummy_GET_request(environ=req_env)
        ec2 = ec2token.EC2Token(app='xyz', conf={})
        self.assertEqual('xyz', ec2.__call__(dummy_req))

    def test_call_auth_nosig(self):
        req_env = {'HTTP_AUTHORIZATION':
                   ('Authorization: foo  Credential=foo/bar, '
                    'SignedHeaders=content-type;host;x-amz-date')}
        dummy_req = self._dummy_GET_request(environ=req_env)
        ec2 = ec2token.EC2Token(app='xyz', conf={})
        self.assertRaises(exception.HeatIncompleteSignatureError,
                          ec2.__call__, dummy_req)

    def test_call_auth_nouser(self):
        req_env = {'HTTP_AUTHORIZATION':
                   ('Authorization: foo '
                    'SignedHeaders=content-type;host;x-amz-date,'
                    'Signature=xyz')}
        dummy_req = self._dummy_GET_request(environ=req_env)
        ec2 = ec2token.EC2Token(app='xyz', conf={})
        self.assertRaises(exception.HeatMissingAuthenticationTokenError,
                          ec2.__call__, dummy_req)

    def test_call_auth_noaccess(self):
        # If there's no accesskey in params or header, but there is a
        # Signature, we expect HeatMissingAuthenticationTokenError
        params = {'Signature': 'foo'}
        dummy_req = self._dummy_GET_request(params)
        ec2 = ec2token.EC2Token(app='xyz', conf={})
        self.assertRaises(exception.HeatMissingAuthenticationTokenError,
                          ec2.__call__, dummy_req)

    def test_call_x_auth_nouser_x_auth_user(self):
        req_env = {'HTTP_X_AUTH_USER': 'foo',
                   'HTTP_AUTHORIZATION':
                   ('Authorization: foo '
                    'SignedHeaders=content-type;host;x-amz-date,'
                    'Signature=xyz')}
        dummy_req = self._dummy_GET_request(environ=req_env)
        ec2 = ec2token.EC2Token(app='xyz', conf={})
        self.assertEqual('xyz', ec2.__call__(dummy_req))

    def _stub_http_connection(self, headers=None, params=None, response=None,
                              req_url='http://123:5000/v2.0/ec2tokens',
                              verify=True, cert=None):

        headers = headers or {}
        params = params or {}

        class DummyHTTPResponse(object):
            text = response

            def json(self):
                return json.loads(self.text)

        body_hash = ('e3b0c44298fc1c149afbf4c8996fb9'
                     '2427ae41e4649b934ca495991b7852b855')
        req_creds = json.dumps({"ec2Credentials":
                                {"access": "foo",
                                 "headers": headers,