68 lines
2.2 KiB
Python
68 lines
2.2 KiB
Python
import urllib.request
|
|
import json
|
|
import ssl
|
|
|
|
ctx = ssl.create_default_context()
|
|
ctx.check_hostname = False
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|
|
|
AK_BASE = "https://authentik.snarfnet.net"
|
|
AK_TOKEN = "SnarfNetPermanentAutomationToken2026"
|
|
|
|
def ak_req(endpoint, method="GET", data=None):
|
|
url = f"{AK_BASE}/api/v3/{endpoint.lstrip('/')}"
|
|
body = json.dumps(data).encode("utf-8") if data else None
|
|
headers = {
|
|
"Authorization": f"Bearer {AK_TOKEN}",
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json"
|
|
}
|
|
req = urllib.request.Request(url, data=body, headers=headers, method=method)
|
|
with urllib.request.urlopen(req, timeout=10, context=ctx) as r:
|
|
raw = r.read().decode("utf-8")
|
|
return r.getcode(), json.loads(raw) if raw else {}
|
|
|
|
# 1. Check or Create Flow
|
|
code, flows = ak_req("flows/instances/")
|
|
flow = next((f for f in flows.get("results", []) if f["slug"] == "ldap-bind-flow"), None)
|
|
|
|
if not flow:
|
|
payload = {
|
|
"name": "LDAP Bind Flow",
|
|
"slug": "ldap-bind-flow",
|
|
"title": "LDAP Bind Flow",
|
|
"designation": "authentication",
|
|
}
|
|
code, flow = ak_req("flows/instances/", method="POST", data=payload)
|
|
print("Created flow:", flow["pk"])
|
|
else:
|
|
print("Flow exists:", flow["pk"])
|
|
|
|
flow_pk = flow["pk"]
|
|
|
|
# 2. Check or Create Bindings
|
|
code, bindings = ak_req(f"flows/bindings/?target={flow_pk}")
|
|
existing_stage_pks = [b["stage"] for b in bindings.get("results", [])]
|
|
|
|
stages = [
|
|
(10, "f46eee4d-5bf7-415a-9eea-ba589536f90a"), # Identification
|
|
(20, "8f68c5d8-52f8-49a2-a578-5f43846d5edb"), # Password
|
|
(30, "84c2290f-eed3-464d-a997-5b70bb435b7e"), # Login
|
|
]
|
|
|
|
for order, stage_pk in stages:
|
|
if stage_pk not in existing_stage_pks:
|
|
b_payload = {
|
|
"target": flow_pk,
|
|
"stage": stage_pk,
|
|
"order": order,
|
|
}
|
|
code, new_b = ak_req("flows/bindings/", method="POST", data=b_payload)
|
|
print(f"Bound stage {stage_pk} at order {order}")
|
|
else:
|
|
print(f"Stage {stage_pk} already bound")
|
|
|
|
# 3. Update Provider 22
|
|
code, prov = ak_req("providers/ldap/22/", method="PATCH", data={"authentication_flow": flow_pk})
|
|
print("Updated Provider 22 authentication_flow to:", prov.get("authentication_flow"))
|