diff --git a/gcloud/credentials.py b/gcloud/credentials.py
index 9e96f5317285..218ddeee9563 100644
--- a/gcloud/credentials.py
+++ b/gcloud/credentials.py
@@ -1,6 +1,24 @@
 """A simple wrapper around the OAuth2 credentials library."""
 
+import base64
+import calendar
+import datetime
+import urllib
+
+from Crypto.Hash import SHA256
+from Crypto.PublicKey import RSA
+from Crypto.Signature import PKCS1_v1_5
 from oauth2client import client
+from OpenSSL import crypto
+import pytz
+
+
+def _utcnow():  # pragma: NO COVER testing replaces
+    """Returns current time as UTC datetime.
+
+    NOTE: on the module namespace so tests can replace it.
+    """
+    return datetime.datetime.utcnow()
 
 
 def get_for_service_account(client_email, private_key_path, scope=None):
@@ -36,3 +54,105 @@ def get_for_service_account(client_email, private_key_path, scope=None):
         service_account_name=client_email,
         private_key=open(private_key_path).read(),
         scope=scope)
+
+
+def generate_signed_url(credentials, endpoint, resource, expiration,
+                        method='GET', content_md5=None, content_type=None):
+    """Generate signed URL to provide query-string auth'n to a resource.
+
+    :type credentials:
+        :class:`oauth2client.client.SignedJwtAssertionCredentials`
+    :param credentials: the credentials used to sign the URL.
+
+    :type endpoint: string
+    :param endpoint: Base API endpoint URL.
+
+    :type resource: string
+    :param resource: A pointer to a specific resource within the endpoint
+                     (e.g., ``/bucket-name/file.txt``).
+
+    :type expiration: int, long, datetime.datetime, datetime.timedelta
+    :param expiration: When the signed URL should expire.
+
+    :type method: string
+    :param method: The HTTP verb that will be used when requesting the URL.
+
+    :type content_md5: string
+    :param content_md5: The MD5 hash of the object referenced by
+                        ``resource``.
+
+    :type content_type: string
+    :param content_type: The content type of the object referenced by
+                            ``resource``.
+
+    :rtype: string
+    :returns: A signed URL you can use to access the resource
+                until expiration.
+    """
+    expiration = _get_expiration_seconds(expiration)
+
+    # Generate the string to sign.
+    signature_string = '\n'.join([
+        method,
+        content_md5 or '',
+        content_type or '',
+        str(expiration),
+        resource])
+
+    # Take our PKCS12 (.p12) key and make it into a RSA key we can use...
+    pkcs12 = crypto.load_pkcs12(
+        base64.b64decode(credentials.private_key),
+        'notasecret')
+    pem = crypto.dump_privatekey(
+        crypto.FILETYPE_PEM, pkcs12.get_privatekey())
+    pem_key = RSA.importKey(pem)
+
+    # Sign the string with the RSA key.
+    signer = PKCS1_v1_5.new(pem_key)
+    signature_hash = SHA256.new(signature_string)
+    signature_bytes = signer.sign(signature_hash)
+    signature = base64.b64encode(signature_bytes)
+
+    # Set the right query parameters.
+    query_params = {
+        'GoogleAccessId': credentials.service_account_name,
+        'Expires': str(expiration),
+        'Signature': signature,
+    }
+
+    # Return the built URL.
+    return '{endpoint}{resource}?{querystring}'.format(
+        endpoint=endpoint, resource=resource,
+        querystring=urllib.urlencode(query_params))
+
+
+def _get_expiration_seconds(expiration):
+    """Convert 'expiration' to a number of seconds in the future.
+
+    :type expiration: int, long, datetime.datetime, datetime.timedelta
+    :param expiration: When the signed URL should expire.
+
+    :rtype: int
+    :returns: a timestamp as an absolute number of seconds.
+    """
+    # If it's a timedelta, add it to `now` in UTC.
+    if isinstance(expiration, datetime.timedelta):
+        now = _utcnow().replace(tzinfo=pytz.utc)
+        expiration = now + expiration
+
+    # If it's a datetime, convert to a timestamp.
+    if isinstance(expiration, datetime.datetime):
+        # Make sure the timezone on the value is UTC
+        # (either by converting or replacing the value).
+        if expiration.tzinfo:
+            expiration = expiration.astimezone(pytz.utc)
+        else:
+            expiration = expiration.replace(tzinfo=pytz.utc)
+
+        # Turn the datetime into a timestamp (seconds, not microseconds).
+        expiration = int(calendar.timegm(expiration.timetuple()))
+
+    if not isinstance(expiration, (int, long)):
+        raise TypeError('Expected an integer timestamp, datetime, or '
+                        'timedelta. Got %s' % type(expiration))
+    return expiration
diff --git a/gcloud/storage/connection.py b/gcloud/storage/connection.py
index c1d62c194338..78c796987545 100644
--- a/gcloud/storage/connection.py
+++ b/gcloud/storage/connection.py
@@ -1,31 +1,16 @@
 """Create / interact with gcloud storage connections."""
 
