用于阿里云CDN的fail2ban配置文件

#!/bin/env python3
# aliyun.py
import argparse
import json
import subprocess
def main():
  parser = argparse.ArgumentParser()
  parser.add_argument(
    "-a",
    "--action",
    choices=["add", "remove"],
    required=True,
    help="Action to perform: add or remove an IP from the blacklist.",
  )
  parser.add_argument("--access-key-id", default="", help="Aliyun Access Key ID.")
  parser.add_argument("--access-key-secret", default="", help="Aliyun Access Key Secret.")
  parser.add_argument("--region", default="cn-hangzhou", help="Aliyun region ID, default is cn-hangzhou.")
  parser.add_argument("-d", "--domain", required=True, help="The CDN domain to modify.")
  parser.add_argument("-i", "--ip", required=True, help="The IP address to add or remove from the blacklist.")
  args = parser.parse_args()
  p = subprocess.Popen(
    [
      "aliyun",
      "cdn",
      "DescribeCdnDomainConfigs",
      "--access-key-id",
      args.access_key_id,
      "--access-key-secret",
      args.access_key_secret,
      "--region",
      args.region,
      "--DomainName",
      args.domain,
      "--FunctionNames",
      "ip_black_list_set",
    ],
    stdout=subprocess.PIPE,
  )
  p.wait()
  fuction_args: list[dict] = json.loads(p.stdout.read().decode("utf-8"))["DomainConfigs"]["DomainConfig"][0][
    "FunctionArgs"
  ]["FunctionArg"]
  ip_list = []
  for arg in fuction_args:
    if arg["ArgName"] == "ip_list":
      ip_list = [i for i in arg["ArgValue"].split(",") if i]
      break
  if args.action == "add":
    if args.ip in ip_list:
      print(f"{args.ip} is already in the blacklist for {args.domain}.")
      exit(0)
    ip_list.append(args.ip)
  elif args.action == "remove":
    if args.ip not in ip_list:
      print(f"{args.ip} is not in the blacklist for {args.domain}.")
      exit(0)
    ip_list.remove(args.ip)
  functions = [
    {
      "functionArgs": [{"argName": "ip_list", "argValue": ",".join(ip_list)}],
      "functionName": "ip_black_list_set",
    }
  ]
  p = subprocess.Popen(
    [
      "aliyun",
      "cdn",
      "BatchSetCdnDomainConfig",
      "--access-key-id",
      args.access_key_id,
      "--access-key-secret",
      args.access_key_secret,
      "--region",
      args.region,
      "--DomainNames",
      args.domain,
      "--Functions",
      json.dumps(functions),
    ],
    stdout=subprocess.PIPE,
  )
  p.wait()
  print(json.loads(p.stdout.read().decode("utf-8")))
if __name__ == "__main__":
  main()