Method delete

st2api/st2api/controllers/v1/keyvalue.py:417–492

Delete the key value pair. Handles requests: DELETE /keys/1

(self, name, requester_user, scope=None, user=None)

        return KeyValuePairAPI.from_model(kvp_db)

    def delete(self, name, requester_user, scope=None, user=None):
        """
        Delete the key value pair.

        Handles requests:
            DELETE /keys/1
        """
        if not scope or scope == ALL_SCOPE:
            # Default to system scope
            scope = FULL_SYSTEM_SCOPE

        if not requester_user:
            requester_user = UserDB(name=cfg.CONF.system_user.user)

        scope = get_datastore_full_scope(scope)
        self._validate_scope(scope=scope)

        user = user or requester_user.name

        rbac_utils = get_rbac_backend().get_utils_class()

        # Validate that the authenticated user is admin if user query param is provided
        rbac_utils.assert_user_is_admin_if_user_query_param_is_provided(
            user_db=requester_user, user=user, require_rbac=True
        )

        # Set key reference for system or user scope
        key_ref = get_key_reference(scope=scope, name=name, user=user)
        extra = {"scope": scope, "name": name, "user": user, "key_ref": key_ref}
        LOG.debug("DELETE /v1/keys/%s", name, extra=extra)

        # Setup a kvp database object used for verifying permission
        kvp_db = KeyValuePairDB(
            uid="%s:%s:%s" % (ResourceType.KEY_VALUE_PAIR, scope, key_ref),
            scope=scope,
            name=key_ref,
        )

        # Check that user has permission to the key value pair.
        # If RBAC is enabled, this check will verify if user has system role with all access.
        # If RBAC is enabled, this check guards against a user accessing another user's kvp.
        # If RBAC is enabled, user needs to be explicitly granted permission to delete a system kvp.
        # The check is sufficient to allow decryption of the system kvp.
        rbac_utils.assert_user_has_resource_db_permission(
            user_db=requester_user,
            resource_db=kvp_db,
            permission_type=PermissionType.KEY_VALUE_PAIR_DELETE,
        )

        # Acquire a lock to avoid race condition between concurrent API calls
        with self._coordinator.get_lock(
            self._get_lock_name_for_key(name=key_ref, scope=scope)
        ):
            from_model_kwargs = {"mask_secrets": True}
            kvp_api = self._get_one_by_scope_and_name(
                name=key_ref, scope=scope, from_model_kwargs=from_model_kwargs
            )
            kvp_db = KeyValuePairAPI.to_model(kvp_api)