Python Doc & Data Automation

Download Attachments from Microsoft 365 with the Graph API

The IMAP intake script that worked for years now fails against the company's Microsoft 365 mailbox with LOGIN failed., and the first Graph API attempt returns errors instead of mail: 401 InvalidAuthenticationToken, 403 ErrorAccessDenied: Access is denied. Check credentials and try again., or a list of messages whose attachments come back without content. Meanwhile the security team asks why an application should be able to read every mailbox in the tenant.

Root Cause

Exchange Online no longer accepts passwords for IMAP, so mailbox automation must use OAuth — and for unattended jobs, the cleanest OAuth route is Microsoft Graph with an application identity rather than a user. That identity needs three separate things before any request succeeds: an app registration with a client secret or certificate, the Mail.Read (or Mail.ReadWrite to move messages) application permission with admin consent, and a token requested for the Graph resource. The 401 usually means the token was requested for the wrong resource or has expired; the 403 means the permission is missing, unconsented, or restricted away from that mailbox. Separately, the Graph message and attachment endpoints behave differently from IMAP: messages are paged, attachment lists may omit content, large attachments need a raw $value download, /me does not exist for app-only tokens, and moving a message changes its id. Finally, Mail.Read as an application permission covers all mailboxes by default, which is why it must be scoped with Exchange's role-based access for applications.

Minimal Diagnostic

Request a token, decode its claims locally to confirm the audience and roles, then make the smallest possible call against the target mailbox and print the exact error body.

# pip install msal requests python-dotenv
import base64
import json
import os
import msal
import requests
from dotenv import load_dotenv

GRAPH = "https://graph.microsoft.com/v1.0"

def token_claims(token: str) -> dict:
    payload = token.split(".")[1]
    payload += "=" * (-len(payload) % 4)
    return json.loads(base64.urlsafe_b64decode(payload))

def diagnose() -> None:
    load_dotenv()
    app = msal.ConfidentialClientApplication(
        os.environ["AZURE_CLIENT_ID"],
        authority=f"https://login.microsoftonline.com/{os.environ['AZURE_TENANT_ID']}",
        client_credential=os.environ["AZURE_CLIENT_SECRET"],
    )
    result = app.acquire_token_for_client(scopes=["https://graph.microsoft.com/.default"])
    if "access_token" not in result:
        raise SystemExit(f"token error: {result.get('error')}: {result.get('error_description')}")
    claims = token_claims(result["access_token"])
    print("aud:", claims.get("aud"), "| roles:", claims.get("roles"), "| app:", claims.get("app_displayname"))
    mailbox = os.environ["MAILBOX"]
    resp = requests.get(f"{GRAPH}/users/{mailbox}/mailFolders/inbox/messages",
                        params={"$top": "1", "$select": "id,subject"},
                        headers={"Authorization": f"Bearer {result['access_token']}"}, timeout=30)
    print("status:", resp.status_code)
    print(resp.text[:400])

if __name__ == "__main__":
    diagnose()

aud: https://graph.microsoft.com | roles: None | app: invoice-intake
status: 403
{"error":{"code":"ErrorAccessDenied","message":"Access is denied. Check credentials and try again."}}

roles: None is the answer: the token carries no application permissions, so the app registration either lacks Mail.Read as an application permission or admin consent was never granted. If roles lists Mail.Read and the call still returns 403, the permission is scoped to other mailboxes.

Reading Graph mail errors The root asks what the token and response show. A token request error means wrong tenant, client id or secret. A 401 with an audience other than graph.microsoft.com means the token was requested for the wrong resource. A 403 with no roles claim means the Mail.Read application permission is missing or not consented. A 403 with Mail.Read in roles means Exchange application access scoping excludes this mailbox. A 404 on the users path means the mailbox address is wrong or has no Exchange licence. What do the token and response show? decode claims, print error body 401, wrong aud Wrong resource use graph .default 403, no roles No app permission add Mail.Read + consent 403, roles ok Scoped away mailbox not in scope 404 on /users Wrong mailbox address or licence

Fix: Least-Privilege App, Then List, Download and Move

Grant Mail.ReadWrite (to move processed messages) as an application permission with admin consent, and restrict the app to the intake mailbox using Exchange Online's role-based access control for applications. That scoping is done once by an Exchange administrator in PowerShell:

# Exchange Online PowerShell, run once by an administrator
New-ServicePrincipal -AppId <client-id> -ObjectId <enterprise-app-object-id> -DisplayName "invoice-intake"
New-ManagementScope -Name "Invoice intake mailbox" -RecipientRestrictionFilter "PrimarySmtpAddress -eq '[email protected]'"
New-ManagementRoleAssignment -App <client-id> -Role "Application Mail.ReadWrite" -CustomResourceScope "Invoice intake mailbox"
Test-ServicePrincipalAuthorization -Identity <client-id> -Resource [email protected]

