Merge lp:~maxiberta/launchpad/named-auth-tokens into lp:launchpad

[Maximiliano Bertacchini]

Add new named ArchiveAuthToken API.

Add new named ArchiveAuthToken API. This is required for UA customer delivery of kernel livepatches. See required DB changes in https://code.launchpad.net/~maxiberta/launchpad/db-named-auth-tokens/+merge/299433 .

Still missing:
* Hide behind a feature flag (to submit in a different MP, if that's ok).

Also:
* Use "authorize" instead of "authorise" in Archive related classes.
* Some extra cleanups.

[Colin Watson (cjwatson)]

Good start, thanks, but some bits to fix up.

review: Needs Fixing

[Maximiliano Bertacchini (maxiberta)]

Will fix asap. Thanks!

[Maximiliano Bertacchini (maxiberta)]

Updated with feature flag + fixes and improvements.

[Colin Watson (cjwatson)]

review: Approve

=== modified file 'lib/lp/services/features/flags.py'
--- lib/lp/services/features/flags.py	2016-01-25 12:21:31 +0000
+++ lib/lp/services/features/flags.py	2016-07-07 21:32:46 +0000
@@ -227,6 +227,12 @@
      'disabled',
      'Git recipes',
      'https://help.launchpad.net/Packaging/SourceBuilds'),
+    ('soyuz.named_auth_token.allow_new',
+     'boolean',
+     'If true, allow creation of named authorization tokens for archives.',
+     'disabled',
+     'Named authorization tokens for archives',
+     ''),
     ])
 
 # The set of all flag names that are documented.

=== modified file 'lib/lp/soyuz/interfaces/archive.py'
--- lib/lp/soyuz/interfaces/archive.py	2016-01-26 15:47:37 +0000
+++ lib/lp/soyuz/interfaces/archive.py	2016-07-07 21:32:46 +0000
@@ -20,6 +20,7 @@
     'CannotUploadToPPA',
     'CannotUploadToPocket',
     'CannotUploadToSeries',
+    'DuplicateTokenName',
     'FULL_COMPONENT_SUPPORT',
     'IArchive',
     'IArchiveAdmin',
@@ -38,6 +39,8 @@
     'InvalidPocketForPPA',
     'IPPA',
     'MAIN_ARCHIVE_PURPOSES',
+    'NAMED_AUTH_TOKEN_FEATURE_FLAG',
+    'NamedAuthTokenFeatureDisabled',
     'NoRightsForArchive',
     'NoRightsForComponent',
     'NoSuchPPA',
@@ -92,6 +95,7 @@
     Text,
     TextLine,
     )
+from zope.security.interfaces import Unauthorized
 
 from lp import _
 from lp.app.errors import NameLookupFailed
@@ -111,6 +115,9 @@
 from lp.soyuz.interfaces.component import IComponent
 
 
+NAMED_AUTH_TOKEN_FEATURE_FLAG = u"soyuz.named_auth_token.allow_new"
+
+
 @error_status(httplib.BAD_REQUEST)
 class ArchiveDependencyError(Exception):
     """Raised when an `IArchiveDependency` does not fit the context archive.
@@ -312,6 +319,21 @@
             self._fmt % {'processor': processor.name})
 
 
+@error_status(httplib.CONFLICT)
+class DuplicateTokenName(Exception):
+    """Raised when creating a named token and an active token for this archive
+     with this name already exists."""
+
+
+@error_status(httplib.UNAUTHORIZED)
+class NamedAuthTokenFeatureDisabled(Unauthorized):
+    """Only certain users can create named authorization tokens."""
+
+    def __init__(self):
+        super(NamedAuthTokenFeatureDisabled, self).__init__(
+            "You do not have permission to create named authorization tokens")
+
+
 class IArchivePublic(IPrivacy, IHasOwner):
     """An Archive interface for publicly available operations."""
     # Most of this stuff should really be on View, but it's needed for
@@ -526,12 +548,12 @@
         """
 
     def newAuthToken(person, token=None, date_created=None):
-        """Create a new authorisation token.
+        """Create a new authorization token.
 
-        :param person: An IPerson whom this token is for
+        :param person: An IPerson whom this token is for.
         :param token: Optional unicode text to use as the token. One will be
