k.rennerandClaude Opus 4.6
e13f365d54
- set_password endpoint returns empty body on success - Remove --silent flag from flux bootstrap (unsupported) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
383 lines
14 KiB
Python
Executable File
383 lines
14 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Migrate Authentik configuration from one instance to another via API.
|
|
|
|
Copies groups, users, OAuth2 providers, and applications. All data is read
|
|
from the source instance API — no hardcoded secrets or IDs.
|
|
|
|
Usage:
|
|
python3 migrate-authentik.py \
|
|
--old-url https://auth.renner.uno \
|
|
--old-token <token> \
|
|
--new-url https://auth.k.renner.uno \
|
|
--new-token <token> \
|
|
[--dry-run]
|
|
"""
|
|
|
|
import argparse
|
|
import sys
|
|
from urllib.request import Request, urlopen
|
|
from urllib.error import HTTPError
|
|
import json
|
|
import ssl
|
|
|
|
# Skip TLS verification (self-signed certs)
|
|
CTX = ssl.create_default_context()
|
|
CTX.check_hostname = False
|
|
CTX.verify_mode = ssl.CERT_NONE
|
|
|
|
|
|
def api(base_url, token, path, method="GET", data=None):
|
|
url = f"{base_url}/api/v3{path}"
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
body = json.dumps(data).encode() if data else None
|
|
req = Request(url, data=body, headers=headers, method=method)
|
|
try:
|
|
with urlopen(req, context=CTX) as resp:
|
|
body = resp.read()
|
|
if not body:
|
|
return {}
|
|
return json.loads(body)
|
|
except HTTPError as e:
|
|
e.read() # drain response
|
|
if e.code != 404: # 404s are expected when probing API paths
|
|
print(f" ERROR {e.code} {method} {path}", file=sys.stderr)
|
|
return None
|
|
|
|
|
|
def paginate(base_url, token, path):
|
|
"""Fetch all pages from a paginated endpoint."""
|
|
results = []
|
|
page = 1
|
|
while True:
|
|
sep = "&" if "?" in path else "?"
|
|
data = api(base_url, token, f"{path}{sep}page={page}&page_size=100")
|
|
if not data or "results" not in data:
|
|
break
|
|
results.extend(data["results"])
|
|
if not data["pagination"].get("next"):
|
|
break
|
|
page += 1
|
|
return results
|
|
|
|
|
|
def migrate_groups(old_url, old_token, new_url, new_token, dry_run):
|
|
"""Migrate groups, skipping built-in ones."""
|
|
print("\n=== Migrating Groups ===")
|
|
old_groups = paginate(old_url, old_token, "/core/groups/")
|
|
|
|
skip_prefixes = ("authentik ",) # built-in groups are recreated by blueprints
|
|
skip_names = {
|
|
"ldap_bind", "ldap_bind_unifi", "music_assistant_jellyfin",
|
|
"Unifi Access", "Unifi VPN", "ocis-spaceadmin", # empty groups
|
|
}
|
|
|
|
# Check which groups already exist on new instance
|
|
new_groups = paginate(new_url, new_token, "/core/groups/")
|
|
existing = {g["name"]: g["pk"] for g in new_groups}
|
|
|
|
group_map = {} # old pk -> new pk
|
|
for g in old_groups:
|
|
name = g["name"]
|
|
if any(name.startswith(p) for p in skip_prefixes):
|
|
# Map built-in groups by name
|
|
if name in existing:
|
|
group_map[g["pk"]] = existing[name]
|
|
print(f" SKIP (built-in) {name}")
|
|
continue
|
|
if name in skip_names:
|
|
continue
|
|
|
|
if name in existing:
|
|
group_map[g["pk"]] = existing[name]
|
|
print(f" EXISTS {name}")
|
|
continue
|
|
|
|
if dry_run:
|
|
print(f" DRY-RUN would create group: {name}")
|
|
continue
|
|
|
|
result = api(new_url, new_token, "/core/groups/", method="POST", data={
|
|
"name": name,
|
|
"is_superuser": g["is_superuser"],
|
|
})
|
|
if result and "pk" in result:
|
|
group_map[g["pk"]] = result["pk"]
|
|
print(f" CREATED {name}")
|
|
else:
|
|
print(f" FAILED {name}")
|
|
|
|
return group_map
|
|
|
|
|
|
def migrate_users(old_url, old_token, new_url, new_token, group_map, dry_run):
|
|
"""Migrate real users, skipping service accounts."""
|
|
print("\n=== Migrating Users ===")
|
|
old_users = paginate(old_url, old_token, "/core/users/")
|
|
|
|
skip_prefixes = ("ak-outpost-", "ldap_bind", "music_assistant")
|
|
|
|
new_users = paginate(new_url, new_token, "/core/users/")
|
|
existing = {u["username"]: u["pk"] for u in new_users}
|
|
|
|
for u in old_users:
|
|
username = u["username"]
|
|
if any(username.startswith(p) for p in skip_prefixes):
|
|
print(f" SKIP (service account) {username}")
|
|
continue
|
|
if username in existing:
|
|
print(f" EXISTS {username}")
|
|
continue
|
|
|
|
# Map old group PKs to new group PKs
|
|
new_groups = []
|
|
for g in u.get("groups_obj", []):
|
|
new_pk = group_map.get(g["pk"])
|
|
if new_pk:
|
|
new_groups.append(new_pk)
|
|
|
|
if dry_run:
|
|
print(f" DRY-RUN would create user: {username} groups={[g['name'] for g in u.get('groups_obj', [])]}")
|
|
continue
|
|
|
|
result = api(new_url, new_token, "/core/users/", method="POST", data={
|
|
"username": username,
|
|
"name": u["name"],
|
|
"email": u.get("email", ""),
|
|
"is_active": u["is_active"],
|
|
"groups": new_groups,
|
|
"path": u.get("path", "users"),
|
|
})
|
|
if result and "pk" in result:
|
|
# Set a temporary password
|
|
api(new_url, new_token, f"/core/users/{result['pk']}/set_password/",
|
|
method="POST", data={"password": "ChangeMeOnFirstLogin!2026"})
|
|
print(f" CREATED {username} (pk={result['pk']}, temp password set)")
|
|
else:
|
|
print(f" FAILED {username}")
|
|
|
|
|
|
def build_flow_map(new_url, new_token):
|
|
"""Map flow slugs to PKs on the new instance."""
|
|
flows = paginate(new_url, new_token, "/flows/instances/")
|
|
return {f["slug"]: f["pk"] for f in flows}
|
|
|
|
|
|
def build_scope_map(old_url, old_token, new_url, new_token):
|
|
"""Map old scope mapping PKs to new ones by scope_name."""
|
|
# Try both old and new API paths
|
|
old_scopes = paginate(old_url, old_token, "/propertymappings/scope/")
|
|
if not old_scopes:
|
|
old_scopes = paginate(old_url, old_token, "/propertymappings/provider/scope/")
|
|
|
|
new_scopes = paginate(new_url, new_token, "/propertymappings/provider/scope/")
|
|
if not new_scopes:
|
|
new_scopes = paginate(new_url, new_token, "/propertymappings/scope/")
|
|
|
|
old_by_scope = {s["scope_name"]: s["pk"] for s in old_scopes}
|
|
new_by_scope = {s["scope_name"]: s["pk"] for s in new_scopes}
|
|
|
|
scope_map = {} # old pk -> new pk
|
|
for scope_name, old_pk in old_by_scope.items():
|
|
if scope_name in new_by_scope:
|
|
scope_map[old_pk] = new_by_scope[scope_name]
|
|
|
|
return scope_map, new_by_scope
|
|
|
|
|
|
def build_signing_key_map(new_url, new_token):
|
|
"""Get the self-signed certificate PK on the new instance."""
|
|
certs = paginate(new_url, new_token, "/crypto/certificatekeypairs/")
|
|
for c in certs:
|
|
if "Self-signed" in c["name"]:
|
|
return c["pk"]
|
|
return None
|
|
|
|
|
|
def migrate_providers(old_url, old_token, new_url, new_token, dry_run):
|
|
"""Migrate OAuth2 providers with their client IDs and secrets."""
|
|
print("\n=== Migrating OAuth2 Providers ===")
|
|
|
|
flow_map = build_flow_map(new_url, new_token)
|
|
scope_map, new_scopes_by_name = build_scope_map(old_url, old_token, new_url, new_token)
|
|
signing_key = build_signing_key_map(new_url, new_token)
|
|
|
|
# Default flow mappings
|
|
implicit_flow = flow_map.get("default-provider-authorization-implicit-consent")
|
|
explicit_flow = flow_map.get("default-provider-authorization-explicit-consent")
|
|
invalidation_flow = flow_map.get("default-provider-invalidation-flow")
|
|
|
|
old_providers = paginate(old_url, old_token, "/providers/oauth2/")
|
|
new_providers = paginate(new_url, new_token, "/providers/oauth2/")
|
|
existing = {p["client_id"]: p["pk"] for p in new_providers}
|
|
|
|
# Skip proxy providers (not needed in K8s with ingress-nginx)
|
|
skip_names = {"Caddy docs.renner.uno"}
|
|
|
|
provider_map = {} # old pk -> new pk
|
|
|
|
for p in old_providers:
|
|
name = p["name"]
|
|
client_id = p["client_id"]
|
|
|
|
if name in skip_names:
|
|
print(f" SKIP (not needed) {name}")
|
|
continue
|
|
|
|
if client_id in existing:
|
|
provider_map[p["pk"]] = existing[client_id]
|
|
print(f" EXISTS {name}")
|
|
continue
|
|
|
|
# Map old authorization flow to new
|
|
old_auth_flow = p.get("authorization_flow")
|
|
# Use explicit consent for Home Assistant (user confirmation), implicit for the rest
|
|
auth_flow = implicit_flow
|
|
|
|
# Map property mappings (scopes)
|
|
new_mappings = []
|
|
for old_pk in p.get("property_mappings", []):
|
|
new_pk = scope_map.get(old_pk)
|
|
if new_pk:
|
|
new_mappings.append(new_pk)
|
|
# If no mappings were transferred, use standard openid+email+profile
|
|
if not new_mappings:
|
|
for scope in ("openid", "email", "profile"):
|
|
if scope in new_scopes_by_name:
|
|
new_mappings.append(new_scopes_by_name[scope])
|
|
|
|
provider_data = {
|
|
"name": name,
|
|
"client_id": client_id,
|
|
"client_secret": p.get("client_secret", ""),
|
|
"client_type": p.get("client_type", "confidential"),
|
|
"redirect_uris": p.get("redirect_uris", []),
|
|
"authorization_flow": auth_flow,
|
|
"invalidation_flow": invalidation_flow,
|
|
"property_mappings": new_mappings,
|
|
"sub_mode": p.get("sub_mode", "hashed_user_id"),
|
|
}
|
|
|
|
# Add signing key if the old provider had one
|
|
if p.get("signing_key") and signing_key:
|
|
provider_data["signing_key"] = signing_key
|
|
|
|
if dry_run:
|
|
print(f" DRY-RUN would create provider: {name} (client_id={client_id[:12]}...)")
|
|
continue
|
|
|
|
result = api(new_url, new_token, "/providers/oauth2/", method="POST", data=provider_data)
|
|
if result and "pk" in result:
|
|
provider_map[p["pk"]] = result["pk"]
|
|
print(f" CREATED {name} (pk={result['pk']})")
|
|
else:
|
|
print(f" FAILED {name}")
|
|
|
|
return provider_map
|
|
|
|
|
|
def migrate_applications(old_url, old_token, new_url, new_token, provider_map, dry_run):
|
|
"""Migrate applications linked to providers."""
|
|
print("\n=== Migrating Applications ===")
|
|
|
|
old_apps = paginate(old_url, old_token, "/core/applications/?superuser_full_list=true")
|
|
new_apps = paginate(new_url, new_token, "/core/applications/?superuser_full_list=true")
|
|
existing = {a["slug"]: a for a in new_apps}
|
|
|
|
for a in old_apps:
|
|
slug = a["slug"]
|
|
name = a["name"]
|
|
|
|
# Fix known slug typos
|
|
if slug == "gittea":
|
|
slug = "gitea"
|
|
|
|
if slug in existing:
|
|
# Update icon if the old app has one and the new one doesn't
|
|
old_icon = a.get("meta_icon", "")
|
|
new_icon = existing[slug].get("meta_icon", "")
|
|
if old_icon and old_icon.startswith("http") and not new_icon:
|
|
if not dry_run:
|
|
api(new_url, new_token, f"/core/applications/{slug}/",
|
|
method="PATCH", data={"meta_icon": old_icon})
|
|
print(f" UPDATED icon for {slug}")
|
|
else:
|
|
print(f" DRY-RUN would update icon for {slug}")
|
|
else:
|
|
print(f" EXISTS {slug}")
|
|
continue
|
|
|
|
# Map old provider PK to new
|
|
old_provider_pk = a.get("provider")
|
|
new_provider_pk = provider_map.get(old_provider_pk) if old_provider_pk else None
|
|
|
|
if old_provider_pk and not new_provider_pk:
|
|
print(f" SKIP {slug} (provider not migrated)")
|
|
continue
|
|
|
|
# Only copy external icon URLs, skip local file references
|
|
icon = a.get("meta_icon", "")
|
|
if icon and not icon.startswith("http"):
|
|
icon = ""
|
|
|
|
app_data = {
|
|
"name": name,
|
|
"slug": slug,
|
|
"meta_launch_url": a.get("meta_launch_url", ""),
|
|
"meta_icon": icon,
|
|
"meta_description": a.get("meta_description", ""),
|
|
"open_in_new_tab": a.get("open_in_new_tab", False),
|
|
}
|
|
if new_provider_pk:
|
|
app_data["provider"] = new_provider_pk
|
|
|
|
if dry_run:
|
|
icon_info = f" icon={icon}" if icon else ""
|
|
print(f" DRY-RUN would create application: {slug}{icon_info}")
|
|
continue
|
|
|
|
result = api(new_url, new_token, "/core/applications/", method="POST", data=app_data)
|
|
if result and "slug" in result:
|
|
print(f" CREATED {slug}")
|
|
else:
|
|
print(f" FAILED {slug}")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Migrate Authentik config between instances")
|
|
parser.add_argument("--old-url", required=True, help="Old Authentik base URL")
|
|
parser.add_argument("--old-token", required=True, help="Old Authentik API token")
|
|
parser.add_argument("--new-url", required=True, help="New Authentik base URL")
|
|
parser.add_argument("--new-token", required=True, help="New Authentik API token")
|
|
parser.add_argument("--dry-run", action="store_true", help="Show what would be done without making changes")
|
|
args = parser.parse_args()
|
|
|
|
# Verify connectivity
|
|
print("Verifying API access...")
|
|
old_config = api(args.old_url, args.old_token, "/root/config/")
|
|
new_config = api(args.new_url, args.new_token, "/root/config/")
|
|
if not old_config or not new_config:
|
|
print("ERROR: Could not connect to one or both instances", file=sys.stderr)
|
|
sys.exit(1)
|
|
print(f" Old instance: OK")
|
|
print(f" New instance: OK")
|
|
|
|
if args.dry_run:
|
|
print("\n*** DRY RUN MODE — no changes will be made ***")
|
|
|
|
group_map = migrate_groups(args.old_url, args.old_token, args.new_url, args.new_token, args.dry_run)
|
|
migrate_users(args.old_url, args.old_token, args.new_url, args.new_token, group_map, args.dry_run)
|
|
provider_map = migrate_providers(args.old_url, args.old_token, args.new_url, args.new_token, args.dry_run)
|
|
migrate_applications(args.old_url, args.old_token, args.new_url, args.new_token, provider_map, args.dry_run)
|
|
|
|
print("\n=== Migration Complete ===")
|
|
if not args.dry_run:
|
|
print("NOTE: All migrated users have temporary password 'ChangeMeOnFirstLogin!2026'")
|
|
print(" Users should change their passwords on first login.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|