pythonazureoauth-2.0microsoft-graph-api

Why do I keep getting "401 unauthorized" response when sending mails through Graph API with delegated access?


I have created an app on Azure to send emails using Graph API with delegated type access App registration

I'm using an interactive flow to generate auth code and access token using oauth2/v2.0/authorize and oauth2/v2.0/token endpoints using the following code

import requests
import webbrowser
import urllib.parse

class GraphEmailSender:
    def __init__(self, client_id, tenant_id):
        """
        Initialize Graph Email Sender
        
        :param client_id: Azure AD application client ID
        :param tenant_id: Azure AD tenant ID
        """
        self.client_id = client_id
        self.tenant_id = tenant_id
        
        # OAuth endpoints
        self.authorize_url = f'https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/authorize'
        self.token_url = f'https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token'
        
        # Graph API endpoint
        self.graph_endpoint = 'https://graph.microsoft.com/v1.0'
        
        # Redirect URI (must be registered in Azure AD app)
        self.redirect_uri = 'http://localhost:8000/callback'
        
        # Scopes for email sending
        self.scopes = [
            'https://graph.microsoft.com/Mail.Send',
            'https://graph.microsoft.com/User.Read',
            'offline_access'
        ]
    
    def get_authorization_code(self):
        """
        Generate authorization URL and prompt for manual code entry
        
        :return: Authorization code
        """
        # Prepare authorization request parameters
        auth_params = {
            'client_id': self.client_id,
            'response_type': 'code',
            'redirect_uri': self.redirect_uri,
            'scope': ' '.join(self.scopes),
            'response_mode':'fragment'
        }
        
        # Construct authorization URL
        auth_url = f"{self.authorize_url}?{urllib.parse.urlencode(auth_params)}"
        
        # Open browser for authorization
        print("Please authorize the application:")
        webbrowser.open(auth_url)
        
        # Manually enter authorization code
        auth_code = input("Enter the authorization code from the redirect URL: ")
        return auth_code
    
    def exchange_code_for_token(self, authorization_code):
        """
        Manually exchange authorization code for access token
        
        :param authorization_code: Authorization code
        :return: Access token
        """
        # Prepare token exchange parameters
        token_params = {
            'client_id': self.client_id,
            'grant_type': 'authorization_code',
            'code': authorization_code,
            'redirect_uri': self.redirect_uri,
            'scope': ' '.join(self.scopes),
            'client_secret':'Ixxxx'
        }
        
        # Send token request
        response = requests.post(
            self.token_url, 
            data=token_params,
            headers={'Content-Type': 'application/x-www-form-urlencoded'}
        )
        # Return access token
        return response.json().get('access_token')
    
    def send_email(self, access_token, to_email, subject, body):
        """
        Send email using access token
        
        :param access_token: OAuth access token
        :param to_email: Recipient email address
        :param subject: Email subject
        :param body: Email body
        :return: Boolean indicating success
        """
        # Prepare email message
        email_message = {
            "message": {
                "subject": subject,
                "body": {
                    "contentType": "Text",
                    "content": body
                },
                "toRecipients": [
                    {
                        "emailAddress": {
                            "address": to_email
                        }
                    }
                ],
                "from":{
                    "emailAddresss":{
                        "address":"lxxx1@gmail.com"
                    }
                }
            },
            "saveToSentItems": "true"
        }
        
        # Prepare headers
        headers = {
            'Authorization': f'Bearer {access_token}',
            'Content-Type': 'application/json'
        }
        try:
            # Send email via Graph API
            response = requests.post(
                f'{self.graph_endpoint}/me/sendMail', 
                json=email_message, 
                headers=headers
            )
            
            # Check response
            if response.status_code in [200, 201, 202]:
                print("Email sent successfully!")
                return True
            else:
                print(f"Failed to send email. Status code: {response.status_code} {response}")
                return False
        
        except Exception as e:
            print(f"Error sending email: {e}")
            return False

# Example usage
def main():
    # Replace with your actual Azure AD application details
    TENANT_ID = '97sss'
    CLIENT_ID = '803csss'
    
    # Create email sender
    email_sender = GraphEmailSender(CLIENT_ID, TENANT_ID)
    
    # Get authorization code (manual process)
    auth_code = email_sender.get_authorization_code()
    
    # Exchange code for access token
    access_token = email_sender.exchange_code_for_token(auth_code)
    
    # Send email
    email_sender.send_email(
        access_token,
        to_email='gagsgdg@gmail.com',
        subject='OAuth Email Test',
        body='Email sent using simplified OAuth flow.'
    )

if __name__ == '__main__':
    main()

It seems like the access token that is generated also has all the permissions(scope) i.e., Mail.Send,User.Read JWT

But, in the end, I got 401 unauthorized error.

Failed to send email. Status code: 401 <Response [401]>

I have also set the redirect_uri and supported account type to all accounts Permissions

Do I need to use an Office365 account or I can use any Microsoft account to send and receive emails?

Edit: Here's the screenshot of the error code and output enter image description here