With the scoped role assigned, remove the tenant-wide Mail.ReadWrite API permission from the app registration so the only grant is the scoped one. Then the Python job lists unread messages with attachments, downloads each file attachment's raw bytes, and moves the message. Changed lines carry comments.

# pip install msal requests python-dotenv
import hashlib
import os
import re
import time
from pathlib import Path
import msal
import requests
from dotenv import load_dotenv

GRAPH = "https://graph.microsoft.com/v1.0"
OUT = Path("inbox/incoming")
ALLOWED_EXT = {".pdf", ".xlsx", ".csv"}

class Graph:
    def __init__(self) -> None:
        load_dotenv()
        self.mailbox = os.environ["MAILBOX"]
        self.app = msal.ConfidentialClientApplication(
            os.environ["AZURE_CLIENT_ID"],
            authority=f"https://login.microsoftonline.com/{os.environ['AZURE_TENANT_ID']}",
            client_credential=os.environ["AZURE_CLIENT_SECRET"])
        self.session = requests.Session()

    def _token(self) -> str:
        result = self.app.acquire_token_for_client(scopes=["https://graph.microsoft.com/.default"])  # changed
        if "access_token" not in result:
            raise RuntimeError(result.get("error_description"))
        return result["access_token"]                     # MSAL caches and refreshes automatically

    def call(self, method: str, url: str, **kwargs) -> requests.Response:
        for attempt in range(5):
            resp = self.session.request(method, url, timeout=60,
                                        headers={"Authorization": f"Bearer {self._token()}"}, **kwargs)
            if resp.status_code in (429, 503, 504):                         # changed: honour throttling
                time.sleep(int(resp.headers.get("Retry-After", 2 ** attempt)))
                continue
            resp.raise_for_status()
            return resp
        raise RuntimeError(f"gave up after throttling: {url}")

    def unread_with_attachments(self):
        url = f"{GRAPH}/users/{self.mailbox}/mailFolders/inbox/messages"   # changed: /users/{mbx}, not /me
        params = {"$filter": "hasAttachments eq true and isRead eq false",
                  "$select": "id,subject,from,receivedDateTime,internetMessageId", "$top": "50"}
        while url:
            data = self.call("GET", url, params=params).json()
            yield from data.get("value", [])
            url, params = data.get("@odata.nextLink"), None                # changed: follow paging

    def file_attachments(self, message_id: str):
        url = f"{GRAPH}/users/{self.mailbox}/messages/{message_id}/attachments"
        data = self.call("GET", url, params={"$select": "id,name,contentType,size,isInline"}).json()
        for att in data.get("value", []):
            if att.get("@odata.type") != "#microsoft.graph.fileAttachment" or att.get("isInline"):
                continue                                                    # changed: skip items and logos
            raw = self.call("GET", f"{url}/{att['id']}/$value").content     # changed: raw bytes, any size
            yield att["name"], raw

    def move(self, message_id: str, folder: str = "archive") -> str:
        resp = self.call("POST", f"{GRAPH}/users/{self.mailbox}/messages/{message_id}/move",
                         json={"destinationId": folder})
        return resp.json()["id"]                                            # changed: id changes on move

def run() -> int:
    graph, saved = Graph(), 0
    OUT.mkdir(parents=True, exist_ok=True)
    for msg in graph.unread_with_attachments():
        for name, raw in graph.file_attachments(msg["id"]):
            ext = Path(name).suffix.lower()
            if ext not in ALLOWED_EXT:
                continue
            stem = re.sub(r"[^\w\-. ]+", "_", Path(name).stem)[:80]
            dest = OUT / f"{msg['receivedDateTime'][:10]}_{hashlib.sha256(raw).hexdigest()[:10]}_{stem}{ext}"
            if not dest.exists():
                tmp = dest.with_suffix(dest.suffix + ".part")
                tmp.write_bytes(raw)
                tmp.replace(dest)
                saved += 1
        graph.move(msg["id"])
    return saved

if __name__ == "__main__":
    try:
        print(f"saved {run()} attachment(s)")
    except (requests.HTTPError, RuntimeError) as exc:
        raise SystemExit(f"intake failed: {exc}")

Downloading through /$value returns the file bytes directly, which avoids the base64 contentBytes property that is not reliably included in list responses and becomes impractical for large files. Moving to the well-known archive folder keeps the inbox as a queue of unprocessed mail; use a folder id from /mailFolders for a custom Processed folder.

Graph calls for one message The job gets a token from Entra ID once and reuses it. It lists unread inbox messages with attachments, following nextLink pages. For each message it lists attachments with a select of id, name, content type, size and isInline, downloads each file attachment with the dollar value endpoint to receive raw bytes, and finally posts a move request, which returns the message's new id in the archive folder. Intake job Entra ID Microsoft Graph client credentials access token GET inbox/messages?$filter=... GET messages/{id}/attachments GET attachments/{aid}/$value POST messages/{id}/move new id in archive

