Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 32 additions & 0 deletions docs/proposals/quarantine/README.md
Original file line number Diff line number Diff line change
@@ -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 '<tenant-guid>' --client-id '<your-app-guid>' --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.
119 changes: 119 additions & 0 deletions docs/proposals/quarantine/quarantine_read.py
Original file line number Diff line number Diff line change
@@ -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:<lab-user> 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
)
)