From 8f7738f92df5ffe2584ef9610de716d4121f3817 Mon Sep 17 00:00:00 2001 From: Enrico Ludwig Date: Thu, 18 Jul 2024 12:09:34 +0200 Subject: [PATCH] Implemented user migration --- gitlab2gitea/gitlab2gitea.py | 233 ++++++++++++++++++++++++++++++++++- 1 file changed, 229 insertions(+), 4 deletions(-) diff --git a/gitlab2gitea/gitlab2gitea.py b/gitlab2gitea/gitlab2gitea.py index 0c2d796..494788a 100644 --- a/gitlab2gitea/gitlab2gitea.py +++ b/gitlab2gitea/gitlab2gitea.py @@ -75,6 +75,10 @@ LOG_FILE = "gitlab2gitea.log" APPEND_LOG = False QUIET = False +# Internal variables - Do not change + +GITEA_RESERVED_USERNAMES = ["ghost", "notifications"] + # Imports import os @@ -396,6 +400,10 @@ def _exception(exception, custom_message=None): # PROGRAM +def is_gitea_reserved_username(username: str) -> bool: + return username in GITEA_RESERVED_USERNAMES + + def gitlab2gitea_visibility(visibility: str) -> str: if visibility == "private": return "private" @@ -589,7 +597,51 @@ def get_gitlab_projects() -> list: # Endpoint: POST /api/{GITEA_API_VERSION}/admin/users def migrate_gitlab_user_to_gitea(user: dict): - pass + + if not user: + raise Exception("User is missing!") + + # Create Gitea user + + _debug(f"REQUEST: POST {GITEA_URL}/api/{GITEA_API_VERSION}/admin/users") + + response = requests.post( + f"{GITEA_URL}/api/{GITEA_API_VERSION}/admin/users", + json={ + "login_name": user["username"], + "username": user["username"], + "email": user["email"], + "full_name": user["name"], + "password": "12345678", # TODO: Change to random password which will be sent to the user + "send_notify": False, # TODO: Change to True as soon as the password is sent to the user + "must_change_password": True, + "admin": user["is_admin"], + }, + headers={ + "Content-Type": "application/json", + "Accept": "application/json", + "Authorization": f"token {GITEA_TOKEN}", + }, + ) + + _trace(f"RESPONSE: {response.json()}") + + if response.status_code != 201: + response_message = ( + response.json()["message"] + if "message" in response.json() + else "Unknown error" + ) + raise Exception(f"Failed to create Gitea user: {response_message}") + else: + user = response.json() + + if user["is_admin"]: + _info(f'Admin user "{user["username"]}" created on Gitea') + else: + _info(f'User "{user["username"]}" created on Gitea') + + return user # Endpoint: GET /api/{GITEA_API_VERSION}/orgs @@ -760,6 +812,122 @@ def update_gitea_org(data: dict) -> dict: return group +# Endpoint: GET /api/{GITEA_API_VERSION}/users/{username} +def get_gitea_user(username: str) -> dict: + + _debug(f"REQUEST: GET {GITEA_URL}/api/{GITEA_API_VERSION}/users/{username}") + + response = requests.get( + f"{GITEA_URL}/api/{GITEA_API_VERSION}/users/{username}", + headers={ + "Content-Type": "application/json", + "Accept": "application/json", + "Authorization": f"token {GITEA_TOKEN}", + }, + ) + + _trace(f"RESPONSE: {response.json()}") + + if response.status_code != 200: + response_message = ( + response.json()["message"] + if "message" in response.json() + else "Unknown error" + ) + raise Exception(f"Failed to get Gitea user: {response_message}") + else: + user = response.json() + + return user + + +# Endpoint: PATCH /api/{GITEA_API_VERSION}/admin/users/{username} +def update_gitea_user(data: dict) -> dict: + + if not data: + raise Exception("Data is missing!") + + username = data["username"] + current_user = get_gitea_user(username) + updated_user = convert_gitlab_user_to_gitea(data, current_user) + + if is_gitea_reserved_username(username): + _warn(f'User "{username}" is a reserved username on Gitea!') + return None + + if not current_user: + raise Exception(f'User "{username}" not found on Gitea!') + + _debug(f"REQUEST: PATCH {GITEA_URL}/api/{GITEA_API_VERSION}/admin/users/{username}") + + response = requests.patch( + f"{GITEA_URL}/api/{GITEA_API_VERSION}/admin/users/{username}", + json=updated_user, + headers={ + "Content-Type": "application/json", + "Accept": "application/json", + "Authorization": f"token {GITEA_TOKEN}", + }, + ) + + _trace(f"RESPONSE: {response.json()}") + + if response.status_code != 200: + response_message = ( + response.json()["message"] + if "message" in response.json() + else "Unknown error" + ) + raise Exception(f"Failed to update Gitea user: {response_message}") + else: + user = response.json() + + return user + + +def cmp_gitea_userdata(userdata_a: dict, userdata_b: dict) -> bool: + + result = {} + + for key in userdata_a: + if key in userdata_b: + if userdata_a[key] != userdata_b[key]: + result[key] = True + else: + result[key] = False + else: + result[key] = True + + has_changes = False + for key in result: + if result[key]: + has_changes = True + break + + return has_changes, result + + +def convert_gitlab_user_to_gitea(user: dict, extra_data: dict = None) -> dict: + + gitea_user = { + "active": user["state"] == "active", + "login_name": user["username"], + "avatar_url": user["avatar_url"], + "created": user["created_at"], + "description": user["bio"], + "email": user["email"], + "full_name": user["name"], + "is_admin": user["is_admin"], + "prohibit_login": user["state"] == "blocked" or user["locked"], + "website": user["website_url"], + } + + if extra_data: + gitea_user.update(extra_data) + + return gitea_user + + def cmp_gitlab_gitea_groups(gitlab_groups: list, gitea_groups: list) -> dict: # 0 = exists on both @@ -1031,7 +1199,37 @@ def create_missing_groups(gitlab_groups: list, gitea_groups: list): def create_missing_users(gitlab_users: list, gitea_users: list): - pass + + for gitlab_user in gitlab_users: + name = gitlab_user["username"] + exists = False + + for gitea_user in gitea_users["data"]: + if name == gitea_user["login"]: + exists = True + break + + if gitea_user["email"] == None or gitea_user["email"] == "": + _warn( + f'User "{name}" does not have an email address and will not be created!' + ) + continue + + if gitea_user["username"] == None or gitea_user["username"] == "": + _warn(f'User "{name}" does not have a username and will not be created!') + continue + + if is_gitea_reserved_username(name): + _warn(f'User "{name}" is a reserved username on Gitea!') + continue + + if not exists: + _info(f'Creating missing user "{name}" on Gitea...') + + try: + migrate_gitlab_user_to_gitea(gitlab_user) + except Exception as e: + _exception(f'Failed to create Gitea user "{name}": {e}', e) def create_missing_projects(gitlab_projects: list, gitea_projects: list): @@ -1053,8 +1251,8 @@ def update_existing_groups(gitlab_groups: list, gitea_groups: list): try: update_gitea_org( - name, { + "path": gitlab_group["path"], "description": gitlab_group["description"], "website": gitlab_group["web_url"], "visibility": gitlab2gitea_visibility( @@ -1068,7 +1266,34 @@ def update_existing_groups(gitlab_groups: list, gitea_groups: list): def update_existing_users(gitlab_users: list, gitea_users: list): - pass + + for gitlab_user in gitlab_users: + name = gitlab_user["username"] + exists = False + + for gitea_user in gitea_users["data"]: + if name == gitea_user["login"]: + exists = True + break + + if gitea_user["email"] == None or gitea_user["email"] == "": + _warn( + f'User "{name}" does not have an email address and will not be updated!' + ) + continue + + if gitea_user["username"] == None or gitea_user["username"] == "": + _warn(f'User "{name}" does not have a username and will not be updated!') + continue + + if exists: + _info(f'Updating existing user "{name}" on Gitea...') + try: + update_gitea_user(gitlab_user) + except Exception as e: + _exception(f'Failed to update Gitea user "{name}": {e}', e) + else: + _warn(f'User "{name}" does not exist on Gitea!') def update_existing_projects(gitlab_projects: list, gitea_projects: list):