# Copyright (c) 2011-2014 Greg Holt
# Copyright (c) 2012-2013 Peter Portante
# Copyright (c) 2012 Iryoung Jeong
# Copyright (c) 2012 Michael Barton
# Copyright (c) 2013 Alex Gaynor
# Copyright (c) 2013 Chuck Thier
# Copyright (c) 2013 David Goetz
# Copyright (c) 2015 Donagh McCabe
# Copyright (c) 2013 Greg Lange
# Copyright (c) 2013 John Dickinson
# Copyright (c) 2013 Kun Huang
# Copyright (c) 2013 Richard Hawkins
# Copyright (c) 2013 Samuel Merritt
# Copyright (c) 2013 Shri Javadekar
# Copyright (c) 2013 Tong Li
# Copyright (c) 2013 ZhiQiang Fan
#
# 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 base64
import hmac
import itertools
import mock
import unittest
import hashlib
import six
from time import time, strftime, gmtime

from swift.common.middleware import tempauth, tempurl
from swift.common.header_key_dict import HeaderKeyDict
from swift.common.swob import Request, Response
from swift.common import utils


class FakeApp(object):

    def __init__(self, status_headers_body_iter=None):
        self.calls = 0
        self.status_headers_body_iter = status_headers_body_iter
        if not self.status_headers_body_iter:
            self.status_headers_body_iter = iter(
                itertools.repeat((
                    '404 Not Found', {
                        'x-test-header-one-a': 'value1',
                        'x-test-header-two-a': 'value2',
                        'x-test-header-two-b': 'value3'},
                    '')))
        self.request = None

    def __call__(self, env, start_response):
        self.calls += 1
        self.request = Request.blank('', environ=env)
        if 'swift.authorize' in env:
            resp = env['swift.authorize'](self.request)
            if resp:
                return resp(env, start_response)
        status, headers, body = next(self.status_headers_body_iter)
        return Response(status=status, headers=headers,
                        body=body)(env, start_response)


