102 lines
4.2 KiB
Python
102 lines
4.2 KiB
Python
import json
|
|
import requests
|
|
|
|
# Authentik API credentials and endpoints
|
|
authentik_url = 'https://auth.delmar.bzh'
|
|
authentik_api_key = 'wsH0RN4wHs6cgsTKRrOPUTpZkXXSr8NwDyhBSMDjNFCjSMiyoOXjSfjjRE2t'
|
|
authentik_username = 'akadmin'
|
|
|
|
# Read the contents of nextcloud_user_info.json
|
|
print("Reading nextcloud_user_info.json...")
|
|
with open('nextcloud_user_info.json', 'r') as file:
|
|
user_info = json.load(file)
|
|
|
|
# Extract all groups from user records
|
|
all_groups = set()
|
|
for user_record in user_info:
|
|
all_groups.update(user_record["groups"])
|
|
|
|
# Create a session
|
|
session = requests.Session()
|
|
|
|
# Define headers
|
|
headers = {
|
|
'Accept': 'application/json',
|
|
'Authorization': 'Bearer ' + authentik_api_key,
|
|
}
|
|
|
|
# Fetch existing groups from Authentik
|
|
print("Fetching existing groups from Authentik...")
|
|
authentik_groups_url = f'{authentik_url}/api/v3/core/groups/' # Updated API endpoint
|
|
response = session.get(authentik_groups_url, headers=headers)
|
|
print(f"Groups URL ({authentik_groups_url}")
|
|
|
|
# Check for successful response and decode JSON if present
|
|
if response.status_code == 200:
|
|
try:
|
|
authentik_groups = response.json()
|
|
print("Existing groups fetched successfully.")
|
|
except json.JSONDecodeError as e:
|
|
print(f"Failed to decode JSON response: {e}")
|
|
authentik_groups = []
|
|
elif response.status_code == 404:
|
|
print("The requested resource could not be found. Please check the URL or the resource.")
|
|
else:
|
|
print(f"Failed to fetch existing groups. Status Code: {response.status_code}")
|
|
authentik_groups = []
|
|
|
|
# Check if groups from Nextcloud exist in Authentik, create missing groups
|
|
missing_groups = all_groups - set(authentik_groups)
|
|
for group in missing_groups:
|
|
group_data = {"name": group}
|
|
create_group_response = session.post(authentik_groups_url, json=group_data, headers=headers)
|
|
if create_group_response.status_code == 201:
|
|
print(f"Group '{group}' created successfully.")
|
|
else:
|
|
print(f"Failed to create group '{group}'. Status Code: {create_group_response.status_code}")
|
|
|
|
# Fetch existing groups from Authentik
|
|
print("Fetching existing groups from Authentik...")
|
|
response = session.get(authentik_groups_url, headers=headers)
|
|
authentik_groups = response.json()
|
|
#print(authentik_groups)
|
|
|
|
# Create a dictionary mapping group names to group UUIDs
|
|
group_name_to_uuid = {group['name']: group['pk'] for group in authentik_groups['results']}
|
|
print(group_name_to_uuid)
|
|
|
|
# Import users to Authentik
|
|
print(f"Importing users to Authentik ({authentik_url})...")
|
|
for idx, user_record in enumerate(user_info, start=1):
|
|
print(f"Processing user {idx}/{len(user_info)} - ID: {user_record['user_id']}")
|
|
|
|
# Map Nextcloud groups to Authentik group UUIDs
|
|
authentik_group_uuids = [group_name_to_uuid[group] for group in user_record['groups'] if group in group_name_to_uuid]
|
|
|
|
|
|
# Format the user record according to Authentik's requirements
|
|
formatted_user = {
|
|
"username": user_record["user_id"],
|
|
"name": user_record["display_name"],
|
|
"email": user_record["email"],
|
|
"groups": authentik_group_uuids,
|
|
"is_active": True, # Assuming all users are active
|
|
"type": "internal", # Assuming all users are internal
|
|
# Add other fields as required by Authentik's API
|
|
# Map or transform data fields according to Authentik's expected format
|
|
}
|
|
|
|
# Send the formatted user data to Authentik's API endpoint for user import
|
|
user_import_url = f"{authentik_url}/api/v3/core/users/" # Updated API endpoint
|
|
user_import_response = session.post(user_import_url, json=formatted_user, headers=headers )
|
|
|
|
# Check response status or handle errors based on the API response
|
|
if user_import_response.status_code == 201:
|
|
print(f"User {formatted_user['username']} imported successfully.")
|
|
elif user_import_response.status_code == 400 and user_import_response.json().get('username') == ["This field must be unique."]:
|
|
print(f"User {formatted_user['username']} already exists. Skipping.")
|
|
else:
|
|
print(f"Failed to import user {formatted_user['username']}. Status Code: {user_import_response.status_code}")
|
|
# Print response content for debugging (if needed)
|
|
print(user_import_response.content)
|