From 6ee2393990de1b9e0800484d9d2836737d26b807 Mon Sep 17 00:00:00 2001 From: wiperp Date: Sat, 17 Oct 2026 00:03:46 +0200 Subject: [PATCH] docs: propose native Python quarantine read sample --- docs/proposals/quarantine/README.md | 32 +++++ docs/proposals/quarantine/quarantine_read.py | 119 +++++++++++++++++++ 2 files changed, 151 insertions(+) create mode 100644 docs/proposals/quarantine/README.md create mode 100644 docs/proposals/quarantine/quarantine_read.py diff --git a/docs/proposals/quarantine/README.md b/docs/proposals/quarantine/README.md new file mode 100644 index 00000000000..ef5691bb869 --- /dev/null +++ b/docs/proposals/quarantine/README.md @@ -0,0 +1,32 @@ +# Native Python quarantine access: discussion sample + +This small extract demonstrates an Exchange quarantine list request using Azure +Identity and Kiota. It authenticates with the caller's own app and sends HTTP +directly; it does not execute PowerShell or import ExchangeOnlineManagement. + +The proposal asks where supported Python quarantine management should belong and +which API contract Microsoft recommends. The Exchange route used here is an +internal module transport, separate from Graph and the public Admin API v2 preview. +This sample does not add an API to `GraphServiceClient` or the generated SDK. + +The underlying lab client successfully completed one list request on 2026-10-16. +This extracted sample is validated offline only. It covers one read, prints no +message data, and deliberately omits paging, CLIXML parsing, export and mutations. + +## Run in a commercial-cloud lab + +Use your own single-tenant public-client app with delegated `Exchange.Manage` +consent and a signed-in lab account with the required Exchange RBAC permissions. +This management permission is broader than the sample's read-only operation. +The script does not create an app or grant permissions. + +```powershell +python -m pip install azure-identity httpx microsoft-kiota-abstractions microsoft-kiota-authentication-azure microsoft-kiota-http +python docs/proposals/quarantine/quarantine_read.py --tenant-id '' --client-id '' --anchor-mailbox 'UPN:admin@contoso.onmicrosoft.com' +``` + +## Related documented interfaces + +- [Quarantine management through PowerShell](https://learn.microsoft.com/en-us/defender-office-365/quarantine-admin-manage-messages-files#use-powershell-to-manage-quarantined-messages) +- [Public Exchange Admin API scope](https://learn.microsoft.com/en-us/exchange/reference/admin-api-overview) +- [Graph analyzed-email list](https://learn.microsoft.com/en-us/graph/api/security-collaborationroot-list-analyzedemails?view=graph-rest-1.0) and [remediation](https://learn.microsoft.com/en-us/graph/api/security-analyzedemail-remediate?view=graph-rest-1.0): related APIs whose coverage, application permissions and Defender licensing must be assessed separately. diff --git a/docs/proposals/quarantine/quarantine_read.py b/docs/proposals/quarantine/quarantine_read.py new file mode 100644 index 00000000000..f5da64ff972 --- /dev/null +++ b/docs/proposals/quarantine/quarantine_read.py @@ -0,0 +1,119 @@ +"""Discussion sample: one native Python read through Exchange's internal transport.""" + +import argparse +import asyncio +import json +from uuid import UUID, uuid4 + +import httpx +from azure.identity import DeviceCodeCredential +from kiota_abstractions.method import Method +from kiota_abstractions.native_response_handler import NativeResponseHandler +from kiota_abstractions.request_information import RequestInformation +from kiota_authentication_azure.azure_identity_authentication_provider import ( + AzureIdentityAuthenticationProvider, +) +from kiota_http.httpx_request_adapter import HttpxRequestAdapter +from kiota_http.kiota_client_factory import KiotaClientFactory +from kiota_http.middleware.options import ( + RedirectHandlerOption, + ResponseHandlerOption, + RetryHandlerOption, +) + + +async def read_quarantine(tenant_id: str, client_id: str, anchor_mailbox: str) -> None: + """Authenticate with the caller's own app and print a count, without message data.""" + tenant_id, client_id = str(UUID(tenant_id)), str(UUID(client_id)) + if ( + not anchor_mailbox.startswith("UPN:") + or not anchor_mailbox[4:].strip() + or any(character in anchor_mailbox for character in "\r\n") + ): + raise ValueError( + "anchor_mailbox must be a nonempty UPN: routing hint" + ) + + request = RequestInformation(Method.POST) + request.url = ( + f"https://outlook.office365.com/adminapi/beta/{tenant_id}/InvokeCommand" + ) + for name, value in { + "Accept": "application/json", + "Accept-Charset": "UTF-8", + "Content-Type": "application/json", + "X-CmdletName": "Get-QuarantineMessage", + "X-AnchorMailbox": anchor_mailbox, + "X-ResponseFormat": "auto", + "X-SerializationLevel": "Partial", + "Prefer": "odata.maxpagesize=1000", + "client-request-id": str(uuid4()), + }.items(): + request.headers.add(name, value) + request.content = json.dumps( + { + "CmdletInput": { + "CmdletName": "Get-QuarantineMessage", + "Parameters": {"Page": 1, "PageSize": 1}, + } + } + ).encode("utf-8") + request.add_request_options( + [ + ResponseHandlerOption(NativeResponseHandler()), + RedirectHandlerOption(should_redirect=False), + RetryHandlerOption(max_retries=3), + ] + ) + + with DeviceCodeCredential( + tenant_id=tenant_id, client_id=client_id + ) as credential: + authentication = AzureIdentityAuthenticationProvider( + credential, + scopes=["https://outlook.office365.com/.default"], + allowed_hosts=["outlook.office365.com"], + ) + async with KiotaClientFactory.create_with_default_middleware() as http_client: + adapter = HttpxRequestAdapter(authentication, http_client=http_client) + response = await adapter.send_primitive_async(request, "bytes", None) + if not isinstance(response, httpx.Response): + raise TypeError("Expected the native HTTPX response") + try: + response.raise_for_status() + payload = response.json() + if not isinstance(payload, dict): + raise TypeError("Expected an Exchange response object") + if "error" in payload or payload.get("@adminapi.errors"): + raise RuntimeError("Exchange returned a command error") + records = payload.get("value") + if not isinstance(records, list) or any( + not isinstance(record, dict) for record in records + ): + raise RuntimeError("Expected a value array of records") + if any("_cliXml" in record for record in records): + raise RuntimeError( + "PowerShell CLIXML deserialization is outside this sample" + ) + if payload.get("@odata.nextLink"): + raise RuntimeError("Transport continuation is outside this sample") + request_id = response.headers.get("request-id") or response.headers.get( + "x-ms-request-id" + ) + print(f"Returned {len(records)} record(s).") + print(f"Request ID: {request_id or '(not supplied)'}") + finally: + await response.aclose() + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--tenant-id", required=True) + parser.add_argument("--client-id", required=True) + parser.add_argument("--anchor-mailbox", required=True) + arguments = parser.parse_args() + asyncio.run( + read_quarantine( + arguments.tenant_id, arguments.client_id, arguments.anchor_mailbox + ) + )