73 lines
2.5 KiB
Python
73 lines
2.5 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_request(endpoint, method="GET", data=None):
|
|
url = f"{AK_BASE.rstrip('/')}/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 Scope Mapping
|
|
code, res = ak_request("propertymappings/provider/scope/")
|
|
existing = next((pm for pm in res.get("results", []) if pm.get("name") == "Kavita Roles Mapping"), None)
|
|
|
|
expr = """
|
|
roles = ["Login", "Download"]
|
|
if request.user.is_superuser or any(g.name.lower() in ["authentik admins", "admins", "admin"] for g in request.user.ak_groups.all()):
|
|
roles.append("Admin")
|
|
|
|
for g in request.user.ak_groups.all():
|
|
roles.append(g.name)
|
|
|
|
return {
|
|
"roles": roles,
|
|
"http://schemas.microsoft.com/ws/2008/06/identity/claims/role": roles,
|
|
}
|
|
""".strip()
|
|
|
|
if existing:
|
|
mapping_pk = existing["pk"]
|
|
ak_request(f"propertymappings/provider/scope/{mapping_pk}/", method="PATCH", data={
|
|
"expression": expr,
|
|
"scope_name": "roles",
|
|
"description": "Kavita OIDC Roles and Permissions"
|
|
})
|
|
print(f"Updated existing Kavita Scope Mapping ({mapping_pk})")
|
|
else:
|
|
payload = {
|
|
"name": "Kavita Roles Mapping",
|
|
"scope_name": "roles",
|
|
"description": "Kavita OIDC Roles and Permissions",
|
|
"expression": expr,
|
|
}
|
|
code, new_pm = ak_request("propertymappings/provider/scope/", method="POST", data=payload)
|
|
mapping_pk = new_pm["pk"]
|
|
print(f"Created Kavita Scope Mapping ({mapping_pk})")
|
|
|
|
# 2. Attach to Provider 16 (Kavita)
|
|
code, prov = ak_request("providers/oauth2/16/")
|
|
current_pms = prov.get("property_mappings", [])
|
|
if mapping_pk not in current_pms:
|
|
current_pms.append(mapping_pk)
|
|
ak_request("providers/oauth2/16/", method="PATCH", data={"property_mappings": current_pms})
|
|
print(f"Attached mapping {mapping_pk} to Provider 16 (Kavita)")
|
|
else:
|
|
print(f"Mapping {mapping_pk} already attached to Provider 16")
|
|
|
|
print("Done! Provider 16 property mappings:", current_pms)
|