From 5fe44a904ce0468dc6e8e2f8feb7484334a20d30 Mon Sep 17 00:00:00 2001 From: Quanah Gibson-Mount Date: Wed, 14 Oct 2026 01:37:47 +0000 Subject: [PATCH] Let a superuser issue a personal token for another user POST /api/v2/users//personal_tokens/ always created a token for the requester, whatever was, and returned 201. GET on the same URL listed the requester's tokens. A user created without a password can't log in to create its own token, so the only way to give one a token was awx-manage create_oauth2_token on the controller. The token now belongs to . Users can still create tokens for themselves; only a superuser may create one for someone else, and anyone else gets 403. This matches OAuth2TokenAccess, which already limits reading and revoking personal tokens to the owner and superusers. Because the sub-list permission check asks can_add about the user in the URL, non-superusers also get 403 from another user's tokens/ and authorized_tokens/; superusers there still get their own token, as before. A token for someone else is refused (400) when the owner is inactive, or is a local user while DISABLE_LOCAL_AUTH is on, since turning that setting on revokes those users' tokens. The external-account check (ALLOW_OAUTH2_FOR_EXTERNAL_USERS) applies to the owner. GET lists 's personal tokens, filtered by what the requester may read. The activity-stream entry for a token whose owner isn't the actor records the owner in changes.user and links the entry to the owner's activity stream; the actor is still the requester. related #1116 Assisted-by: Claude Opus 5.5 Signed-off-by: Quanah Gibson-Mount --- ascender/api/serializers/oauth2.py | 5 +- ascender/api/serializers/user.py | 16 +++ ascender/api/views/user.py | 3 +- ascender/main/access.py | 8 +- ascender/main/signals.py | 10 +- .../main/tests/functional/api/test_oauth.py | 129 ++++++++++++++++++ docs/auth/oauth.md | 7 +- .../rst/administration/oauth2_token_auth.rst | 2 + 8 files changed, 175 insertions(+), 5 deletions(-) diff --git a/ascender/api/serializers/oauth2.py b/ascender/api/serializers/oauth2.py index 62f487eba..58c25cf34 100644 --- a/ascender/api/serializers/oauth2.py +++ b/ascender/api/serializers/oauth2.py @@ -82,8 +82,11 @@ def validate_scope(self, value): raise serializers.ValidationError(_('Must be a simple space-separated string with allowed scopes {}.').format(self.ALLOWED_SCOPES)) return value + def get_owner(self, validated_data): + return self.context['request'].user + def create(self, validated_data): - validated_data['user'] = self.context['request'].user + validated_data['user'] = self.get_owner(validated_data) try: return super(BaseOAuth2TokenSerializer, self).create(validated_data) except oauth2.AccessDeniedError as e: diff --git a/ascender/api/serializers/user.py b/ascender/api/serializers/user.py index 6b3e62342..f18aaeebb 100644 --- a/ascender/api/serializers/user.py +++ b/ascender/api/serializers/user.py @@ -280,6 +280,22 @@ class UserPersonalTokenSerializer(BaseOAuth2TokenSerializer): class Meta: read_only_fields = ('user', 'token', 'expires', 'application') + def validate(self, attrs): + attrs = super(UserPersonalTokenSerializer, self).validate(attrs) + # owned by the user in the URL; OAuth2TokenAccess.can_add decides whether the requester may issue it + owner = self.context['view'].get_parent_object() + if owner != self.context['request'].user: + if not owner.is_active: + raise serializers.ValidationError(_('Cannot create a token for an inactive user.')) + # local as DisableLocalAuthMiddleware decides it + if settings.DISABLE_LOCAL_AUTH and not (owner.profile.ldap_dn or owner.social_auth.exists() or owner.enterprise_auth.exists()): + raise serializers.ValidationError(_('Cannot create a token for a local user while local authentication is disabled.')) + attrs['user'] = owner + return attrs + + def get_owner(self, validated_data): + return validated_data['user'] + def create(self, validated_data): validated_data['token'] = generate_token() validated_data['expires'] = now() + timedelta(seconds=settings.OAUTH2_PROVIDER['ACCESS_TOKEN_EXPIRE_SECONDS']) diff --git a/ascender/api/views/user.py b/ascender/api/views/user.py index 3c02ae84f..1c953c166 100644 --- a/ascender/api/views/user.py +++ b/ascender/api/views/user.py @@ -86,7 +86,8 @@ class UserPersonalTokenList(SubListCreateAPIView): swagger_topic = 'Authentication' def get_queryset(self): - return get_access_token_model().objects.filter(application__isnull=True, user=self.request.user) + # the URL user's tokens, limited to those the requester may read (OAuth2TokenAccess) + return super(UserPersonalTokenList, self).get_queryset().filter(application__isnull=True) class UserTeamsList(SubListAPIView): diff --git a/ascender/main/access.py b/ascender/main/access.py index b21a3ac7c..4fc440f55 100644 --- a/ascender/main/access.py +++ b/ascender/main/access.py @@ -791,7 +791,8 @@ class OAuth2TokenAccess(BaseAccess): - I am the user of the token - I am the superuser I can create an OAuth2 Personal Access Token when: - - I am a user. But I can only create a PAT for myself. + - I am a user, for myself. + - I am a superuser, for any user. """ model = OAuth2AccessToken @@ -814,6 +815,11 @@ def can_change(self, obj, data): return self.can_delete(obj) def can_add(self, data): + if 'user' in data: + # a token for someone else (the user in a users// token URL) is for superusers only + owner = get_object_from_data('user', User, data) + if owner is not None and owner != self.user and not self.user.is_superuser: + return False if 'application' in data: app = get_object_from_data('application', OAuth2Application, data) if app is None: diff --git a/ascender/main/signals.py b/ascender/main/signals.py index 08a59a8b2..68fbf546b 100644 --- a/ascender/main/signals.py +++ b/ascender/main/signals.py @@ -395,15 +395,23 @@ def activity_stream_create(sender, instance, created, **kwargs): changes['labels'] = [label.name for label in instance.labels.iterator()] if 'extra_vars' in changes: changes['extra_vars'] = instance.display_extra_vars() + actor = get_current_user_or_none() + # a token whose owner isn't the actor (issued by an admin, or with no request user) names the owner + token_owner = None if type(instance) == OAuth2AccessToken: changes['token'] = CENSOR_VALUE - activity_entry = get_activity_stream_class()(operation='create', object1=object1, changes=json.dumps(changes), actor=get_current_user_or_none()) + if instance.user_id and (actor is None or instance.user_id != actor.pk): + token_owner = instance.user + changes['user'] = '{} ({})'.format(token_owner.username, token_owner.pk) + activity_entry = get_activity_stream_class()(operation='create', object1=object1, changes=json.dumps(changes), actor=actor) # TODO: Weird situation where cascade SETNULL doesn't work # it might actually be a good idea to remove all of these FK references since # we don't really use them anyway. if instance._meta.model_name != 'setting': # Is not conf.Setting instance activity_entry.save() getattr(activity_entry, object1).add(instance.pk) + if token_owner is not None: + activity_entry.user.add(token_owner) else: activity_entry.setting = conf_to_dict(instance) activity_entry.save() diff --git a/ascender/main/tests/functional/api/test_oauth.py b/ascender/main/tests/functional/api/test_oauth.py index 77cf82a8e..84d141d27 100644 --- a/ascender/main/tests/functional/api/test_oauth.py +++ b/ascender/main/tests/functional/api/test_oauth.py @@ -11,7 +11,9 @@ from ascender.main.utils.encryption import decrypt_value, get_encryption_key from ascender.api.versioning import reverse, drf_reverse +from ascender.main.models import ActivityStream from ascender.main.models.oauth import OAuth2Application as Application, OAuth2AccessToken as AccessToken +from ascender.main.request_context import impersonate from ascender.main.tests.functional import immediate_on_commit from ascender.sso.models import UserEnterpriseAuth from ascender.main.models.oauth import OAuth2RefreshToken @@ -390,3 +392,130 @@ def test_token_values_are_not_rendered_by_str(oauth_application, post, admin): for obj in (access_token, refresh_token): assert obj.token not in str(obj) assert obj.token not in repr(obj) + + +@pytest.mark.django_db +def test_superuser_issues_personal_token_for_another_user(post, get, admin, alice): + response = post(reverse('api:user_personal_token_list', kwargs={'pk': alice.pk}), {'description': 'ci', 'scope': 'read'}, admin, expect=201) + token = AccessToken.objects.get(token=response.data['token']) + assert token.user == alice + assert token.application is None + assert response.data['user'] == alice.pk + with immediate_on_commit(): + resp = get(drf_reverse('api:user_me_list', kwargs={'version': 'v2'}), HTTP_AUTHORIZATION='Bearer ' + token.token, status=200) + assert json.loads(resp.content)['results'][0]['username'] == 'alice' + + +@pytest.mark.django_db +@pytest.mark.parametrize('requester', ['alice', 'org_admin', 'system_auditor']) +def test_personal_token_for_another_user_needs_superuser(post, request, org_member, requester): + # org_admin may change org_member (UserAccess), but personal tokens stay owner-or-superuser (OAuth2TokenAccess) + user = request.getfixturevalue(requester) + post(reverse('api:user_personal_token_list', kwargs={'pk': org_member.pk}), {'scope': 'read'}, user, expect=403) + assert AccessToken.objects.count() == 0 + + +@pytest.mark.django_db +def test_personal_token_for_missing_user(post, admin): + post(reverse('api:user_personal_token_list', kwargs={'pk': 9999}), {'scope': 'read'}, admin, expect=404) + assert AccessToken.objects.count() == 0 + + +@pytest.mark.django_db +def test_personal_token_for_another_external_user(post, admin, alice): + # the external-account check applies to the owner, not to the superuser issuing it + UserEnterpriseAuth(user=alice, provider='radius').save() + with override_settings(RADIUS_SERVER='example.org', ALLOW_OAUTH2_FOR_EXTERNAL_USERS=False): + response = post(reverse('api:user_personal_token_list', kwargs={'pk': alice.pk}), {'scope': 'read'}, admin, expect=403) + assert 'external authentication provider' in str(response.data['detail']) + assert AccessToken.objects.count() == 0 + + +@pytest.mark.django_db +def test_personal_token_for_another_inactive_user(post, admin, alice): + alice.is_active = False + alice.save() + response = post(reverse('api:user_personal_token_list', kwargs={'pk': alice.pk}), {'scope': 'read'}, admin, expect=400) + assert 'inactive user' in str(response.data) + assert AccessToken.objects.count() == 0 + + +@pytest.mark.django_db +@pytest.mark.parametrize('ldap_dn, status', [('', 400), ('cn=alice,dc=example,dc=org', 201)]) +def test_personal_token_for_another_user_with_local_auth_disabled(post, admin, alice, ldap_dn, status): + alice.profile.ldap_dn = ldap_dn + alice.profile.save() + with override_settings(DISABLE_LOCAL_AUTH=True): + response = post(reverse('api:user_personal_token_list', kwargs={'pk': alice.pk}), {'scope': 'read'}, admin, expect=status) + if status == 400: + assert 'local authentication is disabled' in str(response.data) + assert AccessToken.objects.filter(user=alice).count() == (1 if status == 201 else 0) + + +@pytest.mark.django_db +@pytest.mark.parametrize('requester', ['admin', 'alice']) +def test_personal_token_for_self(post, request, requester): + user = request.getfixturevalue(requester) + for url in (reverse('api:user_personal_token_list', kwargs={'pk': user.pk}), reverse('api:o_auth2_token_list')): + response = post(url, {'scope': 'read'}, user, expect=201) + token = AccessToken.objects.get(token=response.data['token']) + assert token.user == user + assert token.application is None + + +@pytest.mark.django_db +def test_other_token_sub_urls_still_issue_to_the_requester(post, admin, org_admin, org_member, organization): + # users//tokens/ and authorized_tokens/ share the permission check, but keep issuing to the requester + app = Application.objects.create(name='app', client_type='confidential', authorization_grant_type='password', organization=organization) + urls = [ + (reverse('api:o_auth2_token_list', kwargs={'pk': org_member.pk}), {'scope': 'read'}), + (reverse('api:user_authorized_token_list', kwargs={'pk': org_member.pk}), {'scope': 'read', 'application': app.pk}), + ] + for url, data in urls: + post(url, data, org_admin, expect=403) + response = post(url, data, admin, expect=201) + assert AccessToken.objects.get(token=response.data['token']).user == admin + assert not AccessToken.objects.filter(user=org_member).exists() + + +@pytest.mark.django_db +def test_personal_token_list_is_the_url_users(oauth_application, post, get, admin, alice, bob, system_auditor): + for user in (admin, alice): + post(reverse('api:user_personal_token_list', kwargs={'pk': user.pk}), {'scope': 'read'}, user, expect=201) + # an application token is not a personal one + post(reverse('api:o_auth2_application_token_list', kwargs={'pk': oauth_application.pk}), {'scope': 'read'}, admin, expect=201) + + def owners(owner, requester): + response = get(reverse('api:user_personal_token_list', kwargs={'pk': owner.pk}), requester, expect=200) + return [t['user'] for t in response.data['results']] + + assert owners(alice, admin) == [alice.pk] + assert owners(alice, alice) == [alice.pk] + assert owners(alice, system_auditor) == [alice.pk] + assert owners(admin, admin) == [admin.pk] + # alice can see the admin user, but not the admin's tokens + assert owners(admin, alice) == [] + # bob can't see alice at all + get(reverse('api:user_personal_token_list', kwargs={'pk': alice.pk}), bob, expect=403) + + +@pytest.mark.django_db +def test_personal_token_for_another_user_activity_stream(post, get, admin, alice, settings): + settings.ACTIVITY_STREAM_ENABLED = True + # the request middleware sets the actor in production; the test client calls the view directly + with impersonate(admin): + response = post(reverse('api:user_personal_token_list', kwargs={'pk': alice.pk}), {'scope': 'read'}, admin, expect=201) + own = post(reverse('api:user_personal_token_list', kwargs={'pk': admin.pk}), {'scope': 'read'}, admin, expect=201) + + entry = ActivityStream.objects.get(operation='create', o_auth2_access_token__token=response.data['token']) + assert entry.actor == admin + assert json.loads(entry.changes)['user'] == 'alice ({})'.format(alice.pk) + assert list(entry.user.all()) == [alice] + alice_stream = get(reverse('api:user_activity_stream_list', kwargs={'pk': alice.pk}), admin, expect=200) + assert entry.pk in [e['id'] for e in alice_stream.data['results']] + + # a token the actor issues for themselves records nothing extra + own_entry = ActivityStream.objects.get(operation='create', o_auth2_access_token__token=own.data['token']) + assert own_entry.actor == admin + assert 'user' not in json.loads(own_entry.changes) + assert not own_entry.user.exists() diff --git a/docs/auth/oauth.md b/docs/auth/oauth.md index 706a7bb68..a6e1c20c1 100644 --- a/docs/auth/oauth.md +++ b/docs/auth/oauth.md @@ -188,8 +188,13 @@ token can be obtained by doing a curl to the `/api/o/token/` endpoint. For examp ``` curl -ku : -H "Content-Type: application/json" -X POST \ -d '{"description":"Tower CLI", "application":null, "scope":"write"}' \ -https:///api/v2/users/1/personal_tokens/ | python -m json.tool +https:///api/v2/users//personal_tokens/ | python -m json.tool ``` + +`` is the token's owner. Users create tokens for themselves; a System Administrator may also +create one for another user, such as a service account that has no password. The activity stream +entry for such a token names both the administrator (`actor`) and the owner (`changes.user`). + Here is an example of using that PAT to access an API endpoint using `curl`: ``` curl -H "Authorization: Bearer kqHqxfpHGRRBXLNCOXxT5Zt3tpJogn" http:///api/v2/credentials/ diff --git a/docs/docsite/rst/administration/oauth2_token_auth.rst b/docs/docsite/rst/administration/oauth2_token_auth.rst index 6c9b0b1f1..969d155f0 100644 --- a/docs/docsite/rst/administration/oauth2_token_auth.rst +++ b/docs/docsite/rst/administration/oauth2_token_auth.rst @@ -154,6 +154,8 @@ The easiest and most common way to obtain an OAuth 2 token is to create a person You could also pipe the JSON output through ``jq``, if installed. +The token belongs to the user whose ID is in the URL. Users can create tokens only for themselves; a System Administrator can also create one for another user, for example a service account created without a password, which cannot log in to create its own. Any other request for someone else's ID is refused with ``403``. ``GET`` on the same endpoint lists that user's personal tokens, as far as the requester is allowed to see them. + Following is an example of using the personal token to access an API endpoint using curl: