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

文档相关

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

  • 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()