Allow admin revocation of API Keys

fence/blueprints/storage_creds/api.py

@@ -4,7 +4,7 @@
 from flask_restful import Resource
 
 from fence.auth import require_auth_header, current_token
-from fence.authz.auth import can_user_get_task_token
+from fence.authz.auth import authorize as authz_authorize, can_user_get_task_token
 from fence.errors import Forbidden, UserError
 from fence.jwt.blacklist import blacklist_token
 from fence.jwt.dpop import jti_seen
@@ -127,9 +127,10 @@ class ApiKey(Resource):
     @require_auth_header({"credentials"})
     def delete(self, access_key):
         """
-        Delete a key for user
+        Delete a key for user, or for any user if the caller is an admin
 
-        :param access_key: existing access key that belongs to this user
+        :param access_key: existing access key that belongs to this user, or to
+            any user if the caller is an admin
 
         :statuscode 204 Success
         :statuscode 403 Forbidden to delete access key
@@ -139,13 +140,18 @@ def delete(self, access_key):
 
         jti = access_key
         with flask.current_app.db.session as session:
-            api_key = (
-                session.query(UserRefreshToken)
-                .filter_by(jti=jti, userid=user_id)
-                .first()
-            )
+            api_key = session.query(UserRefreshToken).filter_by(jti=jti).first()
         if not api_key:
-            flask.abort(404, "token not found with JTI {} for current user".format(jti))
+            flask.abort(404, "token not found with JTI {}".format(jti))
+
+        # Check if the admin can delete the token.
+        if str(api_key.userid) != user_id:
+            authz_authorize(
+                "/services/fence/admin",
+                "delete",
+                user_id=current_token["context"]["user"]["name"],
+            )
+
         blacklist_token(jti, api_key.expires)
 
         return "", 204

tests/credentials/api_key/test_api_key.py

@@ -1,3 +1,9 @@
+from unittest import mock
+
+import jwt
+import pytest
+
+from tests import utils
 from tests.utils.api_key import get_api_key, get_api_key_with_json
 
 
@@ -35,3 +41,59 @@ def test_list_api_key(client, oauth_client, encoded_creds_jwt):
     assert response.status_code == 200, response.text
     assert "jtis" in response.json
     assert len(response.json["jtis"]) == n_keys
+
+
+def test_delete_api_key_access(
+    app,
+    client,
+    oauth_client,
+    encoded_creds_jwt,
+    kid,
+    rsa_private_key,
+    mock_arborist_requests,
+):
+    """
+    Test that an admin can delete another user's API Key and that the deleted API Key cannot be used to get an access token.
+    """
+    response = get_api_key(client, encoded_creds_jwt["jwt"])
+    assert response.status_code == 200, response.text
+    key_id = response.json["key_id"]
+    api_key = response.json["api_key"]
+
+    mock_arborist_requests({"arborist/auth/request": {"POST": ({"auth": True}, 200)}})
+    revoker_jwt = jwt.encode(
+        utils.authorized_download_credentials_context_claims(
+            "other-user", 80, oauth_client["client_id"]
+        ),
+        key=rsa_private_key,
+        headers={"kid": kid},
+        algorithm="RS256",
+    )
+
+    with mock.patch.object(
+        app.arborist, "auth_request", wraps=app.arborist.auth_request
+    ) as auth_request:
+        response = client.delete(
+            f"/credentials/api/{key_id}",
+            headers={"Authorization": f"Bearer {revoker_jwt}"},
+        )
+    assert response.status_code == 204
+
+    auth_request.assert_called_once()
+    assert auth_request.call_args.kwargs["user_id"] == "other-user"
+    assert not auth_request.call_args.kwargs["jwt"]
+
+    response = client.post("/credentials/api/access_token", data={"api_key": api_key})
+    assert response.status_code == 401
+
+
+def test_delete_api_key_not_found(client, encoded_creds_jwt, mock_arborist_requests):
+    """
+    Test ``DELETE /credentials/api/<jti>`` with a JTI that matches no API key
+    """
+    mock_arborist_requests()
+    response = client.delete(
+        "/credentials/api/514a2ccb-8e2e-416b-ad7c-4c52bc03e8e5",
+        headers={"Authorization": "Bearer " + str(encoded_creds_jwt["jwt"])},
+    )
+    assert response.status_code == 404, response.text