diff --git a/docs/index.rst b/docs/index.rst
index dd0e6e90bf70..7f30287711cc 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -40,6 +40,7 @@
   pubsub-topic
   pubsub-subscription
   pubsub-message
+  pubsub-iam
 
 .. toctree::
   :maxdepth: 0
diff --git a/docs/pubsub-iam.rst b/docs/pubsub-iam.rst
new file mode 100644
index 000000000000..701dad385ece
--- /dev/null
+++ b/docs/pubsub-iam.rst
@@ -0,0 +1,8 @@
+IAM Policy
+~~~~~~~~~~
+
+.. automodule:: gcloud.pubsub.iam
+  :members:
+  :undoc-members:
+  :show-inheritance:
+
diff --git a/docs/pubsub-usage.rst b/docs/pubsub-usage.rst
index 99d01d2cde50..2e3678fb846e 100644
--- a/docs/pubsub-usage.rst
+++ b/docs/pubsub-usage.rst
@@ -66,6 +66,23 @@ Delete a topic:
    >>> topic = client.topic('topic_name')
    >>> topic.delete()  # API request
 
+Fetch the IAM policy for a topic:
+
+.. doctest::
+
+   >>> from gcloud import pubsub
+   >>> client = pubsub.Client()
+   >>> topic = client.topic('topic_name')
+   >>> policy = topic.get_iam_policy()  # API request
+   >>> policy.etag
+   'DEADBEEF'
+   >>> policy.owners
+   ['user:phred@example.com']
+   >>> policy.writers
+   ['systemAccount:abc-1234@systemaccounts.example.com']
+   >>> policy.readers
+   ['domain:example.com']
+
 
 Publish messages to a topic
 ---------------------------
diff --git a/gcloud/pubsub/iam.py b/gcloud/pubsub/iam.py
new file mode 100644
index 000000000000..0b85b6b5f96c
--- /dev/null
+++ b/gcloud/pubsub/iam.py
@@ -0,0 +1,164 @@
+# Copyright 2016 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.
+"""PubSub API IAM policy definitions"""
+
+_OWNER_ROLE = 'roles/owner'
+_WRITER_ROLE = 'roles/writer'
+_READER_ROLE = 'roles/reader'
+
+
+class Policy(object):
+    """Combined IAM Policy / Bindings.
+
+    See:
+    https://cloud.google.com/pubsub/reference/rest/Shared.Types/Policy
+    https://cloud.google.com/pubsub/reference/rest/Shared.Types/Binding
+
+    :type etag: string
+    :param etag: ETag used to identify a unique of the policy
+
+    :type version: int
+    :param version: unique version of the policy
+    """
+    def __init__(self, etag=None, version=None):
+        self.etag = etag
+        self.version = version
+        self.owners = set()
+        self.writers = set()
+        self.readers = set()
+
+    @staticmethod
+    def user(email):
+        """Factory method for a user member.
+
+        :type email: string
+        :param email: E-mail for this particular user.
+
+        :rtype: string
+        :returns: A member string corresponding to the given user.
+        """
+        return 'user:%s' % (email,)
+
+    @staticmethod
+    def service_account(email):
+        """Factory method for a service account member.
+
+        :type email: string
+        :param email: E-mail for this particular service account.
+
+        :rtype: string
+        :returns: A member string corresponding to the given service account.
+        """
+        return 'serviceAccount:%s' % (email,)
+
+    @staticmethod
+    def group(email):
+        """Factory method for a group member.
+
+        :type email: string
+        :param email: An id or e-mail for this particular group.
+
+        :rtype: string
+        :returns: A member string corresponding to the given group.
+        """
+        return 'group:%s' % (email,)
+
+    @staticmethod
+    def domain(domain):
+        """Factory method for a domain member.
+
+        :type domain: string
+        :param domain: The domain for this member.
+
+        :rtype: string
+        :returns: A member string corresponding to the given domain.
+        """
+        return 'domain:%s' % (domain,)
+
+    @staticmethod
+    def all_users():
+        """Factory method for a member representing all users.
+
+        :rtype: string
+        :returns: A member string representing all users.
+        """
+        return 'allUsers'
+
+    @staticmethod
+    def authenticated_users():
+        """Factory method for a member representing all authenticated users.
+
+        :rtype: string
+        :returns: A member string representing all authenticated users.
+        """
+        return 'allAuthenticatedUsers'
+
+    @classmethod
+    def from_api_repr(cls, resource):
+        """Create a policy from the resource returned from the API.
+
+        :type resource: dict
+        :param resource: resource returned from the ``getIamPolicy`` API.
+
+        :rtype: :class:`Policy`
+        :returns: the parsed policy
+        """
+        version = resource.get('version')
+        etag = resource.get('etag')
+        policy = cls(etag, version)
+        for binding in resource.get('bindings', ()):
+            role = binding['role']
+            members = set(binding['members'])
+            if role == _OWNER_ROLE:
+                policy.owners = members
+            elif role == _WRITER_ROLE:
+                policy.writers = members
+            elif role == _READER_ROLE:
+                policy.readers = members
+            else:
+                raise ValueError('Unknown role: %s' % (role,))
+        return policy
+
+    def to_api_repr(self):
+        """Construct a Policy resource.
+
+        :rtype: dict
+        :returns: a resource to be passed to the ``setIamPolicy`` API.
+        """
+        resource = {}
+
+        if self.etag is not None:
+            resource['etag'] = self.etag
+
+        if self.version is not None:
+            resource['version'] = self.version
+
+        bindings = []
+
+        if self.owners:
+            bindings.append(
+                {'role': _OWNER_ROLE, 'members': sorted(self.owners)})
+
+        if self.writers:
+            bindings.append(
+                {'role': _WRITER_ROLE, 'members': sorted(self.writers)})
+
+        if self.readers:
+            bindings.append(
+                {'role': _READER_ROLE, 'members': sorted(self.readers)})
+
+        if bindings:
+            resource['bindings'] = bindings
+
+        return resource
diff --git a/gcloud/pubsub/test_iam.py b/gcloud/pubsub/test_iam.py
new file mode 100644
index 000000000000..3efd6df5c49e
--- /dev/null
+++ b/gcloud/pubsub/test_iam.py
@@ -0,0 +1,160 @@
+# Copyright 2016 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.
+
+import unittest2
+
+
+class TestPolicy(unittest2.TestCase):
+
+    def _getTargetClass(self):
+        from gcloud.pubsub.iam import Policy
+        return Policy
+
+    def _makeOne(self, *args, **kw):
+        return self._getTargetClass()(*args, **kw)
+
+    def test_ctor_defaults(self):
+        policy = self._makeOne()
+        self.assertEqual(policy.etag, None)
+        self.assertEqual(policy.version, None)
+        self.assertEqual(list(policy.owners), [])
+        self.assertEqual(list(policy.writers), [])
+        self.assertEqual(list(policy.readers), [])
+
+    def test_ctor_explicit(self):
+        VERSION = 17
+        ETAG = 'ETAG'
+        policy = self._makeOne(ETAG, VERSION)
+        self.assertEqual(policy.etag, ETAG)
+        self.assertEqual(policy.version, VERSION)
+        self.assertEqual(list(policy.owners), [])
+        self.assertEqual(list(policy.writers), [])
+        self.assertEqual(list(policy.readers), [])
+
+    def test_user(self):
+        EMAIL = 'phred@example.com'
+        MEMBER = 'user:%s' % (EMAIL,)
+        policy = self._makeOne()
+        self.assertEqual(policy.user(EMAIL), MEMBER)
+
+    def test_service_account(self):
+        EMAIL = 'phred@example.com'
+        MEMBER = 'serviceAccount:%s' % (EMAIL,)
+        policy = self._makeOne()
+        self.assertEqual(policy.service_account(EMAIL), MEMBER)
+
+    def test_group(self):
+        EMAIL = 'phred@example.com'
+        MEMBER = 'group:%s' % (EMAIL,)
+        policy = self._makeOne()
+        self.assertEqual(policy.group(EMAIL), MEMBER)
+
+    def test_domain(self):
+        DOMAIN = 'example.com'
+        MEMBER = 'domain:%s' % (DOMAIN,)
+        policy = self._makeOne()
+        self.assertEqual(policy.domain(DOMAIN), MEMBER)
+
+    def test_all_users(self):
+        policy = self._makeOne()
+        self.assertEqual(policy.all_users(), 'allUsers')
+
+    def test_authenticated_users(self):
+        policy = self._makeOne()
+        self.assertEqual(policy.authenticated_users(), 'allAuthenticatedUsers')
+
+    def test_from_api_repr_only_etag(self):
+        RESOURCE = {
+            'etag': 'ACAB',
+        }
+        klass = self._getTargetClass()
+        policy = klass.from_api_repr(RESOURCE)
+        self.assertEqual(policy.etag, 'ACAB')
+        self.assertEqual(policy.version, None)
+        self.assertEqual(list(policy.owners), [])
+        self.assertEqual(list(policy.writers), [])
+        self.assertEqual(list(policy.readers), [])
+
+    def test_from_api_repr_complete(self):
+        from gcloud.pubsub.iam import _OWNER_ROLE, _WRITER_ROLE, _READER_ROLE
+        OWNER1 = 'user:phred@example.com'
+        OWNER2 = 'group:cloud-logs@google.com'
+        WRITER1 = 'domain:google.com'
+        WRITER2 = 'user:phred@example.com'
+        READER1 = 'serviceAccount:1234-abcdef@service.example.com'
+        READER2 = 'user:phred@example.com'
+        RESOURCE = {
+            'etag': 'DEADBEEF',
+            'version': 17,
+            'bindings': [
+                {'role': _OWNER_ROLE, 'members': [OWNER1, OWNER2]},
+                {'role': _WRITER_ROLE, 'members': [WRITER1, WRITER2]},
+                {'role': _READER_ROLE, 'members': [READER1, READER2]},
+            ],
+        }
+        klass = self._getTargetClass()
+        policy = klass.from_api_repr(RESOURCE)
+        self.assertEqual(policy.etag, 'DEADBEEF')
+        self.assertEqual(policy.version, 17)
+        self.assertEqual(sorted(policy.owners), [OWNER2, OWNER1])
+        self.assertEqual(sorted(policy.writers), [WRITER1, WRITER2])
+        self.assertEqual(sorted(policy.readers), [READER1, READER2])
+
+    def test_from_api_repr_bad_role(self):
+        BOGUS1 = 'user:phred@example.com'
+        BOGUS2 = 'group:cloud-logs@google.com'
+        RESOURCE = {
+            'etag': 'DEADBEEF',
+            'version': 17,
+            'bindings': [
+                {'role': 'nonesuch', 'members': [BOGUS1, BOGUS2]},
+            ],
+        }
+        klass = self._getTargetClass()
+        with self.assertRaises(ValueError):
+            klass.from_api_repr(RESOURCE)
+
+    def test_to_api_repr_defaults(self):
+        policy = self._makeOne()
+        self.assertEqual(policy.to_api_repr(), {})
+
+    def test_to_api_repr_only_etag(self):
+        policy = self._makeOne('DEADBEEF')
+        self.assertEqual(policy.to_api_repr(), {'etag': 'DEADBEEF'})
+
+    def test_to_api_repr_full(self):
+        from gcloud.pubsub.iam import _OWNER_ROLE, _WRITER_ROLE, _READER_ROLE
+        OWNER1 = 'group:cloud-logs@google.com'
+        OWNER2 = 'user:phred@example.com'
+        WRITER1 = 'domain:google.com'
+        WRITER2 = 'user:phred@example.com'
+        READER1 = 'serviceAccount:1234-abcdef@service.example.com'
+        READER2 = 'user:phred@example.com'
+        EXPECTED = {
+            'etag': 'DEADBEEF',
+            'version': 17,
+            'bindings': [
+                {'role': _OWNER_ROLE, 'members': [OWNER1, OWNER2]},
+                {'role': _WRITER_ROLE, 'members': [WRITER1, WRITER2]},
+                {'role': _READER_ROLE, 'members': [READER1, READER2]},
+            ],
+        }
+        policy = self._makeOne('DEADBEEF', 17)
+        policy.owners.add(OWNER1)
+        policy.owners.add(OWNER2)
+        policy.writers.add(WRITER1)
+        policy.writers.add(WRITER2)
+        policy.readers.add(READER1)
+        policy.readers.add(READER2)
+        self.assertEqual(policy.to_api_repr(), EXPECTED)
diff --git a/gcloud/pubsub/test_topic.py b/gcloud/pubsub/test_topic.py
index fa0c908efacb..595ab33284cf 100644
--- a/gcloud/pubsub/test_topic.py
+++ b/gcloud/pubsub/test_topic.py
@@ -452,6 +452,73 @@ def test_list_subscriptions_missing_key(self):
                          % (PROJECT, TOPIC_NAME))
         self.assertEqual(req['query_params'], {})
 
+    def test_get_iam_policy_w_bound_client(self):
+        OWNER1 = 'user:phred@example.com'
+        OWNER2 = 'group:cloud-logs@google.com'
+        WRITER1 = 'domain:google.com'
+        WRITER2 = 'user:phred@example.com'
+        READER1 = 'serviceAccount:1234-abcdef@service.example.com'
+        READER2 = 'user:phred@example.com'
+        POLICY = {
+            'etag': 'DEADBEEF',
+            'version': 17,
+            'bindings': [
+                {'role': 'roles/owner', 'members': [OWNER1, OWNER2]},
+                {'role': 'roles/writer', 'members': [WRITER1, WRITER2]},
+                {'role': 'roles/reader', 'members': [READER1, READER2]},
+            ],
+        }
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        TOPIC_NAME = 'topic_name'
+        PATH = 'projects/%s/topics/%s:getIamPolicy' % (PROJECT, TOPIC_NAME)
+
+        conn = _Connection(POLICY)
+        CLIENT = _Client(project=PROJECT, connection=conn)
+        topic = self._makeOne(TOPIC_NAME, client=CLIENT)
+
+        policy = topic.get_iam_policy()
+
+        self.assertEqual(policy.etag, 'DEADBEEF')
+        self.assertEqual(policy.version, 17)
+        self.assertEqual(sorted(policy.owners), [OWNER2, OWNER1])
+        self.assertEqual(sorted(policy.writers), [WRITER1, WRITER2])
+        self.assertEqual(sorted(policy.readers), [READER1, READER2])
+
+        self.assertEqual(len(conn._requested), 1)
+        req = conn._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/%s' % PATH)
+
+    def test_get_iam_policy_w_alternate_client(self):
+        POLICY = {
+            'etag': 'ACAB',
+        }
+        TOPIC_NAME = 'topic_name'
+        PROJECT = 'PROJECT'
+        TOPIC_NAME = 'topic_name'
+        PATH = 'projects/%s/topics/%s:getIamPolicy' % (PROJECT, TOPIC_NAME)
+
+        conn1 = _Connection()
+        conn2 = _Connection(POLICY)
+        CLIENT1 = _Client(project=PROJECT, connection=conn1)
+        CLIENT2 = _Client(project=PROJECT, connection=conn2)
+        topic = self._makeOne(TOPIC_NAME, client=CLIENT1)
+
+        policy = topic.get_iam_policy(client=CLIENT2)
+
+        self.assertEqual(policy.etag, 'ACAB')
+        self.assertEqual(policy.version, None)
+        self.assertEqual(sorted(policy.owners), [])
+        self.assertEqual(sorted(policy.writers), [])
+        self.assertEqual(sorted(policy.readers), [])
+
+        self.assertEqual(len(conn1._requested), 0)
+        self.assertEqual(len(conn2._requested), 1)
+        req = conn2._requested[0]
+        self.assertEqual(req['method'], 'GET')
+        self.assertEqual(req['path'], '/%s' % PATH)
+
 
 class TestBatch(unittest2.TestCase):
 
