launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #27592
[Merge] ~cjwatson/launchpad:access-token-api into launchpad:master
Colin Watson has proposed merging ~cjwatson/launchpad:access-token-api into launchpad:master.
Commit message:
Add webservice API for personal access tokens
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~cjwatson/launchpad/+git/launchpad/+merge/410163
This currently supports issuing, querying, and revoking personal access tokens for Git repositories.
--
Your team Launchpad code reviewers is requested to review the proposed merge of ~cjwatson/launchpad:access-token-api into launchpad:master.
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..dabd794 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=False))
- date_expires = Datetime(
+ date_expires = exported(Datetime(
title=_("When the token should expire or was revoked."),
- required=False, readonly=False)
+ required=False, readonly=False))
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=False))
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..0b84f70 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):
@@ -139,10 +147,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 +167,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..32dcf28 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):
@@ -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/webservice/wadl-to-refhtml.xsl b/lib/lp/services/webservice/wadl-to-refhtml.xsl
index 96efcf6..03ca4a8 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><id></var>
+ </xsl:when>
<xsl:when test="@id = 'archive'">
<xsl:text>/</xsl:text>
<var><distribution></var>