-            generated if not given
-        :param date_created: Optional, defaults to now
+            generated if not given.
+        :param date_created: Optional, defaults to now.
 
         :return: A new IArchiveAuthToken
         """
@@ -1324,7 +1346,7 @@
     @operation_returns_collection_of(Interface)
     @export_read_operation()
     def getQueueAdminsForComponent(component_name):
-        """Return `IArchivePermission` records for authorised queue admins.
+        """Return `IArchivePermission` records for authorized queue admins.
 
         :param component_name: An `IComponent` or textual name for the
             component.
@@ -1377,7 +1399,7 @@
     @export_read_operation()
     @operation_for_version("devel")
     def getQueueAdminsForPocket(pocket, distroseries=None):
-        """Return `IArchivePermission` records for authorised queue admins.
+        """Return `IArchivePermission` records for authorized queue admins.
 
         :param pocket: A `PackagePublishingPocket`.
         :param distroseries: An optional `IDistroSeries`.
@@ -2062,6 +2084,7 @@
         :return: a `IArchiveDependency` object targeted to the context
             `IArchive` requiring 'dependency' `IArchive`.
         """
+
     @operation_parameters(
         dependency=Reference(schema=Interface, required=True),
         # Really IArchive
@@ -2074,6 +2097,61 @@
         :param dependency: is an `IArchive` object.
         """
 
+    @operation_parameters(
+        name=TextLine(title=_("Authorization token name"), required=True),
+        token=TextLine(
+            title=_("Optional secret for this named token"), required=False))
+    @export_write_operation()
+    @operation_for_version("devel")
+    def newNamedAuthToken(name, token=None):
+        """Create a new named authorization token.
+
+        :param name: An identifier string for this token.
+        :param token: Optional unicode text to use as the token. One will be
+            generated if not given.
+
+        :return: A dictionary where the value of `token` is the secret and
+            the value of `archive_url` is the externally-usable archive URL
+            including basic auth.
+        """
+
+    @operation_parameters(
+        name=TextLine(title=_("Authorization token name"), required=True))
+    @export_read_operation()
+    @operation_for_version("devel")
+    def getNamedAuthToken(name):
+        """Return a named authorization token for the given name in this
+         archive.
+
+        :param name: The identifier string for a token.
+
+        :return: A dictionary where the value of `token` is the secret and
+            the value of `archive_url` is the externally-usable archive URL
+            including basic auth.
+        :raises NotFoundError: if no matching token could be found.
+        """
+
+    @export_read_operation()
+    @operation_for_version("devel")
+    def getNamedAuthTokens():
+        """Return a list of named authorization tokens for this archive.
+
+        :return: A list of dictionaries where the value of `token` is the
+            secret and the value of `archive_url` is the externally-usable
+            archive URL including basic auth.
+        """
+
+    @operation_parameters(
+        name=TextLine(title=_("Authorization token name"), required=True))
+    @export_write_operation()
+    @operation_for_version("devel")
+    def revokeNamedAuthToken(name):
+        """Deactivates a named authorization token.
+
+        :param name: The identifier string for a token.
+        :raises NotFoundError: if no matching token could be found.
+        """
+
 
 class IArchiveAdmin(Interface):
     """Archive interface for operations restricted by commercial."""

=== modified file 'lib/lp/soyuz/interfaces/archiveauthtoken.py'
--- lib/lp/soyuz/interfaces/archiveauthtoken.py	2013-01-07 02:40:55 +0000
+++ lib/lp/soyuz/interfaces/archiveauthtoken.py	2016-07-07 21:32:46 +0000
@@ -27,16 +27,16 @@
 
 
 class IArchiveAuthTokenView(Interface):
-    """Interface for Archive Authorisation Tokens requiring launchpad.View."""
+    """Interface for Archive Authorization Tokens requiring launchpad.View."""
     id = Int(title=_('ID'), required=True, readonly=True)
 
     archive = Reference(
         IArchive, title=_("Archive"), required=True, readonly=True,
-        description=_("The archive for this authorisation token."))
+        description=_("The archive for this authorization token."))
 
     person = Reference(
-        IPerson, title=_("Person"), required=True, readonly=True,
-        description=_("The person for this authorisation token."))
+        IPerson, title=_("Person"), required=False, readonly=True,
+        description=_("The person for this authorization token."))
     person_id = Attribute('db person value')
 
     date_created = Datetime(
@@ -56,9 +56,20 @@
         description=_(
             "External archive URL including basic auth for this person"))
 
-    def deactivate(self):
+    name = TextLine(
+        title=_("Name"), required=False, readonly=True,
+        description=_(
+            "The name in the case of a named authorization token, or None."))
+
+    def deactivate():
         """Deactivate the token by setting date_deactivated to UTC_NOW."""
 
+    def asDict():
+        """Returns a dictionary where the value of `token` is the secret and
+        the value of `archive_url` is the externally-usable archive URL
+        including basic auth.
+        """
+
 
 class IArchiveAuthTokenEdit(Interface):
     """Interface for Archive Auth Tokens requiring launchpad.Edit."""
@@ -74,15 +85,15 @@
     def get(token_id):
         """Retrieve a token by its database ID.
 
-        :param token_id: The database ID
-        :return: An object conforming to IArchiveAuthToken
+        :param token_id: The database ID.
+        :return: An object conforming to `IArchiveAuthToken`.
         """
 
     def getByToken(token):
         """Retrieve a token by its token text.
 
         :param token: The token text for the token.
-        :return: An object conforming to IArchiveAuthToken
+        :return: An object conforming to `IArchiveAuthToken`.
         """
 
     def getByArchive(archive):
@@ -97,5 +108,20 @@
 
         :param archive: The archive to which the token corresponds.
         :param person: The person to which the token corresponds.
-        :return An object conforming to IArchiveAuthToken or None.
+        :return: An `IArchiveAuthToken` or None.
+        """
+
+    def getActiveNamedTokenForArchive(archive, name):
+        """Retrieve an active named token for the given archive and name.
+
+        :param archive: The archive to which the token corresponds.
+        :param name: The name of a named authorization token.
+        :return An object conforming to `IArchiveAuthToken` or None.
+        """
+
+    def getActiveNamedTokensForArchive(archive):
+        """Retrieve all active named tokens for the given archive.
+
+        :param archive: The archive to which the tokens correspond.
+        :return: A result set containing `IArchiveAuthToken`s.
         """

=== modified file 'lib/lp/soyuz/model/archive.py'
--- lib/lp/soyuz/model/archive.py	2016-03-14 23:42:45 +0000
+++ lib/lp/soyuz/model/archive.py	2016-07-07 21:32:46 +0000
@@ -110,6 +110,7 @@
     sqlvalues,
     )
 from lp.services.database.stormexpr import BulkUpdate
+from lp.services.features import getFeatureFlag
 from lp.services.job.interfaces.job import JobStatus
 from lp.services.librarian.model import (
     LibraryFileAlias,
@@ -149,6 +150,7 @@
     CannotUploadToSeries,
     ComponentNotFound,
     default_name_by_purpose,
+    DuplicateTokenName,
     FULL_COMPONENT_SUPPORT,
     IArchive,
     IArchiveSet,
@@ -160,6 +162,8 @@
     InvalidPocketForPPA,
     IPPA,
     MAIN_ARCHIVE_PURPOSES,
+    NAMED_AUTH_TOKEN_FEATURE_FLAG,
+    NamedAuthTokenFeatureDisabled,
     NoRightsForArchive,
     NoRightsForComponent,
     NoSuchPPA,
@@ -1948,6 +1952,60 @@
         IStore(ArchiveAuthToken).add(archive_auth_token)
         return archive_auth_token
 
+    def newNamedAuthToken(self, name, token=None):
+        """See `IArchive`."""
+
+        if not getFeatureFlag(NAMED_AUTH_TOKEN_FEATURE_FLAG):
+            raise NamedAuthTokenFeatureDisabled()
+
+        # Bail if the archive isn't private
+        if not self.private:
+            raise ArchiveNotPrivate("Archive must be private.")
+
+        try:
+            # Check for duplicate name.
+            self.getNamedAuthToken(name)
+            raise DuplicateTokenName(
+                "An active token with name %s for archive %s already exists." %
+                (name, self.displayname))
+        except NotFoundError:
+            # No duplicate name found: continue.
+            pass
+
+        # Now onto the actual token creation:
+        if token is None:
+            token = create_token(20)
+        archive_auth_token = ArchiveAuthToken()
+        archive_auth_token.archive = self
+        archive_auth_token.name = name
+        archive_auth_token.token = token
+        IStore(ArchiveAuthToken).add(archive_auth_token)
+        return archive_auth_token.asDict()
+
+    def getNamedAuthToken(self, name):
+        """See `IArchive`."""
+        token_set = getUtility(IArchiveAuthTokenSet)
+        auth_token = token_set.getActiveNamedTokenForArchive(self, name)
+        if auth_token is not None:
+            return auth_token.asDict()
+        else:
+            raise NotFoundError(name)
+
+    def getNamedAuthTokens(self):
+        """See `IArchive`."""
+        token_set = getUtility(IArchiveAuthTokenSet)
+        auth_tokens = token_set.getActiveNamedTokensForArchive(self)
+        return [auth_token.asDict() for auth_token in auth_tokens]
+
+    def revokeNamedAuthToken(self, name):
+        """See `IArchive`."""
+        token_set = getUtility(IArchiveAuthTokenSet)
+        auth_token = token_set.getActiveNamedTokenForArchive(self, name)
+        if auth_token is not None:
+            auth_token.deactivate()
+        else:
+            raise NotFoundError(name)
+
     def newSubscription(self, subscriber, registrant, date_expires=None,
                         description=None):
         """See `IArchive`."""

=== modified file 'lib/lp/soyuz/model/archiveauthtoken.py'
--- lib/lp/soyuz/model/archiveauthtoken.py	2015-10-21 09:37:08 +0000
+++ lib/lp/soyuz/model/archiveauthtoken.py	2016-07-07 21:32:46 +0000
@@ -39,7 +39,7 @@
     archive_id = Int(name='archive', allow_none=False)
     archive = Reference(archive_id, 'Archive.id')
 
-    person_id = Int(name='person', allow_none=False)
+    person_id = Int(name='person', allow_none=True)
     person = Reference(person_id, 'Person.id')
 
     date_created = DateTime(
@@ -50,6 +50,8 @@
 
     token = Unicode(name='token', allow_none=False)
 
+    name = Unicode(name='name', allow_none=True)
+
     def deactivate(self):
         """See `IArchiveAuthTokenSet`."""
         self.date_deactivated = UTC_NOW
@@ -58,10 +60,16 @@
     def archive_url(self):
         """Return a custom archive url for basic authentication."""
         normal_url = URI(self.archive.archive_url)
-        auth_url = normal_url.replace(
-            userinfo="%s:%s" % (self.person.name, self.token))
+        if self.name:
+            name = '+' + self.name
+        else:
+            name = self.person.name
+        auth_url = normal_url.replace(userinfo="%s:%s" % (name, self.token))
         return str(auth_url)
 
+    def asDict(self):
+        return {"token": self.token, "archive_url": self.archive_url}
+
 
 @implementer(IArchiveAuthTokenSet)
 class ArchiveAuthTokenSet:
@@ -87,9 +95,15 @@
 
     def getActiveTokenForArchiveAndPerson(self, archive, person):
         """See `IArchiveAuthTokenSet`."""
-        store = Store.of(archive)
-        return store.find(
-            ArchiveAuthToken,
-            ArchiveAuthToken.archive == archive,
-            ArchiveAuthToken.person == person,
-            ArchiveAuthToken.date_deactivated == None).one()
+        return self.getByArchive(archive).find(
+            ArchiveAuthToken.person == person).one()
+
+    def getActiveNamedTokenForArchive(self, archive, name):
+        """See `IArchiveAuthTokenSet`."""
+        return self.getByArchive(archive).find(
+            ArchiveAuthToken.name == name).one()
+
+    def getActiveNamedTokensForArchive(self, archive):
+        """See `IArchiveAuthTokenSet`."""
+        return self.getByArchive(archive).find(
+            ArchiveAuthToken.name != None)

=== modified file 'lib/lp/soyuz/templates/person-archive-subscription.pt'
--- lib/lp/soyuz/templates/person-archive-subscription.pt	2012-03-01 18:17:56 +0000
+++ lib/lp/soyuz/templates/person-archive-subscription.pt	2016-07-07 21:32:46 +0000
@@ -50,7 +50,7 @@
     <div id="regenerate_token" class="portlet" style="clear:both">
       <h2>Reset password</h2>
       <p>If you believe the security of your password for this access
-         has been compromised, you reset your password. After you've
+         has been compromised, you should reset your password. After you've
          requested a new password, you'll see new "sources.list" entries
          on this page. You'll need to update them on your computer.
       </p>

=== modified file 'lib/lp/soyuz/tests/test_archive.py'
--- lib/lp/soyuz/tests/test_archive.py	2016-04-07 00:04:42 +0000
+++ lib/lp/soyuz/tests/test_archive.py	2016-07-07 21:32:46 +0000
@@ -41,6 +41,7 @@
 from lp.registry.interfaces.teammembership import TeamMembershipStatus
 from lp.services.database.interfaces import IStore
 from lp.services.database.sqlbase import sqlvalues
+from lp.services.features.testing import FeatureFixture
 from lp.services.job.interfaces.job import JobStatus
 from lp.services.propertycache import (
     clear_property_cache,
@@ -65,21 +66,26 @@
 from lp.soyuz.interfaces.archive import (
     ArchiveDependencyError,
     ArchiveDisabled,
+    ArchiveNotPrivate,
     CannotCopy,
     CannotModifyArchiveProcessor,
     CannotUploadToPocket,
     CannotUploadToPPA,
     CannotUploadToSeries,
+    DuplicateTokenName,
     IArchiveSet,
     InsufficientUploadRights,
     InvalidPocketForPartnerArchive,
     InvalidPocketForPPA,
+    NAMED_AUTH_TOKEN_FEATURE_FLAG,
+    NamedAuthTokenFeatureDisabled,
     NoRightsForArchive,
     NoRightsForComponent,
     NoSuchPPA,
     RedirectedPocket,
     VersionRequiresName,
     )
+from lp.soyuz.interfaces.archiveauthtoken import IArchiveAuthTokenSet
 from lp.soyuz.interfaces.archivepermission import IArchivePermissionSet
 from lp.soyuz.interfaces.binarypackagebuild import BuildSetStatus
 from lp.soyuz.interfaces.binarypackagename import IBinaryPackageNameSet
@@ -1246,6 +1252,23 @@
             self.assertEqual("really secret", self.archive.buildd_secret)
 
 
+class TestNamedAuthTokenFeatureFlag(TestCaseWithFactory):
+    layer = LaunchpadZopelessLayer
+
+    def test_feature_flag_disabled(self):
+        # With feature flag disabled, we will not create new named auth tokens.
+        private_ppa = self.factory.makeArchive(private=True)
+        with FeatureFixture({NAMED_AUTH_TOKEN_FEATURE_FLAG: u""}):
+            self.assertRaises(NamedAuthTokenFeatureDisabled,
+                              private_ppa.newNamedAuthToken, u"tokenname")
+
+    def test_feature_flag_disabled_by_default(self):
+         # Without a feature flag, we will not create new named auth tokens.
+        private_ppa = self.factory.makeArchive(private=True)
+        self.assertRaises(NamedAuthTokenFeatureDisabled,
+            private_ppa.newNamedAuthToken, u"tokenname")
+
+
 class TestArchiveTokens(TestCaseWithFactory):
     layer = LaunchpadZopelessLayer
 
@@ -1255,13 +1278,14 @@
         self.private_ppa = self.factory.makeArchive(owner=owner, private=True)
         self.joe = self.factory.makePerson(name='joe')
         self.private_ppa.newSubscription(self.joe, owner)
+        self.useFixture(FeatureFixture({NAMED_AUTH_TOKEN_FEATURE_FLAG: u"on"}))
 
     def test_getAuthToken_with_no_token(self):
-        token = self.private_ppa.getAuthToken(self.joe)
-        self.assertEqual(token, None)
+        self.assertIsNone(self.private_ppa.getAuthToken(self.joe))
 
     def test_getAuthToken_with_token(self):
         token = self.private_ppa.newAuthToken(self.joe)
+        self.assertIsNone(token.name)
         self.assertEqual(self.private_ppa.getAuthToken(self.joe), token)
 
     def test_getArchiveSubscriptionURL(self):
@@ -1269,6 +1293,74 @@
         token = self.private_ppa.getAuthToken(self.joe)
         self.assertEqual(token.archive_url, url)
 
+    def test_newNamedAuthToken_private_archive(self):
+        res = self.private_ppa.newNamedAuthToken(u"tokenname")
+        token = getUtility(IArchiveAuthTokenSet).getActiveNamedTokenForArchive(
+            self.private_ppa, u"tokenname")
+        self.assertIsNotNone(token)
+        self.assertIsNone(token.person)
+        self.assertEqual("tokenname", token.name)
+        self.assertIsNotNone(token.token)
+        self.assertEqual(self.private_ppa, token.archive)
+        self.assertIn(
+            "://+%s:%s@" % (token.name, token.token), token.archive_url)
+        self.assertDictEqual(
+            {"token": token.token, "archive_url": token.archive_url},
+            res
+            )
+
+    def test_newNamedAuthToken_public_archive(self):
+        public_ppa = self.factory.makeArchive(private=False)
+        self.assertRaises(ArchiveNotPrivate,
+            public_ppa.newNamedAuthToken, u"tokenname")
+
+    def test_newNamedAuthToken_duplicate_name(self):
+        self.private_ppa.newNamedAuthToken(u"tokenname")
+        self.assertRaises(DuplicateTokenName,
+            self.private_ppa.newNamedAuthToken, u"tokenname")
+
+    def test_newNamedAuthToken_with_custom_secret(self):
+        self.private_ppa.newNamedAuthToken(u"tokenname", u"somesecret")
+        token = getUtility(IArchiveAuthTokenSet).getActiveNamedTokenForArchive(
+            self.private_ppa, u"tokenname")
+        self.assertEqual(u"somesecret", token.token)
+
+    def test_getNamedAuthToken_with_no_token(self):
+        self.assertRaises(
+            NotFoundError, self.private_ppa.getNamedAuthToken, u"tokenname")
+
+    def test_getNamedAuthToken_with_token(self):
+        res = self.private_ppa.newNamedAuthToken(u"tokenname")
+        self.assertEqual(
+            self.private_ppa.getNamedAuthToken(u"tokenname"),
+            res)
+
+    def test_revokeNamedAuthToken_with_token(self):
+        self.private_ppa.newNamedAuthToken(u"tokenname")
+        token = getUtility(IArchiveAuthTokenSet).getActiveNamedTokenForArchive(
+            self.private_ppa, u"tokenname")
+        self.private_ppa.revokeNamedAuthToken(u"tokenname")
+        self.assertIsNotNone(token.date_deactivated)
+
+    def test_revokeNamedAuthToken_with_no_token(self):
+        self.assertRaises(
+            NotFoundError, self.private_ppa.revokeNamedAuthToken, u"tokenname")
+
+    def test_getNamedAuthToken_with_revoked_token(self):
+        self.private_ppa.newNamedAuthToken(u"tokenname")
+        self.private_ppa.revokeNamedAuthToken(u"tokenname")
+        self.assertRaises(
+            NotFoundError, self.private_ppa.getNamedAuthToken, u"tokenname")
+
+    def test_getNamedAuthTokens(self):
+        res1 = self.private_ppa.newNamedAuthToken(u"tokenname1")
+        res2 = self.private_ppa.newNamedAuthToken(u"tokenname2")
+        self.private_ppa.newNamedAuthToken(u"tokenname3")
+        self.private_ppa.revokeNamedAuthToken(u"tokenname3")
+        self.assertContentEqual(
+            [res1, res2],
+            self.private_ppa.getNamedAuthTokens())
+
 
 class TestGetBinaryPackageRelease(TestCaseWithFactory):
     """Ensure that getBinaryPackageRelease works as expected."""
@@ -3505,9 +3597,6 @@
 
     layer = LaunchpadFunctionalLayer
 
-    def assertDictEqual(self, one, two):
-        self.assertContentEqual(one.items(), two.items())
-
     def test_cprov_build_counters_in_sampledata(self):
         cprov_archive = getUtility(IPersonSet).getByName("cprov").archive
         expected_counters = {

=== modified file 'lib/lp/testing/__init__.py'
--- lib/lp/testing/__init__.py	2015-11-08 01:05:24 +0000
+++ lib/lp/testing/__init__.py	2016-07-07 21:32:46 +0000
@@ -619,14 +619,15 @@
         self.assertIsNot(
             None, pattern.search(normalise_whitespace(text)), text)
 
-    def assertIsInstance(self, instance, assert_class):
+    def assertIsInstance(self, instance, assert_class, msg=None):
         """Assert that an instance is an instance of assert_class.
 
         instance and assert_class have the same semantics as the parameters
         to isinstance.
         """
-        self.assertTrue(zope_isinstance(instance, assert_class),
-            '%r is not an instance of %r' % (instance, assert_class))
+        if msg is None:
+            msg = '%r is not an instance of %r' % (instance, assert_class)
+        self.assertTrue(zope_isinstance(instance, assert_class), msg)
 
     def assertIsNot(self, expected, observed, msg=None):
         """Assert that `expected` is not the same object as `observed`."""