Method put

st2api/st2api/controllers/v1/keyvalue.py:320–415

Create a new entry or update an existing one.

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

        return kvp_apis_system + kvp_apis_user

    def put(self, kvp, name, requester_user, scope=None):
        """
        Create a new entry or update an existing one.
        """
        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 = getattr(kvp, "scope", scope)
        scope = get_datastore_full_scope(scope)
        self._validate_scope(scope=scope)

        user = getattr(kvp, "user", requester_user.name) 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("PUT /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 set 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_SET,
        )

        # Validate that the pre-encrypted option can only be used by admins
        encrypted = getattr(kvp, "encrypted", False)
        self._validate_encrypted_query_parameter(
            encrypted=encrypted, scope=scope, requester_user=requester_user
        )

        # 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)
        ):
            try: