はじめに
CloudflareのGUIは直感的で分かりやすく、設定投入も監視も使いやすいですよね。
ただ、業務で使っていると「これ毎回手でやるの面倒だな~」「こういうことできたらいいのにな~」と感じる場面が出てきます。
そこで、Cloudflare API を使って自動化してみたら便利なのでは?と思い立ち、実際に取り組んだ内容をまとめました。
今回扱ったのは次の2つです。
①管理者メンバーの招待、ロールの割り当ての自動化
②デバイス一覧を取得、メールドメインでのフィルターを行い、ユーザ・デバイスを監視する
どちらも Python で実装しており、手順やコードも紹介します。
API・Cloudflare APIとは
APIは、ソフトウェア同士が決められたルールで情報や機能をやり取りするための仕組みです。
CloudflareはREST APIを採用しており、GUIから実施できる多くの操作をAPI経由で実行できます。
APIの基本要素は以下の通りです。
・エンドポイント(URL):アクセス先
・メソッド(GET/POST/PUT/DELETE):何をしたいか
・リクエスト:送るデータ
・レスポンス:返ってくるデータ
・認証:APIキーやトークンなどでのアクセス制御
Cloudflare の API ドキュメントはこちらにまとまっています。
認証はAPIキーではなくAPIトークンを使います。
こちらのドキュメントに掲載されているエンドポイントのみ操作可能です。
ここに載っていない設定は取得も変更もできません。
準備するもの
・APIトークン(認証情報、Cloudflareダッシュボード上で発行)
・Python、VSCode(HTTPSでJSONを扱える環境)
REST APIはHTTPSでJSONをやり取りするため、HTTPSの実行環境が必要になります。
PowerShellでもOKですが、今回はPythonを利用しました。
また、Cloudflareには公式SDKとしてcloudflare-pythonが提供されており、REST APIを直接呼び出すよりもシンプルに実装できます。
PowerShellとの違いやメリットは細かく言うと色々ありますが、私が実感したのは下記です。
・コードをシンプル&簡潔に記述できる
・ライブラリによる補完が利用できる
・exe化することで配布しやすい
①管理者メンバーの招待とロール割り当てを自動化する
新規アカウント作成時に、複数の管理者メンバーをまとめて招待したい場面があります。
GUIから実施することも可能ですが、メンバー数が多い場合は手作業によるミスや作業負荷が発生しやすくなります。
そこで今回は、Cloudflare APIを利用して、
・メンバー招待
・ロール割り当て
を自動で実施するツールを作成しました。
必要なライブラリとexe化コマンド
本記事で紹介するツールを動かすには、事前に以下のPythonライブラリのインストールが必要です。
pip install requests truststore pyinstaller
| ライブラリ | 用途 |
|---|---|
| requests | Cloudflare API へのHTTPリクエスト |
| truststore | OSの証明書ストアを利用(社内プロキシ環境向け) |
| pyinstaller | exe化 |
また、pyファイル(Pythonのスクリプト)のままではPythonをインストールしていないユーザーは実行できないため、誰にでも配布できるようexe化しています。(exeファイルならダブルクリックで誰でも実行可能!)
exe化コマンドは、下記です。Pythonスクリプトを作成し終えたら、実行します。
pyinstaller --onefile invite.py
生成されたexeは dist フォルダ(自動生成)配下に出力されます。
①の users.csv など、参照ファイルがある場合はexeと同じフォルダに配置しなおす必要があります。
ファイル構成
作成したファイル構成は、以下の通りです。
このツールを配布する場合は「Cloudflare_Inviter」フォルダごと共有します。
Cloudflare_Inviter
┝ invite.exe
┝ users.csv
└ token.txt
それぞれのファイルの役割です。
・invite.exe
メンバ招待/ロール割り当てを行うPythonスクリプトをexe化したもの。配布して誰でも実行できるようにしています。
・users.csv
招待したいメンバーのメールアドレスとロールを記載。
アカウントごとにメンバーが変わるのにも対応できるよう、exeに埋め込まず外部ファイルとして読み込む方式にしました。ここを編集することで招待するメンバ/ロールを変更する事ができます。
・token.txt
初回実行時に自動生成される、APIトークンを保存するファイル。
毎回トークンを入力するのは面倒なので、
一度入力 → 自動保存 → 次回以降は自動読み込み
という仕組みにしています。
こちらも、配布して複数ユーザが使えるよう、exeには埋め込まずに初回実行時に入力する形としています。
ファイルの中身
各ファイルの中身は下記のとおりです。
invite.py
pyファイルをexe化して利用します。"""
invite.py
Cloudflare Member Inviter
"""
import truststore
truststore.inject_into_ssl()
import sys
from pathlib import Path
import requests
import csv
BASE_URL = "https://api.cloudflare.com/client/v4"
TOKEN_FILE = Path("token.txt")
def load_users():
users = {}
with open(
"users.csv",
encoding="utf-8-sig"
) as f:
reader = csv.DictReader(f)
for row in reader:
roles = tuple(
r.strip()
for r in row["roles"].split("|")
)
email = row["email"].strip()
users.setdefault(roles, []).append(email)
return users
def get_api_token():
if TOKEN_FILE.exists():
token = TOKEN_FILE.read_text(
encoding="utf-8"
).strip()
if token:
print("保存済みのAPI Tokenを使用します。")
return token
token = input(
"Cloudflare API Tokenを入力してください: "
).strip()
if not token:
sys.exit("API Tokenが入力されていません。")
TOKEN_FILE.write_text(
token,
encoding="utf-8"
)
print("API Tokenを token.txt に保存しました。")
return token
def build_headers(api_token):
return {
"Authorization": f"Bearer {api_token}",
"Content-Type": "application/json",
}
def get_account_name(account_id, headers):
r = requests.get(
f"{BASE_URL}/accounts/{account_id}",
headers=headers
)
r.raise_for_status()
return r.json()["result"]["name"]
def get_role_map(account_id, headers):
role_map = {}
page = 1
while True:
r = requests.get(
f"{BASE_URL}/accounts/{account_id}/roles",
headers=headers,
params={"page": page, "per_page": 100},
)
r.raise_for_status()
result = r.json()["result"]
if not result:
break
for role in result:
role_map[role["name"]] = role["id"]
page += 1
return role_map
def invite(account_id, role_map, headers, users):
success = failed = skipped = 0
print(f"\n===== {get_account_name(account_id, headers)} =====")
for roles, emails in users.items():
role_ids = [role_map[r] for r in roles]
for email in emails:
body = {
"email": email,
"roles": role_ids
}
r = requests.post(
f"{BASE_URL}/accounts/{account_id}/members",
headers=headers,
json=body,
)
if r.ok:
print(f"✅ {email}")
success += 1
else:
txt = r.text.lower()
if "already" in txt or "exists" in txt:
print(f"⏭ {email}")
skipped += 1
else:
print(f"❌ {email}")
print(f" {r.status_code}: {r.text}")
failed += 1
return success, failed, skipped
def main():
users = load_users()
print("============================================================")
print("Cloudflare Member Inviter")
print("============================================================")
api_token = get_api_token()
headers = build_headers(api_token)
ids = [x.strip() for x in input(
"\nAccount IDを入力してください(複数はカンマ区切り)\n> "
).split(",") if x.strip()]
if not ids:
sys.exit("Account IDが入力されていません。")
accounts = [(i, get_account_name(i, headers)) for i in ids]
print("\n対象アカウント")
print("------------------------------------------------------------")
for _, name in accounts:
print(f"・{name}")
print("\n付与する権限")
print("------------------------------------------------------------\n")
for roles, emails in users.items():
if len(roles) == 1:
title = roles[0]
else:
title = " + ".join(roles)
print(f"■ {title}")
for email in emails:
print(f" {email}")
print()
print("------------------------------------------------------------")
print(f"対象アカウント : {len(accounts)}")
print(f"招待ユーザー : {sum(len(v) for v in users.values())}")
print("------------------------------------------------------------")
confirm = input("\n実行しますか? (y/N): ").strip().lower()
if confirm != "y":
print("キャンセルしました。")
return
s = f = sk = 0
for account_id, _ in accounts:
role_map = get_role_map(account_id, headers)
a, b, c = invite(account_id, role_map, headers, users)
s += a
f += b
sk += c
print("\n============================================================")
print("完了")
print("============================================================")
print(f"成功 : {s}")
print(f"スキップ : {sk}")
print(f"失敗 : {f}")
if __name__ == "__main__":
try:
main()
except Exception:
import traceback
print("\n===== エラー =====")
traceback.print_exc()
finally:
input("\nEnterキーを押すと終了します...")
users.csv
email,roles
user1@example.com,"Super Administrator - All Privileges"
user2@example.com,"Administrator Read Only|Cloudflare Zero Trust PII"
利用するAPIエンドポイントは、下記です。
・/accounts/{account_id}
・/accounts/{account_id}/roles
・/accounts/{account_id}/members
実行の流れ
APIトークンの入力
invite.exeを実行すると、まずAPIトークンの入力を求められます。