Solution

  • Initially, I got the same error message like you by using your script:

    enter image description here

    I configured my app registration same like you, Registered Microsoft Entra ID application:

    enter image description here

    Added delegated type Mail.Send API permission and granted admin consent like below:

    enter image description here

    Use below modified Python Script:

    import requests
    import webbrowser
    import urllib.parse
    import json
    import base64
    
    class GraphEmailSender:
        def __init__(self, client_id, tenant_id, client_secret):
            """
            Initialize Graph Email Sender
            
            :param client_id: Azure AD application client ID
            :param tenant_id: Azure AD tenant ID
            :param client_secret: Azure AD application client secret
            """
            self.client_id = client_id
            self.tenant_id = tenant_id
            self.client_secret = client_secret
    
            # OAuth endpoints
            self.authorize_url = f'https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/authorize'
            self.token_url = f'https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token'
    
            # Graph API endpoint
            self.graph_endpoint = 'https://graph.microsoft.com/v1.0'
    
            # Redirect URI (must be registered in Azure AD app)
            self.redirect_uri = 'https://jwt.ms'
    
            # Scopes for email sending
            self.scopes = [
                'Mail.Send',
                'User.Read',
                'offline_access'
            ]
        
        def get_authorization_code(self):
            """
            Generate authorization URL and prompt for manual code entry
            :return: Authorization code
            """
            auth_params = {
                'client_id': self.client_id,
                'response_type': 'code',
                'redirect_uri': self.redirect_uri,
                'scope': ' '.join(self.scopes),
                'response_mode': 'query'
            }
    
            auth_url = f"{self.authorize_url}?{urllib.parse.urlencode(auth_params)}"
    
            print("Please authorize the application:")
            webbrowser.open(auth_url)
    
            auth_code = input("Enter the authorization code from the redirect URL: ")
            return auth_code.strip()
        
        def exchange_code_for_token(self, authorization_code):
            """
            Exchange authorization code for access token
            :param authorization_code: Authorization code
            :return: Access token
            """
            token_params = {
                'client_id': self.client_id,
                'grant_type': 'authorization_code',
                'code': authorization_code,
                'redirect_uri': self.redirect_uri,
                'scope': ' '.join(self.scopes),
                'client_secret': self.client_secret
            }
    
            response = requests.post(
                self.token_url, 
                data=token_params,
                headers={'Content-Type': 'application/x-www-form-urlencoded'}
            )
            
            token_data = response.json()
            
            # Debug token response
            print("\nToken Response:", json.dumps(token_data, indent=2))
            
            if 'access_token' in token_data:
                return token_data['access_token']
            else:
                print("Failed to retrieve access token:", token_data.get("error_description", "Unknown error"))
                return None
        
        def decode_jwt(self, jwt_token):
            """
            Decode JWT token (for debugging scopes)
            """
            parts = jwt_token.split('.')
            if len(parts) < 2:
                print("Invalid JWT Token")
                return None
            payload = json.loads(base64.urlsafe_b64decode(parts[1] + '=='))
            print("\nDecoded JWT Payload:", json.dumps(payload, indent=2))
        
        def send_email(self, access_token, to_email, subject, body):
            """
            Send email using access token
            :param access_token: OAuth access token
            :param to_email: Recipient email address
            :param subject: Email subject
            :param body: Email body
            :return: Boolean indicating success
            """
            email_message = {
                "message": {
                    "subject": subject,
                    "body": {
                        "contentType": "Text",
                        "content": body
                    },
                    "toRecipients": [
                        {
                            "emailAddress": {
                                "address": to_email
                            }
                        }
                    ]
                },
                "saveToSentItems": "true"
            }
    
            headers = {
                'Authorization': f'Bearer {access_token}',
                'Content-Type': 'application/json'
            }
    
            response = requests.post(
                f'{self.graph_endpoint}/me/sendMail', 
                json=email_message, 
                headers=headers
            )
            
            if response.status_code in [200, 201, 202]:
                print(" Email sent successfully!")
                return True
            else:
                print(f" Failed to send email. Status code: {response.status_code}")
                print("Response:", response.text)
                return False
    
    
    # Example usage
    def main():
        # Replace with your actual Azure AD application details
        TENANT_ID = '<TENANT_ID>'
        CLIENT_ID = '<CLIENT_ID>'
        CLIENT_SECRET = '<CLIENT_SECRET>'  # Keep this secret, avoid hardcoding
    
        email_sender = GraphEmailSender(CLIENT_ID, TENANT_ID, CLIENT_SECRET)
    
        # Get authorization code (manual process)
        auth_code = email_sender.get_authorization_code()
    
        # Exchange code for access token
        access_token = email_sender.exchange_code_for_token(auth_code)
        
        if access_token:
            # Decode JWT to check scopes
            email_sender.decode_jwt(access_token)
    
            # Send email
            email_sender.send_email(
                access_token,
                to_email='<RECIPENT_MAIL_ID>',
                subject='OAuth Email Test',
                body='Email sent using Graph API with OAuth.'
            )
        else:
            print(" Access token could not be obtained.")
    
    
    if __name__ == '__main__':
        main()
    
    

    Response:

    enter image description here

    Also, I've verified the same from Outlook portal under Sent Items tab:

    enter image description here