MSP Usecase: User Onboarding Azure + M365

import requests from azure.identity import ClientSecretCredential # Set up Azure AD credentials tenant_id = 'your-tenant-id' client_id = "your-client-id" client_secret = "your-client-secret" # Use environment variables or secure vaults/key management solutions to handle secrets safely # import os # tenant_id = os.getenv("AZURE_TENANT_ID") # client_id = os.getenv("AZURE_CLIENT_ID") # client_secret = os.getenv("AZURE_CLIENT_SECRET") # Define the domain and other user attributes domain = "yashyadav34gmail.onmicrosoft.com" user_name = "johndoe" display_name = "John Doe" job_title = "Software Developer" email = f"{user_name}@example.com" mobile_phone = "+1234567890" office_location = "Building 1, Room 101" preferred_language = "en-US" password = "initialPassword123!" force_change_password = True credential = ClientSecretCredential(tenant_id, client_id, client_secret) # Define the user with detailed information user_data = { "accountEnabled": True, "displayName": display_name, "mailNickname": user_name, "userPrincipalName": f"{user_name}@{domain}", "givenName": "John", # Static givenName for demonstration; adjust as necessary "surname": "Doe", # Static surname for demonstration; adjust as necessary "jobTitle": job_title, "mail": email, "mobilePhone": mobile_phone, "officeLocation": office_location, "preferredLanguage": preferred_language, "passwordProfile": { "forceChangePasswordNextSignIn": force_change_password, "password": password } } # Create the user in Azure AD url = 'https://graph.microsoft.com/v1.0/users' try: access_token = credential.get_token('https://graph.microsoft.com/.default').token headers = { 'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json' } response = requests.post(url, headers=headers, json=user_data) response.raise_for_status() # Print raw JSON output print("Raw JSON response:") print(response.json()) print() # Extract data for table display user_info = response.json() print("User created successfully:") print(f"{'Attribute':<20} | {'Value'}") print("-" * 60) for key in ['id', 'displayName', 'userPrincipalName', 'mail', 'mobilePhone', 'officeLocation', 'givenName', 'surname', 'jobTitle', 'preferredLanguage']: print(f"{key:<20} | {user_info.get(key, 'Not assigned')}") except requests.exceptions.HTTPError as e: print("HTTP request failed:", e, response.text) except Exception as e: print("An error occurred:", e) """ # Sample Output Raw JSON response: {'@odata.context': 'https://graph.microsoft.com/v1.0/$metadata#users/$entity', 'id': '7f16743b-2b12-4697-b065-d9637a1b30e7', 'businessPhones': [], 'displayName': 'John Doe', 'givenName': 'John', 'jobTitle': 'Software Developer', 'mail': 'johndoe@example.com', 'mobilePhone': '+1234567890', 'officeLocation': 'Building 1, Room 101', 'preferredLanguage': 'en-US', 'surname': 'Doe', 'userPrincipalName': 'johndoe@yashyadav34gmail.onmicrosoft.com'} User created successfully: Attribute | Value ------------------------------------------------------------ id | 7f16743b-2b12-4697-b065-d9637a1b30e7 displayName | John Doe userPrincipalName | johndoe@yashyadav34gmail.onmicrosoft.com mail | johndoe@example.com mobilePhone | +1234567890 officeLocation | Building 1, Room 101 givenName | John surname | Doe jobTitle | Software Developer preferredLanguage | en-US """