class TestTempURL(unittest.TestCase):

    def setUp(self):
        self.app = FakeApp()
        self.auth = tempauth.filter_factory({'reseller_prefix': ''})(self.app)
        self.tempurl = tempurl.filter_factory({})(self.auth)

    def _make_request(self, path, environ=None, keys=(), container_keys=None,
                      **kwargs):
        if environ is None:
            environ = {}

        _junk, account, _junk, _junk = utils.split_path(path, 2, 4, True)
        self._fake_cache_environ(environ, account, keys,
                                 container_keys=container_keys)
        req = Request.blank(path, environ=environ, **kwargs)
        return req

    def _fake_cache_environ(self, environ, account, keys, container_keys=None):
        """
        Fake out the caching layer for get_account_info(). Injects account data
        into environ such that keys are the tempurl keys, if set.
        """
        meta = {'swash': 'buckle'}
        for idx, key in enumerate(keys):
            meta_name = 'Temp-URL-key' + (("-%d" % (idx + 1) if idx else ""))
            if key:
                meta[meta_name] = key

        ic = environ.setdefault('swift.infocache', {})
        ic['account/' + account] = {
            'status': 204,
            'container_count': '0',
            'total_object_count': '0',
            'bytes': '0',
            'meta': meta}

        meta = {}
        for i, key in enumerate(container_keys or []):
            meta_name = 'Temp-URL-key' + (("-%d" % (i + 1) if i else ""))
            meta[meta_name] = key

        container_cache_key = 'container/' + account + '/c'
        ic.setdefault(container_cache_key, {'meta': meta})

    def test_passthrough(self):
        resp = self._make_request('/v1/a/c/o').get_response(self.tempurl)
        self.assertEqual(resp.status_int, 401)
        self.assertNotIn(b'Temp URL invalid', resp.body)

    def test_allow_options(self):
        self.app.status_headers_body_iter = iter([('200 Ok', {}, '')])
        resp = self._make_request(
            '/v1/a/c/o?temp_url_sig=abcde&temp_url_expires=12345',
            environ={'REQUEST_METHOD': 'OPTIONS'}).get_response(self.tempurl)
        self.assertEqual(resp.status_int, 200)

    def assert_valid_sig(self, expires, path, keys, sig, environ=None,
                         prefix=None):
        if not environ:
            environ = {}
        if six.PY3 and isinstance(sig, six.binary_type):
            sig = sig.decode('utf-8')
        environ['QUERY_STRING'] = 'temp_url_sig=%s&temp_url_expires=%s' % (
            sig.replace('+', '%2B'), expires)
        if prefix is not None:
            environ['QUERY_STRING'] += '&temp_url_prefix=%s' % prefix
        req = self._make_request(path, keys=keys, environ=environ)
        self.tempurl.app = FakeApp(iter([('200 Ok', (), '123')]))
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 200)
        self.assertEqual(resp.headers['content-disposition'],
                         'attachment; filename="o"; ' + "filename*=UTF-8''o")
        self.assertEqual(resp.headers['expires'],
                         strftime('%a, %d %b %Y %H:%M:%S GMT',
                                  gmtime(expires)))
        self.assertEqual(req.environ['swift.authorize_override'], True)
        self.assertEqual(req.environ['REMOTE_USER'], '.wsgi.tempurl')

    def test_get_valid(self):
        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        self.assert_valid_sig(expires, path, [key], sig)

        sig = hmac.new(key, hmac_body, hashlib.sha256).hexdigest()
        self.assert_valid_sig(expires, path, [key], sig)

        sig = base64.b64encode(hmac.new(
            key, hmac_body, hashlib.sha256).digest())
        self.assert_valid_sig(expires, path, [key], b'sha256:' + sig)

        sig = base64.b64encode(hmac.new(
            key, hmac_body, hashlib.sha512).digest())
        self.assert_valid_sig(expires, path, [key], b'sha512:' + sig)

    def test_get_valid_key2(self):
        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key1 = b'abc123'
        key2 = b'def456'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig1 = hmac.new(key1, hmac_body, hashlib.sha1).hexdigest()
        sig2 = hmac.new(key2, hmac_body, hashlib.sha1).hexdigest()
        for sig in (sig1, sig2):
            self.assert_valid_sig(expires, path, [key1, key2], sig)

    def test_get_valid_container_keys(self):
        ic = {}
        environ = {'swift.infocache': ic}
        # Add two static container keys
        container_keys = ['me', 'other']
        meta = {}
        for idx, key in enumerate(container_keys):
            meta_name = 'Temp-URL-key' + (("-%d" % (idx + 1) if idx else ""))
            if key:
                meta[meta_name] = key
        ic['container/a/c'] = {'meta': meta}

        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key1 = b'me'
        key2 = b'other'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig1 = hmac.new(key1, hmac_body, hashlib.sha1).hexdigest()
        sig2 = hmac.new(key2, hmac_body, hashlib.sha1).hexdigest()
        account_keys = []
        for sig in (sig1, sig2):
            self.assert_valid_sig(expires, path, account_keys, sig, environ)

    @mock.patch('swift.common.middleware.tempurl.time', return_value=0)
    def test_get_valid_with_filename(self, mock_time):
        method = 'GET'
        expires = (((24 + 1) * 60 + 1) * 60) + 1
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(path, keys=[key], environ={
            'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s&'
            'filename=bob%%20%%22killer%%22.txt' % (sig, expires)})
        self.tempurl.app = FakeApp(iter([('200 Ok', (), '123')]))
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 200)
        self.assertEqual(resp.headers['content-disposition'],
                         'attachment; filename="bob %22killer%22.txt"; ' +
                         "filename*=UTF-8''bob%20%22killer%22.txt")
        self.assertIn('expires', resp.headers)
        self.assertEqual('Fri, 02 Jan 1970 01:01:01 GMT',
                         resp.headers['expires'])
        self.assertEqual(req.environ['swift.authorize_override'], True)
        self.assertEqual(req.environ['REMOTE_USER'], '.wsgi.tempurl')

    def test_head_valid_with_filename(self):
        method = 'HEAD'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(path, keys=[key], environ={
            'REQUEST_METHOD': 'HEAD',
            'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s&'
            'filename=bob_killer.txt' % (sig, expires)})
        self.tempurl.app = FakeApp(iter([('200 Ok', (), '123')]))
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 200)
        self.assertEqual(resp.headers['content-disposition'],
                         'attachment; filename="bob_killer.txt"; ' +
                         "filename*=UTF-8''bob_killer.txt")

    def test_head_and_get_headers_match(self):
        method = 'HEAD'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(path, keys=[key], environ={
            'REQUEST_METHOD': 'HEAD',
            'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s'
            % (sig, expires)})
        self.tempurl.app = FakeApp(iter([('200 Ok', (), '123')]))
        resp = req.get_response(self.tempurl)

        get_method = 'GET'
        get_hmac_body = ('%s\n%i\n%s' %
                         (get_method, expires, path)).encode('utf-8')
        get_sig = hmac.new(key, get_hmac_body, hashlib.sha1).hexdigest()
        get_req = self._make_request(path, keys=[key], environ={
            'REQUEST_METHOD': 'GET',
            'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s'
            % (get_sig, expires)})
        self.tempurl.app = FakeApp(iter([('200 Ok', (), '123')]))
        get_resp = get_req.get_response(self.tempurl)
        self.assertEqual(resp.headers, get_resp.headers)

    @mock.patch('swift.common.middleware.tempurl.time', return_value=0)
    def test_get_valid_with_filename_and_inline(self, mock_time):
        method = 'GET'
        expires = 1
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(path, keys=[key], environ={
            'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s&'
            'filename=bob%%20%%22killer%%22.txt&inline=' % (sig, expires)})
        self.tempurl.app = FakeApp(iter([('200 Ok', (), '123')]))
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 200)
        self.assertEqual(resp.headers['content-disposition'],
                         'inline; filename="bob %22killer%22.txt"; ' +
                         "filename*=UTF-8''bob%20%22killer%22.txt")
        self.assertIn('expires', resp.headers)
        self.assertEqual('Thu, 01 Jan 1970 00:00:01 GMT',
                         resp.headers['expires'])
        self.assertEqual(req.environ['swift.authorize_override'], True)
        self.assertEqual(req.environ['REMOTE_USER'], '.wsgi.tempurl')

    def test_get_valid_with_inline(self):
        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(path, keys=[key], environ={
            'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s&'
            'inline=' % (sig, expires)})
        self.tempurl.app = FakeApp(iter([('200 Ok', (), '123')]))
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 200)
        self.assertEqual(resp.headers['content-disposition'], 'inline')
        self.assertIn('expires', resp.headers)
        self.assertEqual(req.environ['swift.authorize_override'], True)
        self.assertEqual(req.environ['REMOTE_USER'], '.wsgi.tempurl')

    def test_get_valid_with_prefix(self):
        method = 'GET'
        expires = int(time() + 86400)
        prefix = 'p1/p2/'
        sig_path = 'prefix:/v1/a/c/' + prefix
        query_path = '/v1/a/c/' + prefix + 'o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' %
                     (method, expires, sig_path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        self.assert_valid_sig(expires, query_path, [key], sig, prefix=prefix)

        query_path = query_path[:-1] + 'p3/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' %
                     (method, expires, sig_path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        self.assert_valid_sig(expires, query_path, [key], sig, prefix=prefix)

    def test_get_valid_with_prefix_empty(self):
        method = 'GET'
        expires = int(time() + 86400)
        sig_path = 'prefix:/v1/a/c/'
        query_path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' %
                     (method, expires, sig_path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        self.assert_valid_sig(expires, query_path, [key], sig, prefix='')

    def test_obj_odd_chars(self):
        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/a\r\nb'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(path, keys=[key], environ={
            'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                sig, expires)})
        self.tempurl.app = FakeApp(iter([('200 Ok', (), '123')]))
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 200)
        self.assertEqual(resp.headers['content-disposition'],
                         'attachment; filename="a%0D%0Ab"; ' +
                         "filename*=UTF-8''a%0D%0Ab")
        self.assertIn('expires', resp.headers)
        self.assertEqual(req.environ['swift.authorize_override'], True)
        self.assertEqual(req.environ['REMOTE_USER'], '.wsgi.tempurl')

    def test_obj_odd_chars_in_content_disposition_metadata(self):
        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(path, keys=[key], environ={
            'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                sig, expires)})
        headers = [('Content-Disposition', 'attachment; filename="fu\nbar"')]
        self.tempurl.app = FakeApp(iter([('200 Ok', headers, '123')]))
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 200)
        self.assertEqual(resp.headers['content-disposition'],
                         'attachment; filename="fu%0Abar"')
        self.assertIn('expires', resp.headers)
        self.assertEqual(req.environ['swift.authorize_override'], True)
        self.assertEqual(req.environ['REMOTE_USER'], '.wsgi.tempurl')

    def test_obj_trailing_slash(self):
        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/o/'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(path, keys=[key], environ={
            'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                sig, expires)})
        self.tempurl.app = FakeApp(iter([('200 Ok', (), '123')]))
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 200)
        self.assertEqual(resp.headers['content-disposition'],
                         'attachment; filename="o"; ' +
                         "filename*=UTF-8''o")
        self.assertIn('expires', resp.headers)
        self.assertEqual(req.environ['swift.authorize_override'], True)
        self.assertEqual(req.environ['REMOTE_USER'], '.wsgi.tempurl')

    def test_filename_trailing_slash(self):
        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(path, keys=[key], environ={
            'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s&'
            'filename=/i/want/this/just/as/it/is/' % (sig, expires)})
        self.tempurl.app = FakeApp(iter([('200 Ok', (), '123')]))
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 200)
        self.assertEqual(
            resp.headers['content-disposition'],
            'attachment; filename="/i/want/this/just/as/it/is/"; ' +
            "filename*=UTF-8''/i/want/this/just/as/it/is/")
        self.assertIn('expires', resp.headers)
        self.assertEqual(req.environ['swift.authorize_override'], True)
        self.assertEqual(req.environ['REMOTE_USER'], '.wsgi.tempurl')

    def test_get_valid_but_404(self):
        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(
            path, keys=[key],
            environ={'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                sig, expires)})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 404)
        self.assertNotIn('content-disposition', resp.headers)
        self.assertNotIn('expires', resp.headers)
        self.assertEqual(req.environ['swift.authorize_override'], True)
        self.assertEqual(req.environ['REMOTE_USER'], '.wsgi.tempurl')

    def test_put_not_allowed_by_get(self):
        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(
            path, keys=[key],
            environ={'REQUEST_METHOD': 'PUT',
                     'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                         sig, expires)})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 401)
        self.assertIn(b'Temp URL invalid', resp.body)
        self.assertIn('Www-Authenticate', resp.headers)

    def test_put_valid(self):
        method = 'PUT'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(
            path, keys=[key],
            environ={'REQUEST_METHOD': 'PUT',
                     'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                         sig, expires)})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 404)
        self.assertEqual(req.environ['swift.authorize_override'], True)
        self.assertEqual(req.environ['REMOTE_USER'], '.wsgi.tempurl')

    def test_get_not_allowed_by_put(self):
        method = 'PUT'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(
            path, keys=[key],
            environ={'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                sig, expires)})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 401)
        self.assertIn(b'Temp URL invalid', resp.body)
        self.assertIn('Www-Authenticate', resp.headers)

    def test_missing_sig(self):
        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(
            path, keys=[key],
            environ={'QUERY_STRING': 'temp_url_expires=%s' % expires})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 401)
        self.assertIn(b'Temp URL invalid', resp.body)
        self.assertIn('Www-Authenticate', resp.headers)

    def test_missing_expires(self):
        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(
            path, keys=[key],
            environ={'QUERY_STRING': 'temp_url_sig=%s' % sig})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 401)
        self.assertIn(b'Temp URL invalid', resp.body)
        self.assertIn('Www-Authenticate', resp.headers)

    def test_bad_path(self):
        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(
            path, keys=[key],
            environ={'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                sig, expires)})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 401)
        self.assertIn(b'Temp URL invalid', resp.body)
        self.assertIn('Www-Authenticate', resp.headers)

    def test_no_key(self):
        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(
            path, keys=[],
            environ={'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                sig, expires)})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 401)
        self.assertIn(b'Temp URL invalid', resp.body)
        self.assertIn('Www-Authenticate', resp.headers)

    def test_head_allowed_by_get(self):
        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(
            path, keys=[key],
            environ={'REQUEST_METHOD': 'HEAD',
                     'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                         sig, expires)})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 404)
        self.assertEqual(req.environ['swift.authorize_override'], True)
        self.assertEqual(req.environ['REMOTE_USER'], '.wsgi.tempurl')

    def test_head_allowed_by_put(self):
        method = 'PUT'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(
            path, keys=[key],
            environ={'REQUEST_METHOD': 'HEAD',
                     'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                         sig, expires)})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 404)
        self.assertEqual(req.environ['swift.authorize_override'], True)
        self.assertEqual(req.environ['REMOTE_USER'], '.wsgi.tempurl')

    def test_head_allowed_by_post(self):
        method = 'POST'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(
            path, keys=[key],
            environ={'REQUEST_METHOD': 'HEAD',
                     'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                         sig, expires)})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 404)
        self.assertEqual(req.environ['swift.authorize_override'], True)
        self.assertEqual(req.environ['REMOTE_USER'], '.wsgi.tempurl')

    def test_head_otherwise_not_allowed(self):
        method = 'PUT'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        # Deliberately fudge expires to show HEADs aren't just automatically
        # allowed.
        expires += 1
        req = self._make_request(
            path, keys=[key],
            environ={'REQUEST_METHOD': 'HEAD',
                     'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                         sig, expires)})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 401)
        self.assertIn('Www-Authenticate', resp.headers)

    def test_post_when_forbidden_by_config(self):
        self.tempurl.conf['methods'].remove('POST')
        method = 'POST'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(
            path, keys=[key],
            environ={'REQUEST_METHOD': 'POST',
                     'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                         sig, expires)})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 401)
        self.assertIn(b'Temp URL invalid', resp.body)
        self.assertIn('Www-Authenticate', resp.headers)

    def test_delete_when_forbidden_by_config(self):
        self.tempurl.conf['methods'].remove('DELETE')
        method = 'DELETE'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(
            path, keys=[key],
            environ={'REQUEST_METHOD': 'DELETE',
                     'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                         sig, expires)})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 401)
        self.assertIn(b'Temp URL invalid', resp.body)
        self.assertIn('Www-Authenticate', resp.headers)

    def test_delete_allowed(self):
        method = 'DELETE'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(
            path, keys=[key],
            environ={'REQUEST_METHOD': 'DELETE',
                     'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                         sig, expires)})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 404)

    def test_unknown_not_allowed(self):
        method = 'UNKNOWN'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(
            path, keys=[key],
            environ={'REQUEST_METHOD': 'UNKNOWN',
                     'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                         sig, expires)})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 401)
        self.assertIn(b'Temp URL invalid', resp.body)
        self.assertIn('Www-Authenticate', resp.headers)

    def test_authorize_limits_scope(self):
        req_other_object = Request.blank("/v1/a/c/o2")
        req_other_container = Request.blank("/v1/a/c2/o2")
        req_other_account = Request.blank("/v1/a2/c2/o2")

        key_kwargs = {
            'keys': ['account-key', 'shared-key'],
            'container_keys': ['container-key', 'shared-key'],
        }

        # A request with the account key limits the pre-authed scope to the
        # account level.
        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'

        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(b'account-key', hmac_body, hashlib.sha1).hexdigest()
        qs = '?temp_url_sig=%s&temp_url_expires=%s' % (sig, expires)

        # make request will setup the environ cache for us
        req = self._make_request(path + qs, **key_kwargs)
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 404)  # sanity check

        authorize = req.environ['swift.authorize']
        # Requests for other objects happen if, for example, you're
        # downloading a large object or creating a large-object manifest.
        oo_resp = authorize(req_other_object)
        self.assertIsNone(oo_resp)
        oc_resp = authorize(req_other_container)
        self.assertIsNone(oc_resp)
        oa_resp = authorize(req_other_account)
        self.assertEqual(oa_resp.status_int, 401)

        # A request with the container key limits the pre-authed scope to
        # the container level; a different container in the same account is
        # out of scope and thus forbidden.
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(b'container-key', hmac_body, hashlib.sha1).hexdigest()
        qs = '?temp_url_sig=%s&temp_url_expires=%s' % (sig, expires)

        req = self._make_request(path + qs, **key_kwargs)
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 404)  # sanity check

        authorize = req.environ['swift.authorize']
        oo_resp = authorize(req_other_object)
        self.assertIsNone(oo_resp)
        oc_resp = authorize(req_other_container)
        self.assertEqual(oc_resp.status_int, 401)
        oa_resp = authorize(req_other_account)
        self.assertEqual(oa_resp.status_int, 401)

        # If account and container share a key (users set these, so this can
        # happen by accident, stupidity, *or* malice!), limit the scope to
        # account level. This prevents someone from shrinking the scope of
        # account-level tempurls by reusing one of the account's keys on a
        # container.
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(b'shared-key', hmac_body, hashlib.sha1).hexdigest()
        qs = '?temp_url_sig=%s&temp_url_expires=%s' % (sig, expires)

        req = self._make_request(path + qs, **key_kwargs)
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 404)  # sanity check

        authorize = req.environ['swift.authorize']
        oo_resp = authorize(req_other_object)
        self.assertIsNone(oo_resp)
        oc_resp = authorize(req_other_container)
        self.assertIsNone(oc_resp)
        oa_resp = authorize(req_other_account)
        self.assertEqual(oa_resp.status_int, 401)

    def test_changed_path_invalid(self):
        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        req = self._make_request(
            path + '2', keys=[key],
            environ={'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                sig, expires)})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 401)
        self.assertIn(b'Temp URL invalid', resp.body)
        self.assertIn('Www-Authenticate', resp.headers)

    def test_changed_sig_invalid(self):
        method = 'GET'
        expires = int(time() + 86400)
        path = '/v1/a/c/o'
        key = b'abc'
        hmac_body = ('%s\n%i\n%s' % (method, expires, path)).encode('utf-8')
        sig = hmac.new(key, hmac_body, hashlib.sha1).hexdigest()
        if sig[-1] != '0':
            sig = sig[:-1] + '0'
        else:
            sig = sig[:-1] + '1'
        req = self._make_request(
            path, keys=[key],
            environ={'QUERY_STRING': 'temp_url_sig=%s&temp_url_expires=%s' % (
                sig, expires)})
        resp = req.get_response(self.tempurl)
        self.assertEqual(resp.status_int, 401)
        self.assertIn(b'Temp URL invalid', resp.body)