-import base64
-import calendar
-import datetime
 import json
 import urllib
 
-from Crypto.Hash import SHA256
-from Crypto.PublicKey import RSA
-from Crypto.Signature import PKCS1_v1_5
-from OpenSSL import crypto
-import pytz
 
 from gcloud import connection
+from gcloud import credentials
 from gcloud.storage import exceptions
 from gcloud.storage.bucket import Bucket
 from gcloud.storage.bucket import BucketIterator
 
 
-def _utcnow():  # pragma: NO COVER testing replaces
-    """Returns current time as UTC datetime.
-
-    NOTE: on the module namespace so tests can replace it.
-    """
-    return datetime.datetime.utcnow()
-
-
 class Connection(connection.Connection):
     """A connection to Google Cloud Storage via the JSON REST API.
 
@@ -453,70 +438,6 @@ def generate_signed_url(self, resource, expiration,
         :returns: A signed URL you can use to access the resource
                   until expiration.
         """
-        expiration = _get_expiration_seconds(expiration)
-
-        # Generate the string to sign.
-        signature_string = '\n'.join([
-            method,
-            content_md5 or '',
-            content_type or '',
-            str(expiration),
-            resource])
-
-        # Take our PKCS12 (.p12) key and make it into a RSA key we can use...
-        pkcs12 = crypto.load_pkcs12(
-            base64.b64decode(self.credentials.private_key),
-            'notasecret')
-        pem = crypto.dump_privatekey(
-            crypto.FILETYPE_PEM, pkcs12.get_privatekey())
-        pem_key = RSA.importKey(pem)
-
-        # Sign the string with the RSA key.
-        signer = PKCS1_v1_5.new(pem_key)
-        signature_hash = SHA256.new(signature_string)
-        signature_bytes = signer.sign(signature_hash)
-        signature = base64.b64encode(signature_bytes)
-
-        # Set the right query parameters.
-        query_params = {
-            'GoogleAccessId': self.credentials.service_account_name,
-            'Expires': str(expiration),
-            'Signature': signature,
-        }
-
-        # Return the built URL.
-        return '{endpoint}{resource}?{querystring}'.format(
-            endpoint=self.API_ACCESS_ENDPOINT, resource=resource,
-            querystring=urllib.urlencode(query_params))
-
-
-def _get_expiration_seconds(expiration):
-    """Convert 'expiration' to a number of seconds in the future.
-
-    :type expiration: int, long, datetime.datetime, datetime.timedelta
-    :param expiration: When the signed URL should expire.
-
-    :rtype: int
-    :returns: a timestamp as an absolute number of seconds.
-    """
-    # If it's a timedelta, add it to `now` in UTC.
-    if isinstance(expiration, datetime.timedelta):
-        now = _utcnow().replace(tzinfo=pytz.utc)
-        expiration = now + expiration
-
-    # If it's a datetime, convert to a timestamp.
-    if isinstance(expiration, datetime.datetime):
-        # Make sure the timezone on the value is UTC
-        # (either by converting or replacing the value).
-        if expiration.tzinfo:
-            expiration = expiration.astimezone(pytz.utc)
-        else:
-            expiration = expiration.replace(tzinfo=pytz.utc)
-
-        # Turn the datetime into a timestamp (seconds, not microseconds).
-        expiration = int(calendar.timegm(expiration.timetuple()))
-
-    if not isinstance(expiration, (int, long)):
-        raise TypeError('Expected an integer timestamp, datetime, or '
-                        'timedelta. Got %s' % type(expiration))
-    return expiration
+        return credentials.generate_signed_url(
+            self.credentials, self.API_ACCESS_ENDPOINT, resource, expiration,
+            method, content_md5, content_type)
diff --git a/gcloud/storage/test_connection.py b/gcloud/storage/test_connection.py
index 036d8f36da00..60f9fd14d334 100644
--- a/gcloud/storage/test_connection.py
+++ b/gcloud/storage/test_connection.py
@@ -577,118 +577,35 @@ def test_new_bucket_w_invalid(self):
         self.assertRaises(TypeError, conn.new_bucket, object())
 
     def test_generate_signed_url_w_expiration_int(self):
-        import base64
-        import urlparse
         from gcloud._testing import _Monkey
-        from gcloud.storage import connection as MUT
+        from gcloud import credentials
 
         ENDPOINT = 'http://api.example.com'
         RESOURCE = '/name/key'
         PROJECT = 'project'
-        SIGNED = base64.b64encode('DEADBEEF')
-        crypto = _Crypto()
-        rsa = _RSA()
-        pkcs_v1_5 = _PKCS1_v1_5()
-        sha256 = _SHA256()
-        conn = self._makeOne(PROJECT, _Credentials())
+        creds = object()
+        conn = self._makeOne(PROJECT, creds)
         conn.API_ACCESS_ENDPOINT = ENDPOINT
 
-        with _Monkey(MUT, crypto=crypto, RSA=rsa, PKCS1_v1_5=pkcs_v1_5,
-                     SHA256=sha256):
-            url = conn.generate_signed_url(RESOURCE, 1000)
-
-        scheme, netloc, path, qs, frag = urlparse.urlsplit(url)
-        self.assertEqual(scheme, 'http')
-        self.assertEqual(netloc, 'api.example.com')
-        self.assertEqual(path, RESOURCE)
-        params = urlparse.parse_qs(qs)
-        self.assertEqual(len(params), 3)
-        self.assertEqual(params['Signature'], [SIGNED])
-        self.assertEqual(params['Expires'], ['1000'])
-        self.assertEqual(params['GoogleAccessId'],
-                         [_Credentials.service_account_name])
-        self.assertEqual(frag, '')
-
-
-class Test__get_expiration_seconds(unittest2.TestCase):
-
-    def _callFUT(self, expiration):
-        from gcloud.storage.connection import _get_expiration_seconds
-
-        return _get_expiration_seconds(expiration)
-
-    def _utc_seconds(self, when):
-        import calendar
-
-        return int(calendar.timegm(when.timetuple()))
-
-    def test__get_expiration_seconds_w_invalid(self):
-        self.assertRaises(TypeError, self._callFUT, object())
-        self.assertRaises(TypeError, self._callFUT, None)
-
-    def test__get_expiration_seconds_w_int(self):
-        self.assertEqual(self._callFUT(123), 123)
-
-    def test__get_expiration_seconds_w_long(self):
-        try:
-            long
-        except NameError:  # pragma: NO COVER Py3K
-            pass
-        else:
-            self.assertEqual(self._callFUT(long(123)), 123)
+        _called_with = []
 
-    def test__get_expiration_w_naive_datetime(self):
-        import datetime
-
-        expiration_no_tz = datetime.datetime(2004, 8, 19, 0, 0, 0, 0)
-        utc_seconds = self._utc_seconds(expiration_no_tz)
-        self.assertEqual(self._callFUT(expiration_no_tz), utc_seconds)
-
-    def test__get_expiration_w_utc_datetime(self):
-        import datetime
-        import pytz
-
-        expiration_utc = datetime.datetime(2004, 8, 19, 0, 0, 0, 0, pytz.utc)
-        utc_seconds = self._utc_seconds(expiration_utc)
-        self.assertEqual(self._callFUT(expiration_utc), utc_seconds)
-
-    def test__get_expiration_w_other_zone_datetime(self):
-        import datetime
-        import pytz
-
-        zone = pytz.timezone('CET')
-        expiration_other = datetime.datetime(2004, 8, 19, 0, 0, 0, 0, zone)
-        utc_seconds = self._utc_seconds(expiration_other)
-        cet_seconds = utc_seconds - (60 * 60)  # CET one hour earlier than UTC
-        self.assertEqual(self._callFUT(expiration_other), cet_seconds)
-
-    def test__get_expiration_seconds_w_timedelta_seconds(self):
-        import datetime
-        from gcloud.storage import connection
-        from gcloud._testing import _Monkey
-
-        dummy_utcnow = datetime.datetime(2004, 8, 19, 0, 0, 0, 0)
-        utc_seconds = self._utc_seconds(dummy_utcnow)
-        expiration_as_delta = datetime.timedelta(seconds=10)
-
-        with _Monkey(connection, _utcnow=lambda: dummy_utcnow):
-            result = self._callFUT(expiration_as_delta)
-
-        self.assertEqual(result, utc_seconds + 10)
-
-    def test__get_expiration_seconds_w_timedelta_days(self):
-        import datetime
-        from gcloud.storage import connection
-        from gcloud._testing import _Monkey
+        def _generate(*args, **kw):
+            _called_with.append((args, kw))
 
-        dummy_utcnow = datetime.datetime(2004, 8, 19, 0, 0, 0, 0)
-        utc_seconds = self._utc_seconds(dummy_utcnow)
-        expiration_as_delta = datetime.timedelta(days=1)
+        with _Monkey(credentials, generate_signed_url=_generate):
+            _ = conn.generate_signed_url(RESOURCE, 1000)
 
-        with _Monkey(connection, _utcnow=lambda: dummy_utcnow):
-            result = self._callFUT(expiration_as_delta)
+        self.assertEqual(len(_called_with), 1)
+        args, kw = _called_with[0]
 
-        self.assertEqual(result, utc_seconds + 86400)
+        self.assertTrue(args[0] is creds)
+        self.assertEqual(args[1], ENDPOINT)
+        self.assertEqual(args[2], RESOURCE)
+        self.assertEqual(args[3], 1000)
+        self.assertEqual(args[4], 'GET')
+        self.assertEqual(args[5], None)
+        self.assertEqual(args[6], None)
+        self.assertEqual(kw, {})
 
 
 class Http(object):
@@ -703,61 +620,3 @@ def __init__(self, headers, content):
     def request(self, **kw):
         self._called_with = kw
         return self._response, self._content
-
-
-class _Credentials(object):
-
-    service_account_name = 'testing@example.com'
-
-    @property
-    def private_key(self):
-        import base64
-        return base64.b64encode('SEEKRIT')
-
-
-class _Crypto(object):
-
-    FILETYPE_PEM = 'pem'
-    _loaded = _dumped = None
-
-    def load_pkcs12(self, buffer, passphrase):
-        self._loaded = (buffer, passphrase)
-        return self
-
-    def get_privatekey(self):
-        return '__PKCS12__'
-
-    def dump_privatekey(self, type, pkey, cipher=None, passphrase=None):
-        self._dumped = (type, pkey, cipher, passphrase)
-        return '__PEM__'
-
-
-class _RSA(object):
-
-    _imported = None
-
-    def importKey(self, pem):
-        self._imported = pem
-        return 'imported:%s' % pem
-
-
-class _PKCS1_v1_5(object):
-
-    _pem_key = _signature_hash = None
-
-    def new(self, pem_key):
-        self._pem_key = pem_key
-        return self
-
-    def sign(self, signature_hash):
-        self._signature_hash = signature_hash
-        return 'DEADBEEF'
-
-
-class _SHA256(object):
-
-    _signature_string = None
-
-    def new(self, signature_string):
-        self._signature_string = signature_string
-        return self
diff --git a/gcloud/test_credentials.py b/gcloud/test_credentials.py
index 15c8b8b4a0b3..1b1c92200330 100644
--- a/gcloud/test_credentials.py
+++ b/gcloud/test_credentials.py
@@ -1,7 +1,7 @@
 import unittest2
 
 
