如何批量删除「CloudFlare KV」中的键对

首先声明，我的长短链使用的是 Url-Shorten-Worker，在很久前，我就好奇 Url-Shorten-Worker 这个 Github Repo 是怎么做到每天清理所有的 key:value 的，也没有认真去了解。

昨天 @Haukeng 疯狂调用我的长链接转短链接服务，为我的长短链注入了 1400+ 数据量，直到他的行为被 CloudFlare ban 掉（否则 1w+ 数据量）。也借着这次计划，我接触到了 CloudFlare 提供的 API，利用这些 API 可以进行批量操作行为。

文档相关

https://developers.cloudflare.com/workers/cli-wrangler/commands#kv

在本次操作中我们只关心上述文档中的 kv:key 的操作方式

可以看到右边的 TOC 很清晰的展示了几个操作：

操作	含义
put	往指定的 `Namespace` 中填充键值对
list	获取全部的 `key`
get	获取某个 `key` 对应的 `value`
delete	删除指定 `key`，及其 `value`

这篇文章的核心思路是：

首先获取所有的 key
查询所有 key 对应的 value
判断对应的 value 是否满足条件
若 value 满足条件，则请求删除对应的 key

利用 Python 自动化

Python 对于写这种小工具来说真的太简单了，其思维也是极其容易理解。

环境

Ubuntu 20.04 WSL, based on Windows 10 20H2
Python 3.8.5
npm 6.14.12
wrangler 1.16.1

理论上，只要求本地安装了 Python >= 3.4，并且通过 npm 安装了 wrangler 即可。
参考 wrangler install，其安装方式为 npm i @cloudflare/wrangler -g

如何自动化操作

在通过 Python 自动化操作中，主要使用到了 Python 原生自带的 subprocess、json 两个库。

subprocess 用于调用系统的 Shell，方便 Python 接入 wrangler API。如果不需要获得输出，或者不需要进行输入，利用 os 库中的 os.system("commands") 也是一样的。subprocess 提供了标准的输入输出数据流，它可以很好的模拟我们需要在控制台交互的行为。
CloudFlare API 通过 list 查询后返回的数据是 JSON 格式的，因此需要使用 JSON 库来对其进行自动规范化，方便后面的数据处理

代码片段

代码主要分为三大部分：查询全部、查询对应的键值，以及删除满足条件的键

获取所有 key

# Query keys withoud values
def get_all_keys(self) -> json:
    keys = subprocess.Popen(f"wrangler kv:key list <YOUR-LOGIN-TOKEN>", shell=True, stdout=subprocess.PIPE, encoding='utf8').stdout
    res = json.load(keys)
    return res

查询对应的键值

def get_values(self):
    for key in keys:
        # Get values with key querying
        value = subprocess.Popen(f"wrangler kv:key get <YOUR-LOGIN-TOKEN> {key['name']}", shell=True, stdout=subprocess.PIPE, encoding='utf8').stdout.read()

删除满足条件的键

if key in BAN_LIST:
    op = subprocess.Popen(f"wrangler kv:key delete <YOUR-LOGIN-TOKEN> {key}", shell=True,   stdin=subprocess.PIPE, stderr=subprocess.PIPE)
    op.stdin.write(bytes('y', encoding='utf8'))
    op.stdin.flush()

完整代码

import subprocess
import json


# Choose one to fill
KV_BINDING =  ''
KV_NAMESPACE_ID = ''

Ban_Keyword = ['https://2to.fun']


def process_bar(percent, start_str='', end_str='', total_length=0):
    bar = ''.join(["\033[31m%s\033[0m"%'   '] * int(percent * total_length)) + ''
    bar = '\r' + start_str + bar.ljust(total_length) + ' {:0>4.1f}%|'.format(percent*100) + end_str
    print(bar, end='', flush=True)


class KV:
    def __init__(self) -> None:
        self.login_token = self.check_login_method()
        self.keys_only = self.get_all_keys()
        self.key_values = []

    def check_login_method(self):
        if KV_BINDING == KV_NAMESPACE_ID == '':
            print("[-] Valid token, please check again")
            exit(-1)
        elif KV_BINDING != '':
            print("[+] Loggin as KV_BINDING")
            return f"-b {KV_BINDING}"
        elif KV_NAMESPACE_ID != '':
            print("[+] Loggin as KV_NAMESPACE_ID")
            return f"-n {KV_NAMESPACE_ID}"
        

    # Get keys withoud values
    def get_all_keys(self) -> json:
        print("[-] Querying KEYS...")
        keys = subprocess.Popen(f"wrangler kv:key list {self.login_token}", shell=True, stdout=subprocess.PIPE, encoding='utf8').stdout
        res = json.load(keys)
        print(f"[+] Query done, total get {len(res)} keys")
        return res

    def run(self):
        print("[-] Paring VALUES...")
        count = 0
        for item_key in self.keys_only:
            count += 1
            key = item_key['name']
            value = subprocess.Popen(f"wrangler kv:key get {self.login_token} {key}", shell=True, stdout=subprocess.PIPE, encoding='utf8').stdout
            value = value.read()

            if value not in Ban_Keyword:
                print(f"[+] Get key {key} - value {value}.")
            else:
                op = subprocess.Popen(f"wrangler kv:key delete {self.login_token} {key}", shell=True, stdin=subprocess.PIPE, stderr=subprocess.PIPE)
                op.stdin.write(bytes('y', encoding='utf8'))
                op.stdin.flush()
                print(f"[+] Get key {key} - value {value}. It has been deleted")

            process_bar(count/len(self.keys_only), start_str='', end_str="100%", total_length=15)

        print('[+] Paring done')

kv = KV()
kv.run()

Author: DioPong

Permalink: https://blog.2to.fun/post/technics/how-to-del-cloudflare-kv/

文章默认使用 CC BY-NC-SA 4.0 协议进行许可，使用时请注意遵守协议。

updated at 2021-12-09

CloudFlare