import sys import random import re import os import time import urllib.parse import requests from cw_rpa import Logger, Input, HttpClient, ResultLevel sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) log = Logger() http_client = HttpClient() input = Input() log.info("Imports completed successfully") msgraph_base_url_base = "https://graph.microsoft.com" msgraph_base_url_path = "/v1.0" msgraph_base_url_beta_base = "https://graph.microsoft.com" msgraph_base_url_beta_path = "/beta" graph_integration_name = "azure_o365" data_to_log = {} bot_name = "M365 - Transfer OneDrive Ownership" log.info("Static variables set") def record_result(log, level, message): log.result_message(level, f"[{bot_name}]: {message}") if level in (ResultLevel.WARNING, ResultLevel.ERROR): data_to_log["status_result"] = "Fail" elif level == ResultLevel.SUCCESS: if "status_result" not in data_to_log or data_to_log["status_result"] != "Fail": data_to_log["status_result"] = "Success" def execute_api_call(log, http_client, method, endpoint, data=None, retries=5, integration_name=None, headers=None, params=None): base_delay = 5 log.info(f"Executing API call: {method.upper()} {endpoint}") for attempt in range(retries): try: if integration_name: response = ( getattr(http_client.third_party_integration(integration_name), method)(url=endpoint, json=data) if data else getattr(http_client.third_party_integration(integration_name), method)(url=endpoint) ) else: request_args = {"url": endpoint} if params: request_args["params"] = params if headers: request_args["headers"] = headers if data: if (headers and headers.get("Content-Type") == "application/x-www-form-urlencoded"): request_args["data"] = data else: request_args["json"] = data response = getattr(requests, method)(**request_args) if 200 <= response.status_code < 300: return response elif response.status_code in [429, 503]: retry_after = response.headers.get("Retry-After") wait_time = int(retry_after) if retry_after else base_delay * (2 ** attempt) + random.uniform(0, 3) log.warning(f"Rate limit exceeded. Retrying in {wait_time:.2f} seconds") time.sleep(wait_time) elif 400 <= response.status_code < 500: if response.status_code == 404: log.warning(f"Skipping non-existent resource [{endpoint}]") return None log.error(f"Client error Status: {response.status_code}, Response: {response.text}") return response elif 500 <= response.status_code < 600: log.warning(f"Server error Status: {response.status_code}, attempt {attempt + 1} of {retries}, Response: {getattr(response, 'text', '')[:1000]}") time.sleep(base_delay * (2 ** attempt) + random.uniform(0, 3)) else: log.error(f"Unexpected response Status: {response.status_code}, Response: {response.text}") return response except Exception as e: log.exception(e, f"Exception during API call to {endpoint}") return None return None def get_aad_user_data(log, http_client, msgraph_base_url_base, msgraph_base_url_path, user_identifier): log.info(f"Resolving user ID and email for [{user_identifier}]") if re.fullmatch(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}", user_identifier): endpoint = f"{msgraph_base_url_base}{msgraph_base_url_path}/users/{user_identifier}" response = execute_api_call(log, http_client, "get", endpoint, integration_name=graph_integration_name) if response: user = response.json() return user.get("id", ""), user.get("userPrincipalName", ""), user.get("onPremisesSamAccountName", ""), user.get("onPremisesSyncEnabled", False) return "", "", "", False filters = [ f"startswith(displayName,'{user_identifier}')", f"startswith(userPrincipalName,'{user_identifier}')", f"startswith(mail,'{user_identifier}')" ] filter_query = " or ".join(filters) endpoint = f"{msgraph_base_url_base}{msgraph_base_url_path}/users?$filter={urllib.parse.quote(filter_query)}" response = execute_api_call(log, http_client, "get", endpoint, integration_name=graph_integration_name) if response: users = response.json().get("value", []) if len(users) > 1: log.error(f"Multiple users found for [{user_identifier}]") return users if users: user = users[0] return user.get("id", ""), user.get("userPrincipalName", ""), user.get("onPremisesSamAccountName", ""), user.get("onPremisesSyncEnabled", False) return "", "", "", False def get_user_drive(log, http_client, msgraph_base_url_base, msgraph_base_url_path, user_id): log.info(f"Retrieving OneDrive for user [{user_id}]") endpoint = f"{msgraph_base_url_base}{msgraph_base_url_path}/users/{user_id}/drive" response = execute_api_call(log, http_client, "get", endpoint, integration_name=graph_integration_name) if response and response.status_code < 300: drive = response.json() drive_id = drive.get("id", "") drive_web_url = drive.get("webUrl", "") quota = drive.get("quota", {}) or {} quota_total = int(quota.get("total", 0) or 0) quota_used = int(quota.get("used", 0) or 0) quota_remaining = int(quota.get("remaining", 0) or 0) log.info(f"Resolved drive ID [{drive_id}] webUrl [{drive_web_url}] total [{quota_total}] used [{quota_used}] remaining [{quota_remaining}]") return drive_id, drive_web_url, quota_total, quota_used, quota_remaining return "", "", 0, 0, 0 def get_drive_root_children(log, http_client, msgraph_base_url_base, msgraph_base_url_path, user_id): log.info(f"Listing root children of OneDrive for user [{user_id}]") items = [] endpoint = f"{msgraph_base_url_base}{msgraph_base_url_path}/users/{user_id}/drive/root/children?$select=id,name,size,folder,file&$top=200" while endpoint: response = execute_api_call(log, http_client, "get", endpoint, integration_name=graph_integration_name) if not response or response.status_code >= 300: return [] data = response.json() items.extend(data.get("value", [])) endpoint = data.get("@odata.nextLink", "") log.info(f"Retrieved [{len(items)}] root items from source OneDrive") return items def get_drive_items_recursive(log, http_client, msgraph_base_url_base, msgraph_base_url_path, user_id): log.info(f"Performing recursive listing of OneDrive for user [{user_id}]") results = [] pending = [("", "/")] while pending: folder_id, folder_path = pending.pop() if folder_id: endpoint = f"{msgraph_base_url_base}{msgraph_base_url_path}/users/{user_id}/drive/items/{folder_id}/children?$select=id,name,size,folder,file,parentReference&$top=200" else: endpoint = f"{msgraph_base_url_base}{msgraph_base_url_path}/users/{user_id}/drive/root/children?$select=id,name,size,folder,file,parentReference&$top=200" while endpoint: response = execute_api_call(log, http_client, "get", endpoint, integration_name=graph_integration_name) if not response or response.status_code >= 300: log.error(f"Failed to enumerate folder [{folder_path}]") return results data = response.json() for child in data.get("value", []): child_name = child.get("name", "") child_id = child.get("id", "") child_size = int(child.get("size", 0) or 0) is_folder = "folder" in child and child.get("folder") is not None child_path = f"{folder_path}{child_name}{'/' if is_folder else ''}" entry = { "id": child_id, "name": child_name, "path": child_path, "size": child_size, "type": "folder" if is_folder else "file" } if is_folder: entry["childCount"] = int(child.get("folder", {}).get("childCount", 0) or 0) results.append(entry) if is_folder and child_id: pending.append((child_id, child_path)) endpoint = data.get("@odata.nextLink", "") log.info(f"Recursive listing complete. Total items: [{len(results)}]") return results def get_drive_root_permissions(log, http_client, msgraph_base_url_base, msgraph_base_url_path, user_id): log.info(f"Retrieving root permissions for OneDrive of user [{user_id}]") permissions = [] endpoint = f"{msgraph_base_url_base}{msgraph_base_url_path}/users/{user_id}/drive/root/permissions?$top=200" while endpoint: response = execute_api_call(log, http_client, "get", endpoint, integration_name=graph_integration_name) if not response or response.status_code >= 300: return permissions data = response.json() for p in data.get("value", []): granted_to = p.get("grantedToV2", {}) or p.get("grantedTo", {}) or {} granted_to_identities = p.get("grantedToIdentitiesV2", []) or p.get("grantedToIdentities", []) or [] holders = [] if granted_to: user = granted_to.get("user", {}) or {} if user: holders.append({"displayName": user.get("displayName", ""), "email": user.get("email", ""), "id": user.get("id", "")}) for g in granted_to_identities: user = g.get("user", {}) or {} if user: holders.append({"displayName": user.get("displayName", ""), "email": user.get("email", ""), "id": user.get("id", "")}) permissions.append({ "id": p.get("id", ""), "roles": p.get("roles", []), "link": (p.get("link", {}) or {}).get("webUrl", ""), "linkType": (p.get("link", {}) or {}).get("type", ""), "linkScope": (p.get("link", {}) or {}).get("scope", ""), "inheritedFrom": p.get("inheritedFrom", {}), "holders": holders }) endpoint = data.get("@odata.nextLink", "") log.info(f"Retrieved [{len(permissions)}] root permissions") return permissions def grant_drive_access(log, http_client, msgraph_base_url_base, msgraph_base_url_path, source_user_id, target_user_email): log.info(f"Granting write access on [{source_user_id}] OneDrive to [{target_user_email}]") endpoint = f"{msgraph_base_url_base}{msgraph_base_url_path}/users/{source_user_id}/drive/root/invite" payload = { "requireSignIn": True, "sendInvitation": False, "roles": ["write"], "recipients": [{"email": target_user_email}] } response = execute_api_call(log, http_client, "post", endpoint, data=payload, integration_name=graph_integration_name) if response and 200 <= response.status_code < 300: permissions = response.json().get("value", []) permission_ids = [p.get("id", "") for p in permissions] log.info(f"Successfully granted access. Permission IDs: {permission_ids}") return True, permission_ids if response is not None: log.error(f"Failed to grant drive access Status: {response.status_code}, Response: {response.text}") return False, [] def ensure_target_folder(log, http_client, msgraph_base_url_base, msgraph_base_url_path, target_user_id, folder_name): log.info(f"Ensuring destination folder [{folder_name}] exists in target OneDrive [{target_user_id}]") encoded_name = urllib.parse.quote(folder_name) get_endpoint = f"{msgraph_base_url_base}{msgraph_base_url_path}/users/{target_user_id}/drive/root:/{encoded_name}" response = execute_api_call(log, http_client, "get", get_endpoint, integration_name=graph_integration_name) if response and 200 <= response.status_code < 300: folder = response.json() folder_id = folder.get("id", "") log.info(f"Destination folder already exists [{folder_id}]") return folder_id create_endpoint = f"{msgraph_base_url_base}{msgraph_base_url_path}/users/{target_user_id}/drive/root/children" payload = {"name": folder_name, "folder": {}, "@microsoft.graph.conflictBehavior": "rename"} response = execute_api_call(log, http_client, "post", create_endpoint, data=payload, integration_name=graph_integration_name) if response and 200 <= response.status_code < 300: folder = response.json() folder_id = folder.get("id", "") log.info(f"Created destination folder [{folder_id}]") return folder_id if response is not None: log.error(f"Failed to create destination folder Status: {response.status_code}, Response: {response.text}") return "" def copy_item_async(log, http_client, msgraph_base_url_base, msgraph_base_url_path, source_user_id, source_item_id, target_drive_id, target_folder_id, item_name): endpoint = f"{msgraph_base_url_base}{msgraph_base_url_path}/users/{source_user_id}/drive/items/{source_item_id}/copy" payload = { "parentReference": {"driveId": target_drive_id, "id": target_folder_id}, "name": item_name, "@microsoft.graph.conflictBehavior": "rename" } response = execute_api_call(log, http_client, "post", endpoint, data=payload, integration_name=graph_integration_name) if response is not None and response.status_code == 202: monitor_url = response.headers.get("Location", "") log.info(f"Queued copy for [{item_name}] monitor [{monitor_url}]") return True, monitor_url if response is not None: log.error(f"Failed to queue copy for [{item_name}] Status: {response.status_code}, Response: {response.text}") return False, "" def main(): try: log.info("Bot execution has started.") try: operation = input.get_value("Operation") source_user_identifier = input.get_value("SourceUser") target_user_identifier = input.get_value("TargetUser") except Exception: record_result(log, ResultLevel.ERROR, "Failed to fetch input values") return operation = operation.strip() if operation else "" source_user_identifier = source_user_identifier.strip() if source_user_identifier else "" target_user_identifier = target_user_identifier.strip() if target_user_identifier else "" log.info(f"Requested operation = [{operation}]") log.info(f"Source user (offboarded) = [{source_user_identifier}]") log.info(f"Target user (recipient) = [{target_user_identifier}]") if not operation: record_result(log, ResultLevel.WARNING, "Operation value is missing or invalid") return if operation not in ("GetInfo", "GrantAccess", "CopyFiles", "GrantAndCopy"): record_result(log, ResultLevel.WARNING, f"Unknown operation [{operation}]. Expected one of: GetInfo, GrantAccess, CopyFiles, GrantAndCopy") return if not source_user_identifier: record_result(log, ResultLevel.WARNING, "Source user identifier is empty or invalid") return if operation != "GetInfo" and not target_user_identifier: record_result(log, ResultLevel.WARNING, "Target user identifier is empty or invalid") return source_user_result = get_aad_user_data(log, http_client, msgraph_base_url_base, msgraph_base_url_path, source_user_identifier) if isinstance(source_user_result, list): details = "\n".join([f"- {u.get('displayName')} | {u.get('userPrincipalName')} | {u.get('id')}" for u in source_user_result]) record_result(log, ResultLevel.ERROR, f"Multiple users found for source [{source_user_identifier}]\n{details}") return source_user_id, source_user_email, source_user_sam, source_user_onpremisessyncenabled = source_user_result if not source_user_id: record_result(log, ResultLevel.ERROR, f"Failed to resolve user ID for source [{source_user_identifier}]") return if not source_user_email: record_result(log, ResultLevel.ERROR, f"Unable to resolve user principal name for source [{source_user_identifier}]") return if operation == "GetInfo": log.info("Executing operation: GetInfo") source_drive_id, source_drive_web_url, source_quota_total, source_quota_used, source_quota_remaining = get_user_drive(log, http_client, msgraph_base_url_base, msgraph_base_url_path, source_user_id) if not source_drive_id: record_result(log, ResultLevel.ERROR, f"Source user [{source_user_email}] has no OneDrive or it is not provisioned") return permissions = get_drive_root_permissions(log, http_client, msgraph_base_url_base, msgraph_base_url_path, source_user_id) items = get_drive_items_recursive(log, http_client, msgraph_base_url_base, msgraph_base_url_path, source_user_id) file_count = sum(1 for i in items if i.get("type") == "file") folder_count = sum(1 for i in items if i.get("type") == "folder") total_bytes = sum(int(i.get("size", 0) or 0) for i in items if i.get("type") == "file") data_to_log["drive"] = { "id": source_drive_id, "webUrl": source_drive_web_url, "quotaTotal": source_quota_total, "quotaUsed": source_quota_used, "quotaRemaining": source_quota_remaining, "owner": source_user_email } data_to_log["permissions"] = permissions data_to_log["items"] = items data_to_log["item_count"] = len(items) data_to_log["file_count"] = file_count data_to_log["folder_count"] = folder_count data_to_log["total_bytes"] = total_bytes record_result(log, ResultLevel.SUCCESS, f"OneDrive info for [{source_user_email}]: [{file_count}] files, [{folder_count}] folders, [{total_bytes}] bytes, [{len(permissions)}] root permissions. URL [{source_drive_web_url}]") return target_user_result = get_aad_user_data(log, http_client, msgraph_base_url_base, msgraph_base_url_path, target_user_identifier) if isinstance(target_user_result, list): details = "\n".join([f"- {u.get('displayName')} | {u.get('userPrincipalName')} | {u.get('id')}" for u in target_user_result]) record_result(log, ResultLevel.ERROR, f"Multiple users found for target [{target_user_identifier}]\n{details}") return target_user_id, target_user_email, target_user_sam, target_user_onpremisessyncenabled = target_user_result if not target_user_id: record_result(log, ResultLevel.ERROR, f"Failed to resolve user ID for target [{target_user_identifier}]") return if not target_user_email: record_result(log, ResultLevel.ERROR, f"Unable to resolve user principal name for target [{target_user_identifier}]") return if source_user_id == target_user_id: record_result(log, ResultLevel.WARNING, "Source and target users are the same") return source_drive_id, source_drive_web_url, source_quota_total, source_quota_used, source_quota_remaining = get_user_drive(log, http_client, msgraph_base_url_base, msgraph_base_url_path, source_user_id) if not source_drive_id: record_result(log, ResultLevel.ERROR, f"Source user [{source_user_email}] has no OneDrive or it is not provisioned") return access_granted = False if operation in ("GrantAccess", "GrantAndCopy"): log.info("Executing operation: GrantAccess") granted, permission_ids = grant_drive_access(log, http_client, msgraph_base_url_base, msgraph_base_url_path, source_user_id, target_user_email) if not granted: record_result(log, ResultLevel.ERROR, f"Failed to grant [{target_user_email}] access to OneDrive of [{source_user_email}]") return access_granted = True if operation == "GrantAccess": record_result(log, ResultLevel.SUCCESS, f"Granted [{target_user_email}] write access to OneDrive of [{source_user_email}] at [{source_drive_web_url}]") return if operation in ("CopyFiles", "GrantAndCopy"): log.info("Executing operation: CopyFiles") target_drive_id, target_drive_web_url, target_quota_total, target_quota_used, target_quota_remaining = get_user_drive(log, http_client, msgraph_base_url_base, msgraph_base_url_path, target_user_id) if not target_drive_id: record_result(log, ResultLevel.ERROR, f"Target user [{target_user_email}] has no OneDrive or it is not provisioned") return root_items = get_drive_root_children(log, http_client, msgraph_base_url_base, msgraph_base_url_path, source_user_id) item_count = len(root_items) total_bytes = sum(int(i.get("size", 0) or 0) for i in root_items) log.info(f"Source OneDrive root items [{item_count}] total size [{total_bytes}] bytes target remaining quota [{target_quota_remaining}] bytes") if item_count == 0: record_result(log, ResultLevel.SUCCESS, f"Source OneDrive of [{source_user_email}] is empty. Nothing to copy. Access granted = [{access_granted}]") return if target_quota_remaining and total_bytes > target_quota_remaining: record_result(log, ResultLevel.ERROR, f"Target user [{target_user_email}] has insufficient quota. Required [{total_bytes}] bytes, available [{target_quota_remaining}] bytes across [{item_count}] root items") return folder_name = f"Archived from {source_user_email}" target_folder_id = ensure_target_folder(log, http_client, msgraph_base_url_base, msgraph_base_url_path, target_user_id, folder_name) if not target_folder_id: record_result(log, ResultLevel.ERROR, f"Failed to create or resolve destination folder [{folder_name}] in target OneDrive") return queued = 0 failed = 0 monitors = [] for item in root_items: item_id = item.get("id", "") item_name = item.get("name", "") if not item_id or not item_name: failed += 1 continue ok, monitor_url = copy_item_async(log, http_client, msgraph_base_url_base, msgraph_base_url_path, source_user_id, item_id, target_drive_id, target_folder_id, item_name) if ok: queued += 1 if monitor_url: monitors.append(monitor_url) else: failed += 1 data_to_log["copy_items_total"] = item_count data_to_log["copy_items_queued"] = queued data_to_log["copy_items_failed"] = failed data_to_log["copy_total_bytes"] = total_bytes data_to_log["copy_destination_folder"] = folder_name data_to_log["copy_monitor_urls"] = monitors if failed and queued == 0: record_result(log, ResultLevel.ERROR, f"Failed to queue any of [{item_count}] root items for copy from [{source_user_email}] to [{target_user_email}]") return if failed: record_result(log, ResultLevel.WARNING, f"Queued [{queued}] of [{item_count}] root items ([{failed}] failed) from [{source_user_email}] into [{target_user_email}]/[{folder_name}]. Access granted = [{access_granted}]") return record_result(log, ResultLevel.SUCCESS, f"Queued [{queued}] root items ([{total_bytes}] bytes) from [{source_user_email}] into [{target_user_email}]/[{folder_name}]. Access granted = [{access_granted}]. Copies run asynchronously in Microsoft 365") return except Exception as e: log.error(f"Unhandled error in main: {str(e)}") record_result(log, ResultLevel.ERROR, "Unhandled exception occurred during execution") finally: log.result_data(data_to_log) if __name__ == "__main__": main()