-class TestCredentials(unittest2.TestCase):
+class Test_get_for_service_account(unittest2.TestCase):
 
     def test_get_for_service_account_wo_scope(self):
         from tempfile import NamedTemporaryFile
@@ -47,6 +47,126 @@ def test_get_for_service_account_w_scope(self):
         self.assertEqual(client._called_with, expected_called_with)
 
 
+class Test_generate_signed_url(unittest2.TestCase):
+
+    def _callFUT(self, *args, **kw):
+        from gcloud.credentials import generate_signed_url
+
+        return generate_signed_url(*args, **kw)
+
+    def test__w_expiration_int(self):
+        import base64
+        import urlparse
+        from gcloud._testing import _Monkey
+        from gcloud import credentials
+
+        ENDPOINT = 'http://api.example.com'
+        RESOURCE = '/name/key'
+        SIGNED = base64.b64encode('DEADBEEF')
+        crypto = _Crypto()
+        rsa = _RSA()
+        pkcs_v1_5 = _PKCS1_v1_5()
+        sha256 = _SHA256()
+        creds = _Credentials()
+
+        with _Monkey(credentials, crypto=crypto, RSA=rsa, PKCS1_v1_5=pkcs_v1_5,
+                     SHA256=sha256):
+            url = self._callFUT(creds, ENDPOINT, RESOURCE, 1000)
+
+        scheme, netloc, path, qs, frag = urlparse.urlsplit(url)
+        self.assertEqual(scheme, 'http')
+        self.assertEqual(netloc, 'api.example.com')
+        self.assertEqual(path, RESOURCE)
+        params = urlparse.parse_qs(qs)
+        self.assertEqual(len(params), 3)
+        self.assertEqual(params['Signature'], [SIGNED])
+        self.assertEqual(params['Expires'], ['1000'])
+        self.assertEqual(params['GoogleAccessId'],
+                         [_Credentials.service_account_name])
+        self.assertEqual(frag, '')
+
+
+class Test__get_expiration_seconds(unittest2.TestCase):
+
+    def _callFUT(self, expiration):
+        from gcloud.credentials import _get_expiration_seconds
+
+        return _get_expiration_seconds(expiration)
+
+    def _utc_seconds(self, when):
+        import calendar
+
+        return int(calendar.timegm(when.timetuple()))
+
+    def test__get_expiration_seconds_w_invalid(self):
+        self.assertRaises(TypeError, self._callFUT, object())
+        self.assertRaises(TypeError, self._callFUT, None)
+
+    def test__get_expiration_seconds_w_int(self):
+        self.assertEqual(self._callFUT(123), 123)
+
+    def test__get_expiration_seconds_w_long(self):
+        try:
+            long
+        except NameError:  # pragma: NO COVER Py3K
+            pass
+        else:
+            self.assertEqual(self._callFUT(long(123)), 123)
+
+    def test__get_expiration_w_naive_datetime(self):
+        import datetime
+
+        expiration_no_tz = datetime.datetime(2004, 8, 19, 0, 0, 0, 0)
+        utc_seconds = self._utc_seconds(expiration_no_tz)
+        self.assertEqual(self._callFUT(expiration_no_tz), utc_seconds)
+
+    def test__get_expiration_w_utc_datetime(self):
+        import datetime
+        import pytz
+
+        expiration_utc = datetime.datetime(2004, 8, 19, 0, 0, 0, 0, pytz.utc)
+        utc_seconds = self._utc_seconds(expiration_utc)
+        self.assertEqual(self._callFUT(expiration_utc), utc_seconds)
+
+    def test__get_expiration_w_other_zone_datetime(self):
+        import datetime
+        import pytz
+
+        zone = pytz.timezone('CET')
+        expiration_other = datetime.datetime(2004, 8, 19, 0, 0, 0, 0, zone)
+        utc_seconds = self._utc_seconds(expiration_other)
+        cet_seconds = utc_seconds - (60 * 60)  # CET one hour earlier than UTC
+        self.assertEqual(self._callFUT(expiration_other), cet_seconds)
+
+    def test__get_expiration_seconds_w_timedelta_seconds(self):
+        import datetime
+        from gcloud import credentials
+        from gcloud._testing import _Monkey
+
+        dummy_utcnow = datetime.datetime(2004, 8, 19, 0, 0, 0, 0)
+        utc_seconds = self._utc_seconds(dummy_utcnow)
+        expiration_as_delta = datetime.timedelta(seconds=10)
+
+        with _Monkey(credentials, _utcnow=lambda: dummy_utcnow):
+            result = self._callFUT(expiration_as_delta)
+
+        self.assertEqual(result, utc_seconds + 10)
+
+    def test__get_expiration_seconds_w_timedelta_days(self):
+        import datetime
+        from gcloud import credentials
+        from gcloud._testing import _Monkey
+
+        dummy_utcnow = datetime.datetime(2004, 8, 19, 0, 0, 0, 0)
+        utc_seconds = self._utc_seconds(dummy_utcnow)
+        expiration_as_delta = datetime.timedelta(days=1)
+
+        with _Monkey(credentials, _utcnow=lambda: dummy_utcnow):
+            result = self._callFUT(expiration_as_delta)
+
+        self.assertEqual(result, utc_seconds + 86400)
+
+
 class _Client(object):
     def __init__(self):
         self._signed = object()
