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

technics

首先声明,我的长短链使用的是 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 原生自带的 subprocessjson 两个库。

  • 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 协议进行许可,使用时请注意遵守协议。

Comments

Unable to load Disqus, please make sure your network can access.