feat(cloudflare): new Cloudflare edge-containment integration
Cloudflare client API v4, 9 commands: list/get zones, block IP (firewall access rule), list/delete access rules, list/create/delete DNS records. API-token (Bearer) auth, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Guillaume BOURGEOIS
@@ -0,0 +1,108 @@
|
||||
id: cloudflare
|
||||
name: Cloudflare
|
||||
version: 1.0.0
|
||||
description: "Cloudflare (client API v4) — network containment at the edge: list zones, block/unblock IPs (firewall access rules), list access rules, and manage DNS records (list/create/delete). API-token (Bearer) authentication; stdlib-only, no extra Python dependencies."
|
||||
changelog: "1.0.0 — Initial release: list/get zones, block IP, list/delete access rules, list/create/delete DNS records."
|
||||
category: network
|
||||
|
||||
# Per-instance configuration. The API token is sent as 'Authorization: Bearer <api_token>'.
|
||||
config_schema:
|
||||
properties:
|
||||
api_token:
|
||||
type: string
|
||||
description: "Cloudflare API token (scoped to the needed zones/permissions)"
|
||||
x-soar-sensitive: true
|
||||
required:
|
||||
- api_token
|
||||
|
||||
commands:
|
||||
- id: list_zones
|
||||
name: cloudflare-list-zones
|
||||
description: "List zones (domains) in the account."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
name: { type: string, description: "Optional zone name filter" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
- id: get_zone
|
||||
name: cloudflare-get-zone
|
||||
description: "Get a single zone by ID."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
zone_id: { type: string, description: "Zone ID" }
|
||||
required: [zone_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: block_ip
|
||||
name: cloudflare-block-ip
|
||||
description: "Create a firewall access rule to block (or challenge) an IP, range, ASN, or country."
|
||||
inputs_schema:
|
||||
properties:
|
||||
zone_id: { type: string, description: "Zone ID" }
|
||||
value: { type: string, description: "IP, CIDR, ASN (AS####), or 2-letter country code" }
|
||||
target: { type: string, description: "ip, ip_range, asn, or country (default ip)" }
|
||||
mode: { type: string, description: "block, challenge, js_challenge, or whitelist (default block)" }
|
||||
notes: { type: string, description: "Optional note" }
|
||||
required: [zone_id, value]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: list_access_rules
|
||||
name: cloudflare-list-access-rules
|
||||
description: "List firewall access rules for a zone."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
zone_id: { type: string, description: "Zone ID" }
|
||||
required: [zone_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: delete_access_rule
|
||||
name: cloudflare-delete-access-rule
|
||||
description: "Delete a firewall access rule (unblock)."
|
||||
inputs_schema:
|
||||
properties:
|
||||
zone_id: { type: string, description: "Zone ID" }
|
||||
rule_id: { type: string, description: "Access rule ID" }
|
||||
required: [zone_id, rule_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: list_dns_records
|
||||
name: cloudflare-list-dns-records
|
||||
description: "List DNS records for a zone."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
zone_id: { type: string, description: "Zone ID" }
|
||||
record_type: { type: string, description: "Optional record type filter (A, AAAA, CNAME, TXT, ...)" }
|
||||
name: { type: string, description: "Optional record name filter" }
|
||||
required: [zone_id]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: create_dns_record
|
||||
name: cloudflare-create-dns-record
|
||||
description: "Create a DNS record (e.g. sinkhole a malicious host)."
|
||||
inputs_schema:
|
||||
properties:
|
||||
zone_id: { type: string, description: "Zone ID" }
|
||||
record_type: { type: string, description: "Record type (A, AAAA, CNAME, TXT, ...)" }
|
||||
name: { type: string, description: "Record name" }
|
||||
content: { type: string, description: "Record content (IP or target)" }
|
||||
ttl: { type: number, description: "TTL seconds (1 = automatic, default 1)" }
|
||||
proxied: { type: boolean, description: "Proxy through Cloudflare (default false)" }
|
||||
required: [zone_id, record_type, name, content]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: delete_dns_record
|
||||
name: cloudflare-delete-dns-record
|
||||
description: "Delete a DNS record."
|
||||
inputs_schema:
|
||||
properties:
|
||||
zone_id: { type: string, description: "Zone ID" }
|
||||
record_id: { type: string, description: "DNS record ID" }
|
||||
required: [zone_id, record_id]
|
||||
outputs_schema: { properties: {} }
|
||||
|
||||
- id: test_connection
|
||||
name: cloudflare-test-connection
|
||||
description: "Verify the API token (used by the Test button)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties: {}
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
@@ -0,0 +1,60 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
BASE = "https://api.cloudflare.com/client/v4"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def request(method, path, cfg, body=None, params=None):
|
||||
url = BASE + path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
url += "?" + urllib.parse.urlencode(clean)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
print(json.dumps(fn(_cfg(), _inputs())))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
zone_id = inputs.get("zone_id")
|
||||
if not zone_id:
|
||||
raise Exception("zone_id is required")
|
||||
value = inputs.get("value")
|
||||
if not value:
|
||||
raise Exception("value is required")
|
||||
target = inputs.get("target") or "ip"
|
||||
mode = inputs.get("mode") or "block"
|
||||
notes = inputs.get("notes")
|
||||
body = {"mode": mode, "configuration": {"target": target, "value": value}}
|
||||
if notes:
|
||||
body["notes"] = notes
|
||||
return request("POST", "/zones/" + q(zone_id) + "/firewall/access_rules/rules", cfg, body=body)
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,69 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
BASE = "https://api.cloudflare.com/client/v4"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def request(method, path, cfg, body=None, params=None):
|
||||
url = BASE + path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
url += "?" + urllib.parse.urlencode(clean)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
print(json.dumps(fn(_cfg(), _inputs())))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
zone_id = inputs.get("zone_id")
|
||||
if not zone_id:
|
||||
raise Exception("zone_id is required")
|
||||
record_type = inputs.get("record_type")
|
||||
if not record_type:
|
||||
raise Exception("record_type is required")
|
||||
name = inputs.get("name")
|
||||
if not name:
|
||||
raise Exception("name is required")
|
||||
content = inputs.get("content")
|
||||
if not content:
|
||||
raise Exception("content is required")
|
||||
ttl = inputs.get("ttl")
|
||||
proxied = inputs.get("proxied")
|
||||
body = {
|
||||
"type": record_type,
|
||||
"name": name,
|
||||
"content": content,
|
||||
"ttl": int(ttl or 1),
|
||||
"proxied": bool(proxied),
|
||||
}
|
||||
return request("POST", "/zones/" + q(zone_id) + "/dns_records", cfg, body=body)
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,54 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
BASE = "https://api.cloudflare.com/client/v4"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def request(method, path, cfg, body=None, params=None):
|
||||
url = BASE + path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
url += "?" + urllib.parse.urlencode(clean)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
print(json.dumps(fn(_cfg(), _inputs())))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
zone_id = inputs.get("zone_id")
|
||||
if not zone_id:
|
||||
raise Exception("zone_id is required")
|
||||
rule_id = inputs.get("rule_id")
|
||||
if not rule_id:
|
||||
raise Exception("rule_id is required")
|
||||
return request("DELETE", "/zones/" + q(zone_id) + "/firewall/access_rules/rules/" + q(rule_id), cfg)
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,54 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
BASE = "https://api.cloudflare.com/client/v4"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def request(method, path, cfg, body=None, params=None):
|
||||
url = BASE + path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
url += "?" + urllib.parse.urlencode(clean)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
print(json.dumps(fn(_cfg(), _inputs())))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
zone_id = inputs.get("zone_id")
|
||||
if not zone_id:
|
||||
raise Exception("zone_id is required")
|
||||
record_id = inputs.get("record_id")
|
||||
if not record_id:
|
||||
raise Exception("record_id is required")
|
||||
return request("DELETE", "/zones/" + q(zone_id) + "/dns_records/" + q(record_id), cfg)
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,51 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
BASE = "https://api.cloudflare.com/client/v4"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def request(method, path, cfg, body=None, params=None):
|
||||
url = BASE + path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
url += "?" + urllib.parse.urlencode(clean)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
print(json.dumps(fn(_cfg(), _inputs())))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
zone_id = inputs.get("zone_id")
|
||||
if not zone_id:
|
||||
raise Exception("zone_id is required")
|
||||
return request("GET", "/zones/" + q(zone_id), cfg)
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,51 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
BASE = "https://api.cloudflare.com/client/v4"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def request(method, path, cfg, body=None, params=None):
|
||||
url = BASE + path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
url += "?" + urllib.parse.urlencode(clean)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
print(json.dumps(fn(_cfg(), _inputs())))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
zone_id = inputs.get("zone_id")
|
||||
if not zone_id:
|
||||
raise Exception("zone_id is required")
|
||||
return request("GET", "/zones/" + q(zone_id) + "/firewall/access_rules/rules", cfg, params={"per_page": 50})
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,58 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
BASE = "https://api.cloudflare.com/client/v4"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def request(method, path, cfg, body=None, params=None):
|
||||
url = BASE + path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
url += "?" + urllib.parse.urlencode(clean)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
print(json.dumps(fn(_cfg(), _inputs())))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
zone_id = inputs.get("zone_id")
|
||||
if not zone_id:
|
||||
raise Exception("zone_id is required")
|
||||
record_type = inputs.get("record_type")
|
||||
name = inputs.get("name")
|
||||
return request(
|
||||
"GET",
|
||||
"/zones/" + q(zone_id) + "/dns_records",
|
||||
cfg,
|
||||
params={"type": record_type, "name": name, "per_page": 100},
|
||||
)
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,46 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
BASE = "https://api.cloudflare.com/client/v4"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def request(method, path, cfg, body=None, params=None):
|
||||
url = BASE + path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
url += "?" + urllib.parse.urlencode(clean)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
print(json.dumps(fn(_cfg(), _inputs())))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
name = inputs.get("name")
|
||||
return request("GET", "/zones", cfg, params={"name": name, "per_page": 50})
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,46 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
BASE = "https://api.cloudflare.com/client/v4"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def request(method, path, cfg, body=None, params=None):
|
||||
url = BASE + path
|
||||
if params:
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
if clean:
|
||||
url += "?" + urllib.parse.urlencode(clean)
|
||||
data = json.dumps(body).encode("utf-8") if body is not None else None
|
||||
headers = {"Authorization": "Bearer " + str(cfg.get("api_token", "")), "Accept": "application/json"}
|
||||
if data is not None:
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def _run(fn):
|
||||
try:
|
||||
print(json.dumps(fn(_cfg(), _inputs())))
|
||||
except urllib.error.HTTPError as e:
|
||||
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"error": str(e)}))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def main(cfg, inputs):
|
||||
request("GET", "/user/tokens/verify", cfg)
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
_run(main)
|
||||
Reference in New Issue
Block a user