@@ -54,3 +174,61 @@ def __init__(self):
     def SignedJwtAssertionCredentials(self, **kw):
         self._called_with = kw
         return self._signed
+
+
+class _Credentials(object):
+
+    service_account_name = 'testing@example.com'
+
+    @property
+    def private_key(self):
+        import base64
+        return base64.b64encode('SEEKRIT')
+
+
+class _Crypto(object):
+
+    FILETYPE_PEM = 'pem'
+    _loaded = _dumped = None
+
+    def load_pkcs12(self, buffer, passphrase):
+        self._loaded = (buffer, passphrase)
+        return self
+
+    def get_privatekey(self):
+        return '__PKCS12__'
+
+    def dump_privatekey(self, type, pkey, cipher=None, passphrase=None):
+        self._dumped = (type, pkey, cipher, passphrase)
+        return '__PEM__'
+
+
+class _RSA(object):
+
+    _imported = None
+
+    def importKey(self, pem):
+        self._imported = pem
+        return 'imported:%s' % pem
+
+
+class _PKCS1_v1_5(object):
+
+    _pem_key = _signature_hash = None
+
+    def new(self, pem_key):
+        self._pem_key = pem_key
+        return self
+
+    def sign(self, signature_hash):
+        self._signature_hash = signature_hash
+        return 'DEADBEEF'
+
+
+class _SHA256(object):
+
+    _signature_string = None
+
+    def new(self, signature_string):
+        self._signature_string = signature_string
+        return self
diff --git a/pylintrc_default b/pylintrc_default
index 1caa15be925c..82382304c45e 100644
--- a/pylintrc_default
+++ b/pylintrc_default
@@ -24,7 +24,7 @@ ignore = datastore_v1_pb2.py
 disable = I, protected-access, maybe-no-member, no-member,
           redefined-builtin, star-args, missing-format-attribute,
           similarities, arguments-differ,
-          too-many-public-methods, too-few-public-methods
+          too-many-public-methods, too-few-public-methods, too-many-locals
 
 [REPORTS]
 reports = no