# -*- coding: utf-8 -*-
# Copyright 2014 Google Inc. 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.
"""Tests for signurl command."""
from datetime import timedelta
import pkgutil
import gslib.commands.signurl
from gslib.commands.signurl import HAVE_OPENSSL
from gslib.exception import CommandException
import gslib.tests.testcase as testcase
from gslib.tests.testcase.integration_testcase import SkipForS3
from gslib.tests.util import ObjectToURI as suri
from gslib.tests.util import unittest
# pylint: disable=protected-access
@unittest.skipUnless(HAVE_OPENSSL, 'signurl requires pyopenssl.')
@SkipForS3('Signed URLs are only supported for gs:// URLs.')
class TestSignUrl(testcase.GsUtilIntegrationTestCase):
"""Integration tests for signurl command."""
def _GetKsFile(self):
if not hasattr(self, 'ks_file'):
# Dummy pkcs12 keystore generated with the command
# openssl req -new -passout pass:notasecret -batch \
# -x509 -keyout signed_url_test.key -out signed_url_test.pem \
# -subj '/CN=test.apps.googleusercontent.com'
# &&
# openssl pkcs12 -export -passin pass:notasecret \
# -passout pass:notasecret -inkey signed_url_test.key \
# -in signed_url_test.pem -out test.p12
# &&
# rm signed_url_test.key signed_url_test.pem
contents = pkgutil.get_data('gslib', 'tests/test_data/test.p12')
self.ks_file = self.CreateTempFile(contents=contents)
return self.ks_file
def testSignUrlOutput(self):
"""Tests signurl output of a sample object."""
object_url = self.CreateObject(contents='z')
stdout = self.RunGsUtil(['signurl', '-p', 'notasecret',
self._GetKsFile(), suri(object_url)],
return_stdout=True)
self.assertIn(object_url.uri, stdout)
self.assertIn('test@developer.gserviceaccount.com', stdout)
self.assertIn('Expires=', stdout)
self.assertIn('\tGET\t', stdout)
stdout = self.RunGsUtil(['signurl', '-m', 'PUT', '-p',
'notasecret', self._GetKsFile(),
'gs://test/test.txt'], return_stdout=True)
self.assertIn('test@developer.gserviceaccount.com', stdout)
self.assertIn('Expires=', stdout)
self.assertIn('\tPUT\t', stdout)
def testSignUrlWithURLEncodeRequiredChars(self):
objs = ['gs://example.org/test 1', 'gs://example.org/test/test 2',
'gs://example.org/Аудиоарi хив']
expected_partial_urls = [
('https://storage.googleapis.com/example.org/test%201?GoogleAccessId=te'
'st@developer.gserviceaccount.com'),
('https://storage.googleapis.com/example.org/test/test%202?GoogleAccess'
'Id=test@developer.gserviceaccount.com'),
('https://storage.googleapis.com/example.org/%D0%90%D1%83%D0%B4%D0%B8%D'
'0%BE%D0%B0%D1%80i%20%D1%85%D0%B8%D0%B2?GoogleAccessId=test@developer.'
'gserviceaccount.com')
]
self.assertEquals(len(objs), len(expected_partial_urls))
cmd_args = ['signurl', '-p', 'notasecret', self._GetKsFile()]
cmd_args.extend(objs)
stdout = self.RunGsUtil(cmd_args, return_stdout=True)
lines = stdout.split('\n')
# Header, signed urls, trailing newline.
self.assertEquals(len(lines), len(objs) + 2)
# Strip the header line to make the indices line up.
lines = lines[1:]
for obj, line, partial_url in zip(objs, lines, expected_partial_urls):
self.assertIn(obj, line)
self.assertIn(partial_url, line)
def testSignUrlWithWildcard(self):
objs = ['test1', 'test2', 'test3']
bucket = self.CreateBucket()
obj_urls = []
for obj_name in objs:
obj_urls.append(self.CreateObject(bucket_uri=bucket,
object_name=obj_name, contents=''))
stdout = self.RunGsUtil(['signurl', '-p',
'notasecret', self._GetKsFile(),
suri(bucket) + '/*'], return_stdout=True)
# Header, 3 signed urls, trailing newline
self.assertEquals(len(stdout.split('\n')), 5)
for obj_url in obj_urls:
self.assertIn(suri(obj_url), stdout)
def testSignUrlOfNonObjectUrl(self):
"""Tests the signurl output of a non-existent file."""
self.RunGsUtil(['signurl', self._GetKsFile(), 'gs://'],
expected_status=1, stdin='notasecret')
self.RunGsUtil(['signurl', 'file://tmp/abc'], expected_status=1)
@unittest.skipUnless(HAVE_OPENSSL, 'signurl requires pyopenssl.')
class UnitTestSignUrl(testcase.GsUtilUnitTestCase):
"""Unit tests for the signurl command."""
def setUp(self):
super(UnitTestSignUrl, self).setUp()
self.ks_contents = pkgutil.get_data('gslib', 'tests/test_data/test.p12')
def testDurationSpec(self):
tests = [('1h', timedelta(hours=1)),
('2d', timedelta(days=2)),
('5D', timedelta(days=5)),
('35s', timedelta(seconds=35)),
('1h', timedelta(hours=1)),
('33', timedelta(hours=33)),
('22m', timedelta(minutes=22)),
('3.7', None),
('27Z', None),
]
for inp, expected in tests:
try:
td = gslib.commands.signurl._DurationToTimeDelta(inp)
self.assertEquals(td, expected)
except CommandException:
if expected is not None:
self.fail('{0} failed to parse')
def testSignPut(self):
"""Tests the return value of the _GenSignedUrl function with \
a PUT method."""
expected = ('https://storage.googleapis.com/test/test.txt?'
'GoogleAccessId=test@developer.gserviceaccount.com'
'&Expires=1391816302&Signature=A6QbgTA8cXZCtjy2xCr401bdi0e'
'7zChTBQ6BX61L7AfytTGEQDMD%2BbvOQKjX7%2FsEh77cmzcSxOEKqTLUD'
'bbkPgPqW3j8sGPSRX9VM58bgj1vt9yU8cRKoegFHXAqsATx2G5rc%2FvEl'
'iFp9UWMfVj5TaukqlBAVuzZWlyx0aQa9tCKXRtC9YcxORxG41RfiowA2kd8'
'XBTQt4M9XTzpVyr5rVMzfr2LvtGf9UAJvlt8p6T6nThl2vy9%2FwBoPcMFa'
'OWQcGTagwjyKWDcI1vQPIFQLGftAcv3QnGZxZTtg8pZW%2FIxRJrBhfFfcA'
'c62hDKyaU2YssSMy%2FjUJynWx3TIiJjhg%3D%3D')
expiration = 1391816302
ks, client_id = (gslib.commands.signurl
._ReadKeystore(self.ks_contents, 'notasecret'))
signed_url = (gslib.commands.signurl
._GenSignedUrl(ks.get_privatekey(),
client_id, 'PUT', '',
'', expiration, 'test/test.txt'))
self.assertEquals(expected, signed_url)
def testSignurlPutContentype(self):
"""Tests the return value of the _GenSignedUrl function with \
a PUT method and specified content type."""
expected = ('https://storage.googleapis.com/test/test.txt?'
'GoogleAccessId=test@developer.gserviceaccount.com&'
'Expires=1391816302&Signature=APn%2BCCVcQrfc1fKQXrs'
'PEZFj9%2FmASO%2BolR8xwgBY6PbWMkcCtrUVFBauP6t4NxqZO'
'UnbOFYTZYzul0RC57ZkEWJp3VcyDIHcn6usEE%2FTzUHhbDCDW'
'awAkZS7p8kO8IIACuJlF5s9xZmZzaEBtzF0%2BBOsGgBPBlg2y'
'zrhFB6cyyAwNiUgmhLQaVkdobnSwtI5QJkvXoIjJb6hhLiVbLC'
'rWdgSZVusjAKGlWCJsM%2B4TkCR%2Bi8AnrkECngcMHuJ9mYbS'
'XI1VfEmcnRVcfkKkJGZGctaDIWK%2FMTEmfYCW6USt3Zk2WowJ'
'SGuJHqEcFz0kyfAlkpmG%2Fl5E1FQROYqLN2kZQ%3D%3D')
expiration = 1391816302
ks, client_id = (gslib.commands.signurl
._ReadKeystore(self.ks_contents,
'notasecret'))
signed_url = (gslib.commands.signurl
._GenSignedUrl(ks.get_privatekey(),
client_id, 'PUT', '',
'text/plain', expiration,
'test/test.txt'))
self.assertEquals(expected, signed_url)
def testSignurlGet(self):
"""Tests the return value of the _GenSignedUrl function with \
a GET method."""
expected = ('https://storage.googleapis.com/test/test.txt?'
'GoogleAccessId=test@developer.gserviceaccount.com&'
'Expires=0&Signature=TCZwe32cU%2BMksmLiSY9shHXQjLs1'
'F3y%2F%2F1M0UhiK4qsPRVNZVwI7YWvv2qa2Xa%2BVBBafboF0'
'1%2BWvx3ZG316pwpNIRR6y7jNnE0LvQmHE8afbm2VYCi%2B2JS'
'ZK2YZFJAyEek8si53jhYQEmaRq1zPfGbX84B2FJ8v4iI%2FTC1'
'I9OE5vHF0sWwIR9d73JDrFLjaync7QYFWRExdwvqlQX%2BPO3r'
'OG9Ns%2BcQFIN7npnsVjH28yNY9gBzXya8LYmNvUx6bWHWZMiu'
'fLwDZ0jejNeDZTOfQGRM%2B0vY7NslzaT06W1wo8P7McSkAZEl'
'DCbhR0Vo1fturPMwmAhi88f0qzRzywbg%3D%3D')
expiration = 0
ks, client_id = (gslib.commands.signurl
._ReadKeystore(self.ks_contents,
'notasecret'))
signed_url = (gslib.commands.signurl
._GenSignedUrl(ks.get_privatekey(),
client_id, 'GET', '',
'', expiration, 'test/test.txt'))
self.assertEquals(expected, signed_url)