diff --git a/gcloud/pubsub/topic.py b/gcloud/pubsub/topic.py
index b2143b86be72..2269993dc302 100644
--- a/gcloud/pubsub/topic.py
+++ b/gcloud/pubsub/topic.py
@@ -21,6 +21,7 @@
 from gcloud.exceptions import NotFound
 from gcloud.pubsub._helpers import subscription_name_from_path
 from gcloud.pubsub._helpers import topic_name_from_path
+from gcloud.pubsub.iam import Policy
 from gcloud.pubsub.subscription import Subscription
 
 
@@ -258,6 +259,25 @@ def list_subscriptions(self, page_size=None, page_token=None, client=None):
             subscriptions.append(Subscription(sub_name, self))
         return subscriptions, resp.get('nextPageToken')
 
+    def get_iam_policy(self, client=None):
+        """Fetch the IAM policy for the topic.
+
+        See:
+        https://cloud.google.com/pubsub/reference/rest/v1/projects.topics/getIamPolicy
+
+        :type client: :class:`gcloud.pubsub.client.Client` or ``NoneType``
+        :param client: the client to use.  If not passed, falls back to the
+                       ``client`` stored on the current batch.
+
+        :rtype: :class:`gcloud.pubsub.iam.Policy`
+        :returns: policy created from the resource returned by the
+                  ``getIamPolicy`` API request.
+        """
+        client = self._require_client(client)
+        path = '%s:getIamPolicy' % (self.path,)
+        resp = client.connection.api_request(method='GET', path=path)
+        return Policy.from_api_repr(resp)
+
 
 class Batch(object):
     """Context manager:  collect messages to publish via a single API call.