Variant Fix 1: Certificates Instead of Client Secrets

Client secrets expire (at most two years) and are easy to leak into logs or repositories. Certificate credentials are the recommended option for production jobs:

# pip install msal cryptography python-dotenv
import os
from pathlib import Path
import msal
from cryptography import x509
from cryptography.hazmat.primitives import hashes

def cert_app() -> msal.ConfidentialClientApplication:
    pem_key = Path(os.environ["AZURE_CERT_KEY_PATH"]).read_text()
    cert = x509.load_pem_x509_certificate(Path(os.environ["AZURE_CERT_PATH"]).read_bytes())
    thumbprint = cert.fingerprint(hashes.SHA1()).hex()
    return msal.ConfidentialClientApplication(
        os.environ["AZURE_CLIENT_ID"],
        authority=f"https://login.microsoftonline.com/{os.environ['AZURE_TENANT_ID']}",
        client_credential={"private_key": pem_key, "thumbprint": thumbprint},
    )

Upload the public certificate to the app registration, keep the private key readable only by the service account that runs the job, and record the certificate's expiry date in your monitoring so renewal is scheduled rather than discovered.

Variant Fix 2: Only Fetch What Is New Since Last Run

Filtering on isRead eq false depends on nobody reading the intake mailbox by hand. A delta query tracks changes on the server side and returns only messages added since the previous call, independent of read state:

# pip install msal requests
from pathlib import Path

DELTA_STATE = Path("state/graph-delta-link.txt")

def new_messages(graph) -> list[dict]:
    url = DELTA_STATE.read_text().strip() if DELTA_STATE.exists() else (
        f"{GRAPH}/users/{graph.mailbox}/mailFolders/inbox/messages/delta"
        "?$select=id,subject,hasAttachments,receivedDateTime")
    found = []
    while url:
        data = graph.call("GET", url).json()
        found += [m for m in data.get("value", []) if m.get("hasAttachments") and "@removed" not in m]
        if "@odata.deltaLink" in data:
            DELTA_STATE.parent.mkdir(parents=True, exist_ok=True)
            DELTA_STATE.write_text(data["@odata.deltaLink"])      # resume point for the next run
            break
        url = data.get("@odata.nextLink")
    return found

Save the delta link only after the messages it returned have been processed successfully; otherwise a crash between saving the link and saving the files skips those messages permanently.

Choosing how to find new mail Filtering on isRead false is simple but breaks when people read the intake mailbox. Filtering on receivedDateTime greater than the last run is independent of read state but can miss messages delivered late with older timestamps. Delta queries return exactly what changed since the saved delta link and are the most reliable, at the cost of storing and handling the link carefully. Method Robust to humans Late deliveries Complexity isRead eq false no caught simple receivedDateTime > last yes can miss simple Delta query yes caught store deltaLink

Verification

Check the permission scoping as well as the download. An intake job that can read other mailboxes is a finding in any security review, even if it never does.

# pip install msal requests python-dotenv
import requests

def verify_scope(graph, other_mailbox: str) -> None:
    ok = graph.call("GET", f"{GRAPH}/users/{graph.mailbox}/mailFolders/inbox", params={"$select": "id"})
    assert ok.status_code == 200, "cannot read the intake mailbox"
    try:
        graph.session.get(f"{GRAPH}/users/{other_mailbox}/mailFolders/inbox",
                          headers={"Authorization": f"Bearer {graph._token()}"},
                          timeout=30).raise_for_status()
    except requests.HTTPError as exc:
        assert exc.response.status_code in (403, 404), f"unexpected status {exc.response.status_code}"
        print(f"scope verified: intake mailbox readable, {other_mailbox} denied")
        return
    raise AssertionError(f"app can read {other_mailbox}: restrict it with an application role scope")

if __name__ == "__main__":
    verify_scope(Graph(), "[email protected]")

Run the scope check after any permission change and quarterly. Pair it with the file checks from Processing Email Attachments Automatically — magic bytes, no leftover .part files — to verify the download side.

FAQ

Can I use delegated permissions and my own account instead? For a script someone runs interactively, yes, with a device-code or interactive login. Unattended scheduled jobs should use an application identity so they do not depend on a person's password, MFA prompt or employment.

Why do some attachments have no $value? Item attachments (attached emails, calendar items) and reference attachments (cloud links) are not files. Handle #microsoft.graph.itemAttachment separately with ?$expand=microsoft.graph.itemattachment/item, and follow reference attachments' sourceUrl with appropriate permissions.

How large can attachments be? Exchange Online allows messages up to the tenant's configured limit, often 25–150 MB. /$value streams them; pass stream=True to requests and write in chunks for the largest files.

Does moving a message break my state tracking? It changes the Graph id. Track messages by internetMessageId, which stays constant, as the IMAP intake does with Message-ID.

Part of Processing Email Attachments Automatically.