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: