Merge ~cjwatson/launchpad:access-token-api into launchpad:master

[Colin Watson]

Add webservice API for personal access tokens

This currently supports issuing, querying, and revoking personal access tokens for Git repositories.

[Ioana Lasc (ilasc)]

looks good!

review: Approve

diff --git a/lib/lp/_schema_circular_imports.py b/lib/lp/_schema_circular_imports.py
index 7c3c549..204fa61 100644
--- a/lib/lp/_schema_circular_imports.py
+++ b/lib/lp/_schema_circular_imports.py
@@ -139,6 +139,7 @@ from lp.registry.interfaces.sourcepackage import (
 from lp.registry.interfaces.ssh import ISSHKey
 from lp.registry.interfaces.teammembership import ITeamMembership
 from lp.registry.interfaces.wikiname import IWikiName
+from lp.services.auth.interfaces import IAccessToken
 from lp.services.comments.interfaces.conversation import IComment
 from lp.services.fields import InlineObject
 from lp.services.messages.interfaces.message import (
@@ -691,6 +692,9 @@ patch_collection_property(
 patch_collection_property(
     IHasSpecifications, 'api_valid_specifications', ISpecification)
 
+# IAccessToken
+patch_reference_property(IAccessToken, 'git_repository', IGitRepository)
+
 
 ###
 #
diff --git a/lib/lp/code/interfaces/gitrepository.py b/lib/lp/code/interfaces/gitrepository.py
index dec0bf6..363356d 100644
--- a/lib/lp/code/interfaces/gitrepository.py
+++ b/lib/lp/code/interfaces/gitrepository.py
@@ -85,6 +85,8 @@ from lp.registry.interfaces.personociproject import IPersonOCIProjectFactory
 from lp.registry.interfaces.personproduct import IPersonProductFactory
 from lp.registry.interfaces.product import IProduct
 from lp.registry.interfaces.role import IPersonRoles
+from lp.services.auth.enums import AccessTokenScope
+from lp.services.auth.interfaces import IAccessTokenTarget
 from lp.services.fields import (
     InlineObject,
     PersonChoice,
@@ -690,20 +692,41 @@ class IGitRepositoryView(IHasRecipes):
         :return: A `ResultSet` of `IGitActivity`.
         """
 
+    # XXX cjwatson 2021-10-13: This should move to IAccessTokenTarget, but
+    # currently has rather too much backward-compatibility code for that.
+    @operation_parameters(
+        description=TextLine(
+            title=_("A short description of the token."), required=False),
+        scopes=List(
+            title=_("A list of scopes to be granted by this token."),
+            value_type=Choice(vocabulary=AccessTokenScope), required=False),
+        date_expires=Datetime(
+            title=_("When the token should expire."), required=False))
     @export_write_operation()
     @operation_for_version("devel")
-    def issueAccessToken():
+    def issueAccessToken(description=None, scopes=None, date_expires=None):
         """Issue an access token for this repository.
 
         Access tokens can be used to push to this repository over HTTPS.
         They are only valid for a single repository, and have a short expiry
-        period (currently one week), so at the moment they are only suitable
-        in some limited situations.
+        period (currently fixed at one week), so at the moment they are only
+        suitable in some limited situations.  By default they are currently
+        implemented as macaroons.
+
+        If `description` and `scopes` are both given, then issue a personal
+        access token instead, either non-expiring or with an expiry time
+        given by `date_expires`.  These may be used in webservice API
+        requests for certain methods on this repository.
 
         This interface is experimental, and may be changed or removed
         without notice.
 
-        :return: A serialised macaroon.
+        :return: If `description` and `scopes` are both given, the secret
+            for a new personal access token (Launchpad only records the hash
+            of this secret and not the secret itself, so the caller must be
+            careful to save this; personal access tokens are in development
+            and may not entirely work yet).  Otherwise, a serialised
+            macaroon.
         """
 
 
@@ -782,7 +805,7 @@ class IGitRepositoryExpensiveRequest(Interface):
         that is not an admin or a registry expert."""
 
 
-class IGitRepositoryEdit(IWebhookTarget):
+class IGitRepositoryEdit(IWebhookTarget, IAccessTokenTarget):
     """IGitRepository methods that require launchpad.Edit permission."""
 
     @mutator_for(IGitRepositoryView["name"])
diff --git a/lib/lp/code/model/gitrepository.py b/lib/lp/code/model/gitrepository.py
index 8fb7cda..99d451b 100644
--- a/lib/lp/code/model/gitrepository.py
+++ b/lib/lp/code/model/gitrepository.py
@@ -177,6 +177,9 @@ from lp.registry.model.accesspolicy import (
     )
 from lp.registry.model.person import Person
 from lp.registry.model.teammembership import TeamParticipation
+from lp.services.auth.interfaces import IAccessTokenSet
+from lp.services.auth.model import AccessTokenTargetMixin
+from lp.services.auth.utils import create_access_token_secret
 from lp.services.config import config
 from lp.services.database import bulk
 from lp.services.database.constants import (
@@ -290,7 +293,8 @@ def git_repository_modified(repository, event):
 
 
 @implementer(IGitRepository, IHasOwner, IPrivacy, IInformationType)
-class GitRepository(StormBase, WebhookTargetMixin, GitIdentityMixin):
+class GitRepository(StormBase, WebhookTargetMixin, AccessTokenTargetMixin,
+                    GitIdentityMixin):
     """See `IGitRepository`."""
 
     __storm_table__ = 'GitRepository'
@@ -1580,19 +1584,43 @@ class GitRepository(StormBase, WebhookTargetMixin, GitIdentityMixin):
         return DecoratedResultSet(
             results, pre_iter_hook=preloadDataForActivities)
 
-    def issueAccessToken(self):
-        """See `IGitRepository`."""
+    def _issuePersonalAccessToken(self, user, description, scopes,
+                                  date_expires=None):
+        """Issue a personal access token for this repository."""
+        if user is None:
+            raise Unauthorized(
+                "Personal access tokens may only be issued for a logged-in "
+                "user.")
+        secret = create_access_token_secret()
+        getUtility(IAccessTokenSet).new(
+            secret, owner=user, description=description, target=self,
+            scopes=scopes, date_expires=date_expires)
+        return secret
+
+    # XXX cjwatson 2021-10-13: Remove this once lp.code.xmlrpc.git accepts
+    # pushes using personal access tokens.
+    def _issueMacaroon(self, user):
+        """Issue a macaroon for this repository."""
         issuer = getUtility(IMacaroonIssuer, "git-repository")
+        # Our security adapter has already done the checks we need, apart
+        # from forbidding anonymous users which is done by the issuer.
+        return removeSecurityProxy(issuer).issueMacaroon(
+            self, user=user).serialize()
+
+    def issueAccessToken(self, owner=None, description=None, scopes=None,
+                         date_expires=None):
+        """See `IGitRepository`."""
         # It's more usual in model code to pass the user as an argument,
         # e.g. using @call_with(user=REQUEST_USER) in the webservice
         # interface.  However, in this case that would allow anyone who
         # constructs a way to call this method not via the webservice to
         # issue a token for any user, which seems like a bad idea.
         user = getUtility(ILaunchBag).user
-        # Our security adapter has already done the checks we need, apart
-        # from forbidding anonymous users which is done by the issuer.
-        return removeSecurityProxy(issuer).issueMacaroon(
-            self, user=user).serialize()
+        if description is not None and scopes is not None:
+            return self._issuePersonalAccessToken(
+                user, description, scopes, date_expires=date_expires)
+        else:
+            return self._issueMacaroon(user)
 
     def canBeDeleted(self):
         """See `IGitRepository`."""
diff --git a/lib/lp/code/model/tests/test_gitrepository.py b/lib/lp/code/model/tests/test_gitrepository.py
index 78c3353..a832554 100644
--- a/lib/lp/code/model/tests/test_gitrepository.py
+++ b/lib/lp/code/model/tests/test_gitrepository.py
@@ -143,6 +143,8 @@ from lp.registry.interfaces.persondistributionsourcepackage import (
 from lp.registry.interfaces.personociproject import IPersonOCIProjectFactory
 from lp.registry.interfaces.personproduct import IPersonProductFactory
 from lp.registry.tests.test_accesspolicy import get_policies_for_artifact
+from lp.services.auth.enums import AccessTokenScope
+from lp.services.auth.interfaces import IAccessTokenSet
 from lp.services.authserver.xmlrpc import AuthServerAPIView
 from lp.services.config import config
 from lp.services.database.constants import UTC_NOW
@@ -4717,6 +4719,76 @@ class TestGitRepositoryWebservice(TestCaseWithFactory):
             b"git-repository macaroons may only be issued for a logged-in "
             b"user.", response.body)
 
+    def test_issueAccessToken_personal(self):
+        # A user can request a personal access token via the webservice API.
+        repository = self.factory.makeGitRepository()
+        # Write access to the repository isn't checked at this stage
+        # (although the access token will only be useful if the user has
+        # some kind of write access).
+        requester = self.factory.makePerson()
+        with person_logged_in(requester):
+            repository_url = api_url(repository)
+        webservice = webservice_for_person(
+            requester, permission=OAuthPermission.WRITE_PUBLIC,
+            default_api_version="devel")
+        response = webservice.named_post(
+            repository_url, "issueAccessToken", description="Test token",
+            scopes=["repository:build_status"])
+        self.assertEqual(200, response.status)
+        secret = response.jsonBody()
+        with person_logged_in(requester):
+            token = getUtility(IAccessTokenSet).getBySecret(secret)
+            self.assertThat(token, MatchesStructure(
+                owner=Equals(requester),
+                description=Equals("Test token"),
+                target=Equals(repository),
+                scopes=Equals([AccessTokenScope.REPOSITORY_BUILD_STATUS]),
+                date_expires=Is(None)))
+
+    def test_issueAccessToken_personal_with_expiry(self):
+        # A user can set an expiry time when requesting a personal access
+        # token via the webservice API.
+        repository = self.factory.makeGitRepository()
+        # Write access to the repository isn't checked at this stage
+        # (although the access token will only be useful if the user has
+        # some kind of write access).
+        requester = self.factory.makePerson()
+        with person_logged_in(requester):
+            repository_url = api_url(repository)
+        webservice = webservice_for_person(
+            requester, permission=OAuthPermission.WRITE_PUBLIC,
+            default_api_version="devel")
+        date_expires = datetime.now(pytz.UTC) + timedelta(days=30)
+        response = webservice.named_post(
+            repository_url, "issueAccessToken", description="Test token",
+            scopes=["repository:build_status"],
+            date_expires=date_expires.isoformat())
+        self.assertEqual(200, response.status)
+        secret = response.jsonBody()
+        with person_logged_in(requester):
+            token = getUtility(IAccessTokenSet).getBySecret(secret)
+            self.assertThat(token, MatchesStructure.byEquality(
+                owner=requester,
+                description="Test token",
+                target=repository,
+                scopes=[AccessTokenScope.REPOSITORY_BUILD_STATUS],
+                date_expires=date_expires))
+
+    def test_issueAccessToken_personal_anonymous(self):
+        # An anonymous user cannot request a personal access token via the
+        # webservice API.
+        repository = self.factory.makeGitRepository()
+        with person_logged_in(repository.owner):
+            repository_url = api_url(repository)
+        webservice = webservice_for_person(None, default_api_version="devel")
+        response = webservice.named_post(
+            repository_url, "issueAccessToken", description="Test token",
+            scopes=["repository:build_status"])
+        self.assertEqual(401, response.status)
+        self.assertEqual(
+            b"Personal access tokens may only be issued for a logged-in user.",
+            response.body)
+
 
 class TestGitRepositoryMacaroonIssuer(MacaroonTestMixin, TestCaseWithFactory):
     """Test GitRepository macaroon issuing and verification."""
diff --git a/lib/lp/services/auth/configure.zcml b/lib/lp/services/auth/configure.zcml
index bd76aa6..f300218 100644
--- a/lib/lp/services/auth/configure.zcml
+++ b/lib/lp/services/auth/configure.zcml
@@ -4,7 +4,9 @@
 
 <configure
     xmlns="http://namespaces.zope.org/zope"
+    xmlns:browser="http://namespaces.zope.org/browser"
     xmlns:i18n="http://namespaces.zope.org/i18n"
+    xmlns:webservice="http://namespaces.canonical.com/webservice"
     i18n_domain="launchpad">
 
     <class class="lp.services.auth.model.AccessToken">
@@ -23,4 +25,11 @@
         provides="lp.services.auth.interfaces.IAccessTokenSet">
         <allow interface="lp.services.auth.interfaces.IAccessTokenSet" />
     </securedutility>
+
+    <browser:url
+        for="lp.services.auth.interfaces.IAccessToken"
+        path_expression="string:+access-token/${id}"
+        attribute_to_parent="target" />
+
+    <webservice:register module="lp.services.auth.webservice" />
 </configure>
diff --git a/lib/lp/services/auth/interfaces.py b/lib/lp/services/auth/interfaces.py
index b8e3e7a..7e20ee0 100644
--- a/lib/lp/services/auth/interfaces.py
+++ b/lib/lp/services/auth/interfaces.py
@@ -7,9 +7,20 @@ __metaclass__ = type
 __all__ = [
     "IAccessToken",
     "IAccessTokenSet",
+    "IAccessTokenTarget",
     "IAccessTokenVerifiedRequest",
     ]
 
+from lazr.restful.declarations import (
+    call_with,
+    export_read_operation,
+    export_write_operation,
+    exported,
+    exported_as_webservice_entry,
+    operation_for_version,
+    operation_returns_collection_of,
+    REQUEST_USER,
+    )
 from lazr.restful.fields import Reference
 from zope.interface import Interface
 from zope.schema import (
@@ -22,54 +33,67 @@ from zope.schema import (
     )
 
 from lp import _
-from lp.code.interfaces.gitrepository import IGitRepository
 from lp.services.auth.enums import AccessTokenScope
 from lp.services.fields import PublicPersonChoice
+from lp.services.webservice.apihelpers import patch_reference_property
 
 
+# XXX cjwatson 2021-10-13 bug=760849: "beta" is a lie to get WADL
+# generation working.  Individual attributes must set their version to
+# "devel".
+@exported_as_webservice_entry(as_of="beta")
 class IAccessToken(Interface):
     """A personal access token for the webservice API."""
 
     id = Int(title=_("ID"), required=True, readonly=True)
 
-    date_created = Datetime(
-        title=_("When the token was created."), required=True, readonly=True)
+    date_created = exported(Datetime(
+        title=_("When the token was created."), required=True, readonly=True))
 
-    owner = PublicPersonChoice(
+    owner = exported(PublicPersonChoice(
         title=_("The person who created the token."),
-        vocabulary="ValidPersonOrTeam", required=True, readonly=True)
+        vocabulary="ValidPersonOrTeam", required=True, readonly=True))
 
-    description = TextLine(
-        title=_("A short description of the token."), required=True)
+    description = exported(TextLine(
+        title=_("A short description of the token."), required=True))
 
     git_repository = Reference(
         title=_("The Git repository for which the token was issued."),
-        schema=IGitRepository, required=True, readonly=True)
+        # Really IGitRepository, patched in _schema_circular_imports.py.
+        schema=Interface, required=True, readonly=True)
+
+    target = exported(Reference(
+        title=_("The target for which the token was issued."),
+        # Really IAccessTokenTarget, patched in _schema_circular_imports.py.
+        schema=Interface, required=True, readonly=True))
 
-    scopes = List(
+    scopes = exported(List(
         value_type=Choice(vocabulary=AccessTokenScope),
         title=_("A list of scopes granted by the token."),
-        required=True, readonly=True)
+        required=True, readonly=True))
 
-    date_last_used = Datetime(
+    date_last_used = exported(Datetime(
         title=_("When the token was last used."),
-        required=False, readonly=False)
+        required=False, readonly=True))
 
-    date_expires = Datetime(
+    date_expires = exported(Datetime(
         title=_("When the token should expire or was revoked."),
-        required=False, readonly=False)
+        required=False, readonly=True))
 
     is_expired = Bool(
         title=_("Whether this token has expired."),
         required=False, readonly=True)
 
-    revoked_by = PublicPersonChoice(
+    revoked_by = exported(PublicPersonChoice(
         title=_("The person who revoked the token, if any."),
-        vocabulary="ValidPersonOrTeam", required=False, readonly=False)
+        vocabulary="ValidPersonOrTeam", required=False, readonly=True))
 
     def updateLastUsed():
         """Update this token's last-used date, if possible."""
 
+    @call_with(revoked_by=REQUEST_USER)
+    @export_write_operation()
+    @operation_for_version("devel")
     def revoke(revoked_by):
         """Revoke this token."""
 
@@ -77,7 +101,7 @@ class IAccessToken(Interface):
 class IAccessTokenSet(Interface):
     """The set of all personal access tokens."""
 
-    def new(secret, owner, description, target, scopes):
+    def new(secret, owner, description, target, scopes, date_expires=None):
         """Return a new access token with a given secret.
 
         :param secret: A text string.
@@ -87,6 +111,8 @@ class IAccessTokenSet(Interface):
             issued.
         :param scopes: A list of `AccessTokenScope`s to be granted by the
             token.
+        :param date_expires: The time when this token should expire, or
+            None.
         """
 
     def getBySecret(secret):
@@ -101,12 +127,32 @@ class IAccessTokenSet(Interface):
         :param owner: An `IPerson`.
         """
 
-    def findByTarget(target):
+    def findByTarget(target, visible_by_user=None):
         """Return all access tokens for this target.
 
-        :param target: An `IGitRepository`.
+        :param target: An `IAccessTokenTarget`.
+        :param visible_by_user: If given, return only access tokens visible
+            by this user.
         """
 
 
 class IAccessTokenVerifiedRequest(Interface):
     """Marker interface for a request with a verified access token."""
+
+
+# XXX cjwatson 2021-10-13 bug=760849: "beta" is a lie to get WADL
+# generation working.  Individual attributes must set their version to
+# "devel".
+@exported_as_webservice_entry(as_of="beta")
+class IAccessTokenTarget(Interface):
+    """An object that can be a target for access tokens."""
+
+    @call_with(visible_by_user=REQUEST_USER)
+    @operation_returns_collection_of(IAccessToken)
+    @export_read_operation()
+    @operation_for_version("devel")
+    def getAccessTokens(visible_by_user=None):
+        """Return personal access tokens for this target."""
+
+
+patch_reference_property(IAccessToken, "target", IAccessTokenTarget)
diff --git a/lib/lp/services/auth/model.py b/lib/lp/services/auth/model.py
index 437be7e..39a6684 100644
--- a/lib/lp/services/auth/model.py
+++ b/lib/lp/services/auth/model.py
@@ -6,6 +6,7 @@
 __metaclass__ = type
 __all__ = [
     "AccessToken",
+    "AccessTokenTargetMixin",
     ]
 
 from datetime import (
@@ -20,6 +21,7 @@ from storm.expr import (
     And,
     Cast,
     Or,
+    Select,
     SQL,
     Update,
     )
@@ -29,9 +31,13 @@ from storm.locals import (
     Reference,
     Unicode,
     )
+from zope.component import getUtility
 from zope.interface import implementer
+from zope.security.proxy import removeSecurityProxy
 
+from lp.code.interfaces.gitcollection import IAllGitRepositories
 from lp.code.interfaces.gitrepository import IGitRepository
+from lp.registry.model.teammembership import TeamParticipation
 from lp.services.auth.enums import AccessTokenScope
 from lp.services.auth.interfaces import (
     IAccessToken,
@@ -78,7 +84,8 @@ class AccessToken(StormBase):
 
     resolution = timedelta(minutes=10)
 
-    def __init__(self, secret, owner, description, target, scopes):
+    def __init__(self, secret, owner, description, target, scopes,
+                 date_expires=None):
         """Construct an `AccessToken`."""
         self._token_sha256 = hashlib.sha256(secret.encode()).hexdigest()
         self.owner = owner
@@ -89,6 +96,7 @@ class AccessToken(StormBase):
             raise TypeError("Unsupported target: {!r}".format(target))
         self.scopes = scopes
         self.date_created = UTC_NOW
+        self.date_expires = date_expires
 
     @property
     def target(self):
@@ -109,7 +117,8 @@ class AccessToken(StormBase):
 
     def updateLastUsed(self):
         """See `IAccessToken`."""
-        IMasterStore(AccessToken).execute(Update(
+        store = IMasterStore(AccessToken)
+        store.execute(Update(
             {AccessToken.date_last_used: UTC_NOW},
             where=And(
                 # Skip the update if the AccessToken row is already locked,
@@ -124,6 +133,7 @@ class AccessToken(StormBase):
                     AccessToken.date_last_used <
                         UTC_NOW - Cast(self.resolution, 'interval'))),
             table=AccessToken))
+        store.invalidate(self)
 
     @property
     def is_expired(self):
@@ -139,10 +149,13 @@ class AccessToken(StormBase):
 @implementer(IAccessTokenSet)
 class AccessTokenSet:
 
-    def new(self, secret, owner, description, target, scopes):
+    def new(self, secret, owner, description, target, scopes,
+            date_expires=None):
         """See `IAccessTokenSet`."""
         store = IStore(AccessToken)
-        token = AccessToken(secret, owner, description, target, scopes)
+        token = AccessToken(
+            secret, owner, description, target, scopes,
+            date_expires=date_expires)
         store.add(token)
         return token
 
@@ -156,11 +169,30 @@ class AccessTokenSet:
         """See `IAccessTokenSet`."""
         return IStore(AccessToken).find(AccessToken, owner=owner)
 
-    def findByTarget(self, target):
+    def findByTarget(self, target, visible_by_user=None):
         """See `IAccessTokenSet`."""
-        kwargs = {}
+        clauses = []
         if IGitRepository.providedBy(target):
-            kwargs["git_repository"] = target
+            clauses.append(AccessToken.git_repository == target)
+            if visible_by_user is not None:
+                collection = getUtility(IAllGitRepositories).visibleByUser(
+                    visible_by_user).ownedByTeamMember(visible_by_user)
+                ids = collection.getRepositoryIds()
+                clauses.append(Or(
+                    AccessToken.owner_id.is_in(Select(
+                        TeamParticipation.teamID,
+                        where=TeamParticipation.person == visible_by_user.id)),
+                    AccessToken.git_repository_id.is_in(
+                        removeSecurityProxy(ids)._get_select())))
         else:
             raise TypeError("Unsupported target: {!r}".format(target))
-        return IStore(AccessToken).find(AccessToken, **kwargs)
+        return IStore(AccessToken).find(AccessToken, *clauses)
+
+
+class AccessTokenTargetMixin:
+    """Mix this into classes that implement `IAccessTokenTarget`."""
+
+    def getAccessTokens(self, visible_by_user=None):
+        """See `IAccessTokenTarget`."""
+        return getUtility(IAccessTokenSet).findByTarget(
+            self, visible_by_user=visible_by_user)
diff --git a/lib/lp/services/auth/tests/test_model.py b/lib/lp/services/auth/tests/test_model.py
index f97d498..c64dbd9 100644
--- a/lib/lp/services/auth/tests/test_model.py
+++ b/lib/lp/services/auth/tests/test_model.py
@@ -31,12 +31,18 @@ from lp.services.database.sqlbase import (
     get_transaction_timestamp,
     )
 from lp.services.webapp.authorization import check_permission
+from lp.services.webapp.interfaces import OAuthPermission
 from lp.testing import (
+    api_url,
     login,
     login_person,
+    person_logged_in,
+    record_two_runs,
     TestCaseWithFactory,
     )
 from lp.testing.layers import DatabaseFunctionalLayer
+from lp.testing.matchers import HasQueryCount
+from lp.testing.pages import webservice_for_person
 
 
 class TestAccessToken(TestCaseWithFactory):
@@ -79,7 +85,7 @@ class TestAccessToken(TestCaseWithFactory):
         _, token = self.factory.makeAccessToken(owner=owner)
         login_person(owner)
         recent = datetime.now(pytz.UTC) - timedelta(minutes=1)
-        token.date_last_used = recent
+        removeSecurityProxy(token).date_last_used = recent
         transaction.commit()
         token.updateLastUsed()
         self.assertEqual(recent, token.date_last_used)
@@ -91,7 +97,7 @@ class TestAccessToken(TestCaseWithFactory):
         _, token = self.factory.makeAccessToken(owner=owner)
         login_person(owner)
         recent = datetime.now(pytz.UTC) - timedelta(hours=1)
-        token.date_last_used = recent
+        removeSecurityProxy(token).date_last_used = recent
         transaction.commit()
         token.updateLastUsed()
         now = get_transaction_timestamp(Store.of(token))
@@ -145,9 +151,9 @@ class TestAccessToken(TestCaseWithFactory):
         owner = self.factory.makePerson()
         login_person(owner)
         _, current_token = self.factory.makeAccessToken(owner=owner)
-        _, expired_token = self.factory.makeAccessToken(owner=owner)
-        expired_token.date_expires = (
-            datetime.now(pytz.UTC) - timedelta(minutes=1))
+        _, expired_token = self.factory.makeAccessToken(
+            owner=owner,
+            date_expires=datetime.now(pytz.UTC) - timedelta(minutes=1))
         self.assertFalse(current_token.is_expired)
         self.assertTrue(expired_token.is_expired)
 
@@ -219,3 +225,84 @@ class TestAccessTokenSet(TestCaseWithFactory):
             [tokens[2]], getUtility(IAccessTokenSet).findByTarget(targets[1]))
         self.assertContentEqual(
             [], getUtility(IAccessTokenSet).findByTarget(targets[2]))
+
+    def test_findByTarget_visible_by_user(self):
+        targets = [self.factory.makeGitRepository() for _ in range(3)]
+        owners = [self.factory.makePerson() for _ in range(3)]
+        tokens = [
+            self.factory.makeAccessToken(
+                owner=owners[owner_index], target=targets[target_index])[1]
+            for owner_index, target_index in (
+                (0, 0), (0, 0), (1, 0), (1, 1), (2, 1))]
+        for owner_index, target_index, expected_tokens in (
+                (0, 0, tokens[:2]),
+                (0, 1, []),
+                (0, 2, []),
+                (1, 0, [tokens[2]]),
+                (1, 1, [tokens[3]]),
+                (1, 2, []),
+                (2, 0, []),
+                (2, 1, [tokens[4]]),
+                (2, 2, []),
+                ):
+            self.assertContentEqual(
+                expected_tokens,
+                getUtility(IAccessTokenSet).findByTarget(
+                    targets[target_index],
+                    visible_by_user=owners[owner_index]))
+
+
+class TestAccessTokenTargetBase:
+
+    layer = DatabaseFunctionalLayer
+
+    def setUp(self):
+        super().setUp()
+        self.target = self.makeTarget()
+        self.owner = self.target.owner
+        self.target_url = api_url(self.target)
+        self.webservice = webservice_for_person(
+            self.owner, permission=OAuthPermission.WRITE_PRIVATE)
+
+    def test_getAccessTokens(self):
+        with person_logged_in(self.owner):
+            for description in ("Test token 1", "Test token 2"):
+                self.factory.makeAccessToken(
+                    owner=self.owner, description=description,
+                    target=self.target)
+        response = self.webservice.named_get(
+            self.target_url, "getAccessTokens", api_version="devel")
+        self.assertEqual(200, response.status)
+        self.assertContentEqual(
+            ["Test token 1", "Test token 2"],
+            [entry["description"] for entry in response.jsonBody()["entries"]])
+
+    def test_getAccessTokens_permissions(self):
+        webservice = webservice_for_person(None)
+        response = webservice.named_get(
+            self.target_url, "getAccessTokens", api_version="devel")
+        self.assertEqual(401, response.status)
+        self.assertIn(b"launchpad.Edit", response.body)
+
+    def test_getAccessTokens_query_count(self):
+        def get_tokens():
+            response = self.webservice.named_get(
+                self.target_url, "getAccessTokens", api_version="devel")
+            self.assertEqual(200, response.status)
+            self.assertIn(len(response.jsonBody()["entries"]), {0, 2, 4})
+
+        def create_token():
+            with person_logged_in(self.owner):
+                self.factory.makeAccessToken(
+                    owner=self.owner, target=self.target)
+
+        get_tokens()
+        recorder1, recorder2 = record_two_runs(get_tokens, create_token, 2)
+        self.assertThat(recorder2, HasQueryCount.byEquality(recorder1))
+
+
+class TestAccessTokenTargetGitRepository(
+        TestAccessTokenTargetBase, TestCaseWithFactory):
+
+    def makeTarget(self):
+        return self.factory.makeGitRepository()
diff --git a/lib/lp/services/auth/webservice.py b/lib/lp/services/auth/webservice.py
 new file mode 100644
index 0000000..08e7ca1
--- /dev/null
+++ b/lib/lp/services/auth/webservice.py
@@ -0,0 +1,14 @@
+# Copyright 2021 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Personal access token webservice registrations."""
+
+__all__ = [
+    "IAccessToken",
+    "IAccessTokenTarget",
+    ]
+
+from lp.services.auth.interfaces import (
+    IAccessToken,
+    IAccessTokenTarget,
+    )
diff --git a/lib/lp/services/webapp/tests/test_servers.py b/lib/lp/services/webapp/tests/test_servers.py
index aa5123f..a5ffe0a 100644
--- a/lib/lp/services/webapp/tests/test_servers.py
+++ b/lib/lp/services/webapp/tests/test_servers.py
@@ -68,7 +68,6 @@ from lp.services.webapp.servers import (
 from lp.testing import (
     EventRecorder,
     logout,
-    person_logged_in,
     TestCase,
     TestCaseWithFactory,
     )
@@ -845,9 +844,9 @@ class TestWebServiceAccessTokens(TestCaseWithFactory):
 
     def test_expired(self):
         owner = self.factory.makePerson()
-        secret, token = self.factory.makeAccessToken(owner=owner)
-        with person_logged_in(owner):
-            token.date_expires = datetime.now(pytz.UTC) - timedelta(days=1)
+        secret, token = self.factory.makeAccessToken(
+            owner=owner,
+            date_expires=datetime.now(pytz.UTC) - timedelta(days=1))
         transaction.commit()
 
         request, publication = get_request_and_publication(
diff --git a/lib/lp/services/webservice/wadl-to-refhtml.xsl b/lib/lp/services/webservice/wadl-to-refhtml.xsl
index 96efcf6..dc13955 100644
--- a/lib/lp/services/webservice/wadl-to-refhtml.xsl
+++ b/lib/lp/services/webservice/wadl-to-refhtml.xsl
@@ -166,7 +166,8 @@
             <xsl:with-param name="url">
                 <xsl:choose>
                     <xsl:when test="
-                        @id = 'bug_link_target'
+                        @id = 'access_token_target'
+                        or @id = 'bug_link_target'
                         or @id = 'bug_target'
                         or @id = 'faq_target'
                         or @id = 'git_target'
@@ -190,6 +191,10 @@
     <xsl:template name="find-entry-uri">
         <xsl:value-of select="$base"/>
         <xsl:choose>
+            <xsl:when test="@id = 'access_token'">
+                <xsl:text>/[target URL]/+access-token/</xsl:text>
+                <var>&lt;id&gt;</var>
+            </xsl:when>
             <xsl:when test="@id = 'archive'">
                 <xsl:text>/</xsl:text>
                 <var>&lt;distribution&gt;</var>
diff --git a/lib/lp/testing/factory.py b/lib/lp/testing/factory.py
index 4fccdec..50f2ed4 100644
--- a/lib/lp/testing/factory.py
+++ b/lib/lp/testing/factory.py
@@ -4519,7 +4519,7 @@ class BareLaunchpadObjectFactory(ObjectFactory):
         return request_token.createAccessToken()
 
     def makeAccessToken(self, secret=None, owner=None, description=None,
-                        target=None, scopes=None):
+                        target=None, scopes=None, date_expires=None):
         """Create a personal access token.
 
         :return: A tuple of the secret for the new token and the token
@@ -4536,7 +4536,8 @@ class BareLaunchpadObjectFactory(ObjectFactory):
         if scopes is None:
             scopes = []
         token = getUtility(IAccessTokenSet).new(
-            secret, owner, description, target, scopes)
+            secret, owner, description, target, scopes,
+            date_expires=date_expires)
         return secret, token
 
     def makeCVE(self, sequence, description=None,