import requests from azure.identity import ClientSecretCredential # Set up Azure AD credentials tenant_id = 'your-tenant-id' client_id = "your-client-id" client_secret = "your-client-secret" # Initialize the Azure credentials using Azure's ClientSecretCredential credential = ClientSecretCredential(tenant_id, client_id, client_secret) # Define the User Principal Name (UPN) of the user to be modified # user_principal_name = "johndoe@yashyadav34gmail.onmicrosoft.com" # Specify the user_principal_name like this if running the task in a standalone manner # user_info to be received from upstream task user_principal_name = user_info["userPrincipalName"] # Function to enable or disable a user account def update_user_account_enabled(user_principal_name, enable=True): # Construct the Microsoft Graph API URL for updating user details url = f"https://graph.microsoft.com/v1.0/users/{user_principal_name}" # Obtain an access token from Azure AD access_token = credential.get_token('https://graph.microsoft.com/.default').token # Set the authorization header with the bearer token headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json'} # Define the payload to update the 'accountEnabled' field payload = {"accountEnabled": enable} # Send a PATCH request to update the user's account status response = requests.patch(url, headers=headers, json=payload) try: response.raise_for_status() # This raises an HTTPError for bad requests (400 or 500 level) if response.status_code == 204: print("Update successful, no content returned.") # No content is expected on successful PATCH elif response.text: print("Update successful:", response.json()) # Print the response if any content is returned else: print("Update successful, no content to display.") except requests.exceptions.HTTPError as e: print(f"Failed to update user: {e.response.status_code} - {e.response.text}") # Function to retrieve the 'accountEnabled' status of a user def get_user_account_enabled(user_principal_name): # Construct the URL with a query parameter to select specific fields url = f"https://graph.microsoft.com/v1.0/users/{user_principal_name}?$select=displayName,accountEnabled" # Obtain an access token from Azure AD access_token = credential.get_token('https://graph.microsoft.com/.default').token # Set the authorization header with the bearer token headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json'} # Send a GET request to retrieve user details response = requests.get(url, headers=headers) try: response.raise_for_status() # This raises an HTTPError for bad HTTP responses if response.text: user_data = response.json() print(user_data) account_status = user_data.get("accountEnabled") print("User account status:", "Enabled" if account_status else "Disabled or not set") else: print("No data returned for user.") except requests.exceptions.HTTPError as e: print(f"Failed to retrieve user: {e.response.status_code} - {e.response.text}") # Example usage of the functions update_user_account_enabled(user_principal_name, enable=True) # Enable the user account current_status = get_user_account_enabled(user_principal_name) # Retrieve the current account status """ # Sample Output Update successful, no content returned. {'@odata.context': 'https://graph.microsoft.com/v1.0/$metadata#users(displayName,accountEnabled)/$entity', 'displayName': 'John Doe', 'accountEnabled': True} User account status: Enabled """

import requests from azure.identity import ClientSecretCredential # Set up Azure AD credentials tenant_id = 'your-tenant-id' client_id = "your-client-id" client_secret = "your-client-secret" # Initialize the Azure credentials credential = ClientSecretCredential(tenant_id, client_id, client_secret) def get_user_details(user_id): """Fetch user details by user ID.""" url = f"https://graph.microsoft.com/v1.0/users/{user_id}" access_token = credential.get_token('https://graph.microsoft.com/.default').token headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json'} response = requests.get(url, headers=headers) try: response.raise_for_status() return response.json() except requests.exceptions.HTTPError as e: print(f"Failed to retrieve user details: {e.response.status_code} - {e.response.text}") return None def get_group_id_by_name(group_name): """Retrieve the object ID of a group by its display name.""" url = f"https://graph.microsoft.com/v1.0/groups?$filter=displayName eq '{group_name}'" access_token = credential.get_token('https://graph.microsoft.com/.default').token headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json'} response = requests.get(url, headers=headers) try: response.raise_for_status() groups = response.json().get('value') if groups: group_id = groups[0]['id'] print(f"Group '{group_name}' found with ID: {group_id}") return group_id else: print(f"No group found with the name: {group_name}") return None except requests.exceptions.HTTPError as e: print(f"Failed to retrieve group: {e.response.status_code} - {e.response.text}") return None def add_user_to_group(user_id, group_id, group_name): """Add a user to a group using the user and group object IDs.""" user_details = get_user_details(user_id) if user_details: url = f"https://graph.microsoft.com/v1.0/groups/{group_id}/members/$ref" access_token = credential.get_token('https://graph.microsoft.com/.default').token headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json'} payload = {"@odata.id": f"https://graph.microsoft.com/v1.0/directoryObjects/{user_id}"} response = requests.post(url, headers=headers, json=payload) try: response.raise_for_status() print(f"User '{user_details['displayName']}' with ID: {user_id} was successfully added to group '{group_name}' (ID: {group_id}).") except requests.exceptions.HTTPError as e: print(f"Failed to add user to group: {e.response.status_code} - {e.response.text}") # Example usage: # group_name = "m365_test_group" # Specify the display name of the group # user_id = "7f16743b-2b12-4697-b065-d9637a1b30e7" # Specify the object ID of the user # user_info to be received from upstream task user_id = user_info["id"] # Get the group ID by name group_id = get_group_id_by_name(group_name) if group_id: # Add the user to the group if the group was found add_user_to_group(user_id, group_id, group_name) """ # Sample Output Group 'm365_test_group' found with ID: 80fc1ee5-90b5-4d89-81ab-718f5a74d202 User 'John Doe' with ID: 7f16743b-2b12-4697-b065-d9637a1b30e7 was successfully added to group 'm365_test_group' (ID: 80fc1ee5-90b5-4d89-81ab-718f5a74d202). """

import requests from azure.identity import ClientSecretCredential # Set up Azure AD credentials tenant_id = 'your-tenant-id' client_id = "your-client-id" client_secret = "your-client-secret" # Initialize Azure credentials credential = ClientSecretCredential(tenant_id, client_id, client_secret) def get_directory_role_id(role_name): """Retrieve the role ID by role name from directory roles.""" url = f"https://graph.microsoft.com/v1.0/directoryRoles?$filter=displayName eq '{role_name}'" access_token = credential.get_token('https://graph.microsoft.com/.default').token headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json'} response = requests.get(url, headers=headers) try: response.raise_for_status() roles = response.json().get('value') if roles: role_id = roles[0]['id'] print(f"Directory role '{role_name}' found with ID: {role_id}") return role_id else: print(f"No directory role found with the name: {role_name}") return None except requests.exceptions.HTTPError as e: print(f"Failed to retrieve directory role: {e.response.status_code} - {e.response.text}") return None def assign_role_to_user(user_id, role_id): """Assign a directory role to a user using the role ID and user's object ID.""" url = f"https://graph.microsoft.com/v1.0/directoryRoles/{role_id}/members/$ref" access_token = credential.get_token('https://graph.microsoft.com/.default').token headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json'} payload = {"@odata.id": f"https://graph.microsoft.com/v1.0/directoryObjects/{user_id}"} response = requests.post(url, headers=headers, json=payload) try: response.raise_for_status() print(f"User with ID: {user_id} was successfully assigned to the role with ID: {role_id}.") except requests.exceptions.HTTPError as e: print(f"Failed to assign role to user: {e.response.status_code} - {e.response.text}") # Example usage # role_name = "Global Reader" # user_id = "7f16743b-2b12-4697-b065-d9637a1b30e7" # user_info to be received from upstream task user_id = user_info["id"] # Get the directory role ID by name role_id = get_directory_role_id(role_name) if role_id: # Assign the role to the user if the role was found assign_role_to_user(user_id, role_id) """ # Sample Output Directory role 'Global Reader' found with ID: b3610c45-b84c-40c9-8de2-bec0341b3843 User with ID: 7f16743b-2b12-4697-b065-d9637a1b30e7 was successfully assigned to the role with ID: b3610c45-b84c-40c9-8de2-bec0341b3843. """