※2度目の実行時は、トークンが自動で保存されるので、入力不要になります。

アカウントIDの入力
次に、アカウント IDの入力を求められます。
複数アカウントを同時に作成した場合は、カンマ区切りで複数指定することも可能です。

メンバー情報の確認
Enterを押すと、指定したアカウントIDに紐づく アカウント名、そして users.csvに記載した 招待対象のメールアドレスとロール が一覧で表示されます。
内容に問題なければ y を入力して確定します。

自動で招待を実行
各アカウントに対して Cloudflare API を使って 管理者メンバーの招待処理を自動実行します。
処理結果はすべて画面に表示され、成功・失敗が一目で分かります。

正常に招待が送信されれば、これで完了です。
「アカウントID」、「y」、「Enter」だけで、GUIよりも圧倒的に早く、正確に実行できるようになりました!
②デバイス一覧を取得、メールドメインでのフィルターを行い、ユーザ・デバイスを監視する
Cloudflare Zero Trustのダッシュボードではデバイス一覧を確認できます。
しかし運用を続けていると、次のような課題を感じることがあります。
・特定ドメインのユーザだけを抽出したい
・同一ユーザが複数端末を使っているため、ユーザ数を正確に把握しづらい
・デバイス数が多いと複数ページに分かれ、一覧性が低い
・会社別にCloudflare One Clientの配布状況を簡単に可視化したい
そこで、Cloudflare APIを使って以下の処理を自動化しました。
★デバイス一覧を API で取得する
↓
★メールドメインでフィルターする(例:@example.com)
↓
★ユーザ単位で重複を排除する
↓
★Excel に出力して一覧化する
API・Pythonを使うことで、ダッシュボードでは難しいユーザ単位の正確な集計が簡単にできるようになります。
さらに今回は、サードパーティ製ライブラリであるcustomtkinterを利用し、GUI化にも挑戦しました!
CLIのみでも十分実用的ですが、GUI化することでPython経験のないメンバーでも利用しやすくなります。
必要なライブラリとexe化コマンド
本ツールを動かすには、事前に以下のPythonライブラリのインストールが必要です。
pip install requests openpyxl customtkinter truststore pyinstaller
| ライブラリ | 用途 |
|---|---|
| requests | Cloudflare API へのHTTPリクエスト |
| openpyxl | Excel(.xlsx)の出力 |
| customtkinter | GUI画面の作成 |
| truststore | OSの証明書ストアを利用(社内プロキシ環境向け) |
| pyinstaller | exe化 |
今回のexe化コマンドは、下記です。Pythonスクリプトを作成し終えたら、実行します。
pyinstaller --onefile --noconsole main.py
①のツールでは付いていなかった--noconsole オプションは黒いコンソール画面を非表示にするオプションです。
本ツールのようなGUIツールでは付けますが、①のようにコンソールへ入力させるツールに付けると入力画面ごと消えてしまい操作できなくなるので注意してください。
生成されたexeは dist フォルダ(自動生成)配下に出力されます。
ファイル構成
作成したファイル構成は、以下の通りです。
※Get_devices.pyは配布不要
Get_Devices
┝ Get_devices.py
└ main.exe
それぞれのファイルの役割です。
・Get_devices.py
デバイス一覧を取得し、Excel出力するPythonスクリプト。
・main.exe
GUI表示させるためのPythonスクリプトをexe化し、配布可能な形にしたもの。
ファイルの中身
各ファイルの中身は、下記です。
Get_devices.py
import truststore
truststore.inject_into_ssl()
import requests
from datetime import datetime
from collections import Counter
from openpyxl import Workbook
from openpyxl.styles import Font
import os
def create_report(
token,
account_id,
target_domain,
log_callback=None,
progress_callback=None,
info_callback=None
):
"""
Cloudflare Device Export
"""
# ==========================
# GUIログ表示
# ==========================
def log(message):
if log_callback:
log_callback(message)
else:
print(message)
# ==========================
# プログレスバー更新
# ==========================
def progress(value):
if progress_callback:
progress_callback(value)
progress(0.0)
headers = {
"Authorization": f"Bearer {token}"
}
# ==========================
# アカウント情報取得
# ==========================
log("Cloudflareへ接続中...")
progress(0.05)
account_url = (
f"https://api.cloudflare.com/client/v4/accounts/{account_id}"
)
account_response = requests.get(
account_url,
headers=headers
)
if account_response.status_code != 200:
raise Exception(
f"アカウント情報の取得に失敗しました。\n"
f"{account_response.text}"
)
account_name = account_response.json()["result"]["name"]
# GUIへ対象アカウントを表示
if info_callback:
info_callback(account_name, target_domain)
if log_callback:
log_callback(f"対象アカウント : {account_name}")
log(f"対象アカウント : {account_name}")
log(f"対象ドメイン : {target_domain}")
progress(0.15)
# ==========================
# デバイス一覧取得
# ==========================
log("デバイス一覧取得中...")
devices = []
cursor = None
while True:
url = (
f"https://api.cloudflare.com/client/v4/accounts/"
f"{account_id}/devices/physical-devices"
"?per_page=100"
)
if cursor:
url += f"&cursor={cursor}"
response = requests.get(
url,
headers=headers
)
if response.status_code != 200:
raise Exception(response.text)
data = response.json()
if not data["success"]:
raise Exception(str(data))
devices.extend(data["result"])
cursor = data["result_info"].get("cursor")
if not cursor:
break
log(f"{len(devices)} 台取得しました")
progress(0.40)
# ==========================
# ドメイン抽出
# ==========================
target_devices = []
log("対象ユーザー抽出中...")
for d in devices:
user = d.get("last_seen_user") or {}
email = user.get("email")
if email and email.lower().endswith(target_domain.lower()):
target_devices.append({
"device": d.get("name"),
"email": email,
"version": d.get("client_version")
})
progress(0.60)
# ==========================
# メールアドレス順
# ==========================
target_devices.sort(
key=lambda x: x["email"].lower()
)
# ==========================
# ユーザー集計
# ==========================
counter = Counter(
d["email"] for d in target_devices
)
log(f"対象ユーザー数 : {len(counter)}")
progress(0.70)
# ==========================
# Excel作成
# ==========================
log("Excel作成中...")
wb = Workbook()
ws = wb.active
ws.title = "Summary"
ws.append(["項目", "件数"])
for cell in ws[1]:
cell.font = Font(bold=True)
ws.append(["総デバイス数", len(devices)])
ws.append(["対象デバイス数", len(target_devices)])
ws.append(["ユーザー数", len(counter)])
progress(0.80)
# ---------------- Users ----------------
ws = wb.create_sheet("Users")
ws.append([
"メールアドレス",
"デバイス数"
])
for cell in ws[1]:
cell.font = Font(bold=True)
for email in sorted(counter):
ws.append([
email,
counter[email]
])
# ---------------- Devices ----------------
ws = wb.create_sheet("Devices")
ws.append([
"デバイス名",
"メールアドレス",
"クライアントバージョン"
])
for cell in ws[1]:
cell.font = Font(bold=True)
for d in target_devices:
ws.append([
d["device"],
d["email"],
d["version"]
])
progress(0.90)
# ==========================
# 保存
# ==========================
log("Excel保存中...")
desktop = os.path.join(
os.path.expanduser("~"),
"Desktop"
)
filename = os.path.join(
desktop,
f"Cloudflare_Devices_{datetime.now():%Y%m%d_%H%M%S}.xlsx"
)
wb.save(filename)
progress(1.0)
log("")
log("========================================")
log("レポート作成完了")
log(f"総デバイス数 : {len(devices)}")
log(f"対象デバイス数 : {len(target_devices)}")
log(f"ユーザー数 : {len(counter)}")
log(filename)
# Excelを開く
os.startfile(filename)
return (
filename,
len(devices),
len(target_devices),
len(counter),
account_name
)
# ==========================================
# 単体実行
# ==========================================
if __name__ == "__main__":
token = input("API Token : ")
account_id = input("Account ID : ")
target_domain = input("Target Domain : ")
create_report(
token,
account_id,
target_domain
)
input("\nEnterキーで終了...")
main.py
pyファイルをexe化して利用します。
import truststore
truststore.inject_into_ssl()
import customtkinter as ctk
from tkinter import messagebox
from Get_devices import create_report
# ==========================================
# テーマ
# ==========================================
ctk.set_appearance_mode("System")
ctk.set_default_color_theme("blue")
# ==========================================
# ウィンドウ
# ==========================================
app = ctk.CTk()
app.title("Cloudflare Device Export Tool")
app.geometry("760x950")
app.resizable(False, False)
# ==========================================
# コールバック
# ==========================================
def add_log(message):
log.configure(state="normal")
log.insert("end", message + "\n")
log.see("end")
log.configure(state="disabled")
app.update()
def update_progress(value):
progress.set(value)
app.update()
def set_info(account_name, target_domain):
account_value.configure(text=account_name)
domain_value.configure(text=target_domain)
app.update()
def set_summary(total, target, users):
total_value.configure(text=str(total))
target_value.configure(text=str(target))
user_value.configure(text=str(users))
app.update()
# ==========================================
# ボタン処理
# ==========================================
def run():
token = token_entry.get().strip()
account = account_entry.get().strip()
domain = domain_entry.get().strip()
if not token or not account or not domain:
messagebox.showwarning(
"入力エラー",
"API Token・Account ID・Target Domainを入力してください。"
)
return
button.configure(state="disabled")
progress.set(0)
log.configure(state="normal")
log.delete("1.0", "end")
log.configure(state="disabled")
set_info("取得中...", domain)
try:
filename, total, target, users, account_name = create_report(
token,
account,
domain,
add_log,
update_progress,
set_info
)
set_summary(
total,
target,
users
)
messagebox.showinfo(
"完了",
"レポートを作成しました!"
)
except Exception as e:
messagebox.showerror(
"エラー",
str(e)
)
finally:
button.configure(state="normal")
# ==========================================
# メインフレーム
# ==========================================
frame = ctk.CTkFrame(app)
frame.pack(
fill="both",
expand=True,
padx=20,
pady=20
)
# ==========================================
# タイトル
# ==========================================
title = ctk.CTkLabel(
frame,
text="Cloudflare Device Export Tool",
font=("Yu Gothic UI", 28, "bold")
)
title.pack(pady=20)
# ==========================================
# API Token
# ==========================================
ctk.CTkLabel(
frame,
text="API Token",
font=("Yu Gothic UI", 12, "bold")
).pack(anchor="w", padx=25)
token_entry = ctk.CTkEntry(
frame,
width=720,
show="*",
corner_radius=10
)
token_entry.pack(
padx=25,
pady=(5, 15)
)
# ==========================================
# Account ID
# ==========================================
ctk.CTkLabel(
frame,
text="Account ID",
font=("Yu Gothic UI", 12, "bold")
).pack(anchor="w", padx=25)
account_entry = ctk.CTkEntry(
frame,
width=720,
corner_radius=10
)
account_entry.pack(
padx=25,
pady=(5, 15)
)
# ==========================================
# Target Domain
# ==========================================
ctk.CTkLabel(
frame,
text="Target Domain",
font=("Yu Gothic UI", 12, "bold")
).pack(anchor="w", padx=25)
domain_entry = ctk.CTkEntry(
frame,
width=720,
corner_radius=10
)
domain_entry.pack(
padx=25,
pady=(5, 20)
)
# ==========================================
# レポート作成ボタン
# ==========================================
button = ctk.CTkButton(
frame,
text="Excel作成",
command=run,
width=140,
height=40,
corner_radius=8, # 角丸
font=("Meiryo", 14),
fg_color="#0078D7",
hover_color="#005A9E"
)
button.pack(pady=20)
# ==========================================
# 対象情報カード
# ==========================================
info_frame = ctk.CTkFrame(frame)
info_frame.pack(
fill="x",
padx=20,
pady=(0, 15)
)
ctk.CTkLabel(
info_frame,
text="対象アカウント",
font=("Yu Gothic UI", 12, "bold")
).pack(
anchor="w",
padx=20,
pady=(15, 0)
)
account_value = ctk.CTkLabel(
info_frame,
text="-",
font=("Yu Gothic UI", 22, "bold")
)
account_value.pack(
anchor="w",
padx=20
)
ctk.CTkLabel(
info_frame,
text="対象ドメイン",
font=("Yu Gothic UI", 12, "bold")
).pack(
anchor="w",
padx=20,
pady=(10, 0)
)
domain_value = ctk.CTkLabel(
info_frame,
text="-",
font=("Yu Gothic UI", 18)
)
domain_value.pack(
anchor="w",
padx=20,
pady=(0, 15)
)
# ==========================================
# 集計カード
# ==========================================
cards = ctk.CTkFrame(frame)
cards.pack(
fill="x",
padx=20,
pady=(0, 20)
)
titles = [
"総デバイス",
"対象デバイス",
"ユーザー数"
]
values = []
for title in titles:
card = ctk.CTkFrame(cards)
card.pack(
side="left",
expand=True,
fill="both",
padx=6
)
value = ctk.CTkLabel(
card,
text="0",
font=("Yu Gothic UI", 28, "bold")
)
value.pack(
pady=(15, 5)
)
ctk.CTkLabel(
card,
text=title,
font=("Yu Gothic UI", 12)
).pack(
pady=(0, 15)
)
values.append(value)
total_value = values[0]
target_value = values[1]
user_value = values[2]
# ==========================================
# Progress
# ==========================================
ctk.CTkLabel(
frame,
text="進捗",
font=("Yu Gothic UI", 12, "bold")
).pack(
anchor="w",
padx=20
)
progress = ctk.CTkProgressBar(
frame,
width=720,
height=18
)
progress.pack(
padx=20,
pady=(5, 20)
)
progress.set(0)
# ==========================================
# 実行ログ
# ==========================================
ctk.CTkLabel(
frame,
text="実行ログ",
font=("Yu Gothic UI", 12, "bold")
).pack(
anchor="w",
padx=20
)
log = ctk.CTkTextbox(
frame,
width=720,
height=250,
font=("Consolas", 12)
)
log.pack(
padx=20,
pady=(5, 20)
)
log.insert(
"end",
"待機中...\n"
)
log.configure(
state="disabled"
)
# ==========================================
# 起動
# ==========================================
app.mainloop()
実行の流れ
main.exeを実行すると、下記のような画面が表示されます。

ここに、APIトークン、対象のアカウントID、フィルターしたいメールドメインを入力し、中央のExcel作成ボタンを押すと、デバイス一覧の取得が開始します。
入力したアカウントIDからアカウント名を取得し、アカウント名と対象メールドメイン、また取得の進捗が画面には表示されます。
取得が終了すると、下記のように対象アカウント、メールドメインのデバイス一覧をフィルターしたExcelが出力されます!



GUIの作りこみやExcelの出力の内容には、まだまだブラッシュアップの余地はあると思いますが、
ダッシュボードだけでは管理が難しかったものがAPIとPythonで簡単に出力できるアプリにすることができました!
終わりに
生成AIによって自動化のハードルはかなり低く、こんなもの作りたいな~というアイデアだけで簡単に形にできてしまうということが分かりました。
お客様の課題解決や、Cloudflare以外への応用にもつなげられるよう、理解を深めていきたいです!