import requests from azure.identity import ClientSecretCredential # Set up Azure AD credentials tenant_id = 'your-tenant-id' # Replace with your Azure AD tenant ID client_id = "your-client-id" # Replace with your Azure AD client ID client_secret = "your-client-secret" # Replace with your Azure AD client secret # Initialize the Azure credentials credential = ClientSecretCredential(tenant_id, client_id, client_secret) def get_service_principal_by_app_id(app_id): """Retrieve the service principal ID by the application (client) ID.""" url = f"https://graph.microsoft.com/v1.0/servicePrincipals?$filter=appId eq '{app_id}'" access_token = credential.get_token('https://graph.microsoft.com/.default').token headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json'} response = requests.get(url, headers=headers) try: response.raise_for_status() service_principals = response.json().get('value') if service_principals: service_principal_id = service_principals[0]['id'] print(f"Service Principal for App ID '{app_id}' found with ID: {service_principal_id}") return service_principal_id else: print(f"No Service Principal found for App ID: {app_id}") return None except requests.exceptions.HTTPError as e: print(f"Failed to retrieve Service Principal: {e.response.status_code} - {e.response.text}") return None def get_app_role_id_by_name(service_principal_id, role_name): """Retrieve the app role ID by role name from a service principal.""" url = f"https://graph.microsoft.com/v1.0/servicePrincipals/{service_principal_id}/appRoles" access_token = credential.get_token('https://graph.microsoft.com/.default').token headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json'} response = requests.get(url, headers=headers) try: response.raise_for_status() app_roles = response.json().get('value') for role in app_roles: if role['displayName'] == role_name: print(f"App Role '{role_name}' found with ID: {role['id']}") return role['id'] print(f"No App Role found with the name: {role_name}") return None except requests.exceptions.HTTPError as e: print(f"Failed to retrieve App Role: {e.response.status_code} - {e.response.text}") return None def assign_app_role_to_user(user_id, service_principal_id, app_role_id): """Assign an app role to a user.""" url = f"https://graph.microsoft.com/v1.0/users/{user_id}/appRoleAssignments" access_token = credential.get_token('https://graph.microsoft.com/.default').token headers = {'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json'} payload = { "principalId": user_id, "resourceId": service_principal_id, "appRoleId": app_role_id } response = requests.post(url, headers=headers, json=payload) try: response.raise_for_status() print(f"App role with ID: {app_role_id} assigned to user with ID: {user_id} for Service Principal ID: {service_principal_id}") except requests.exceptions.HTTPError as e: print(f"Failed to assign app role to user: {e.response.status_code} - {e.response.text}") # Example usage: app_id = "your-app-id" # Application (client) ID of the Azure AD application # user_id = "7f16743b-2b12-4697-b065-d9637a1b30e7" # Object ID of the user # user_info to be received from upstream task user_id = user_info["id"] role_name = "Writers_Test_App_Role" # Role name to fetch the role ID # Get the Service Principal ID by application ID service_principal_id = get_service_principal_by_app_id(app_id) if service_principal_id: app_role_id = get_app_role_id_by_name(service_principal_id, role_name) if app_role_id: # Assign the app role to the user if the role was found assign_app_role_to_user(user_id, service_principal_id, app_role_id) """ # Sample Output Service Principal for App ID 'eff3e8cd-ecca-442f-936c-dffadf4fb44f' found with ID: 91da2692-4b04-4c22-bbe1-6b4bc5fe6784 App Role 'Writers_Test_App_Role' found with ID: 61c079fd-f368-4205-ad71-2f8abcd3b81a App role with ID: 61c079fd-f368-4205-ad71-2f8abcd3b81a assigned to user with ID: 7f16743b-2b12-4697-b065-d9637a1b30e7 for Service Principal ID: 91da2692-4b04-4c22-bbe1-6b4bc5fe6784 """

# This script does the following: # # # # # > Authenticate using Azure credentials. # # > Retrieve available licenses from Microsoft 365. # # > Map user roles to specific licenses. # # > Assign the appropriate licenses to the user based on their role. # # # # import requests from azure.identity import ClientSecretCredential # Azure AD credentials setup tenant_id = "your-tenant-id" client_id = "your-client-id" client_secret = "your-client-secret" # user_id = "7f16743b-2b12-4697-b065-d9637a1b30e7" # User ID from user creation step # user_info to be received from upstream task user_id = user_info["id"] # Role to license mapping (example mapping, update as necessary) role_license_map = { "Software Developer": "ENTERPRISEPACK", # Office 365 E3 "HR Manager": "SPE_E3", # Office 365 E3 + EMS E3 "Sales Executive": "DYN365_ENTERPRISE_SALES" } # User's role (this would be dynamically determined or inputted) user_role = "Software Developer" # Initialize the Azure credentials credential = ClientSecretCredential(tenant_id, client_id, client_secret) # Function to get the list of available licenses def get_available_licenses(): url = "https://graph.microsoft.com/v1.0/subscribedSkus" access_token = credential.get_token('https://graph.microsoft.com/.default').token headers = { 'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json' } response = requests.get(url, headers=headers) response.raise_for_status() return response.json() # Function to assign licenses to a user def assign_license_to_user(user_id, sku_id): url = f"https://graph.microsoft.com/v1.0/users/{user_id}/assignLicense" access_token = credential.get_token('https://graph.microsoft.com/.default').token headers = { 'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json' } license_assignments = { "addLicenses": [{"skuId": sku_id}], "removeLicenses": [] } response = requests.post(url, headers=headers, json=license_assignments) response.raise_for_status() return response.json() # Main logic to assign licenses based on the user's role try: available_licenses = get_available_licenses() license_to_assign = next((item for item in available_licenses['value'] if item['skuPartNumber'] == role_license_map[user_role]), None) if license_to_assign: result = assign_license_to_user(user_id, license_to_assign['skuId']) print("License assigned successfully:", result) else: print("No matching license found for the role.") except requests.exceptions.HTTPError as e: print("HTTP request failed:", e.response.text) except Exception as e: print("An error occurred:", e)