Script to deploy add-ins to O365, using reversed engineered endpoints.

deploy_o365_addin.py

import logging
import re
import os
import sys
import xml
import json
import time
import httpx
import asyncio
import xmltodict

logger = logging.getLogger()


class M365AdminClient:
    def __init__(self, username, password):
        self._client = httpx.AsyncClient(
            base_url="https://admin.microsoft.com",
            timeout=60,
        )
        self._username = username
        self._password = password

    async def get(self, path: str, **kwargs):
        resp = await self._client.get(path, follow_redirects=True, **kwargs)
        if resp.status_code >= 400:
            raise httpx.HTTPError(
                f"M365AdminClient error: {resp.status_code} {resp.text}"
            )
        try:
            return resp.json()
        except json.JSONDecodeError:
            return resp.text

    async def post(self, path: str, **kwargs):
        resp = await self._client.post(path, **kwargs)
        if resp.status_code >= 400:
            raise httpx.HTTPError(
                f"M365AdminClient error: {resp.status_code} {resp.text}"
            )
        try:
            return resp.json()
        except json.JSONDecodeError:
            return resp.text

    async def authorize(self):
        # Auto redirects to MSOL login
        config_resp = await self.get("/login")
        msol_config = json.loads(re.search(r"\$Config=(.*?);", config_resp)[1])
        # Set cookies and get token
        token_resp = await self.post(
            "https://login.microsoftonline.com/common/login",
            data={
                "type": 11,
                "login": self._username,
                "passwd": self._password,
                "flowToken": msol_config["sFT"],
                "ctx": msol_config["sCtx"],
                "canary": msol_config["canary"],
                "hpgrequestid": msol_config["sessionId"],
            },
        )
        try:
            token_content = xmltodict.parse(token_resp)["html"]["body"]["form"]["input"]
        except (xml.parsers.expat.ExpatError, KeyError):
            raise ValueError("Unable to log into Microsoft Online")

        # Login to M365 admin portal to set cookies
        await self.post(
            "/landing",
            data={e["@name"]: e["@value"] for e in token_content},
        )
        # Set header so the portal won't redirect to login
        self._client.headers.update(
            {
                "Accept": "application/json, text/plain, */*",
                "AjaxSessionKey": self._client.cookies["s.AjaxSessionKey"],
            }
        )

    async def update_custom_addin_file(
        self,
        product_id: str,
        manifest_path: str,
        locale: str = "en-US",
    ):
        return await self.post(
            "/fd/addins/api/apps/uploadCustomApp",
            params={
                "workloads": "AzureActiveDirectory,WXPO,MetaOS,Teams,SharePoint",
            },
            files={
                "AppFile": open(manifest_path, "rb"),
            },
            data={
                "ProductId": product_id,
                "Locale": locale,
                "ContentMarket": locale,
                "ActionType": "UPDATEAPP",
                "WorkloadType": "WXPO",
            },
        )

    async def update_addin(
        self,
        product_id: str,
        product_version: str,
        file_name: str,
        locale: str = "en-US",
    ):
        return await self.post(
            "/fd/addins/api/apps",
            json={
                "Locale": locale,
                "ContentMarket": locale,
                "WorkloadManagementList": [
                    {
                        "AppsourceAssetID": None,
                        "ProductID": product_id,
                        "Command": "UPDATEAPP",
                        "ActiveDirectoryAppId": None,
                        "Version": product_version,
                        "AppType": "LOB",
                        "AppFileName": file_name,
                        "ApplicationTemplateId": None,
                        "Workload": "WXPO",
                        "MosOperationId": None,
                        "TitleID": None,
                    }
                ],
                "SendEmailToUsers": False,
            },
        )

    async def get_addin_deploy_status(self, request_id: str):
        return await self.get(f"/fd/addins/api/deploymentRequestStatus/{request_id}")


def parse_addin_manifest(manifest_path: str):
    with open(manifest_path, "r") as r:
        file = xmltodict.parse(r.read())
    return file["OfficeApp"]["Id"], file["OfficeApp"]["Version"]


async def update_addin(file_path: str):
    global client
    ID, VER = parse_addin_manifest(file_path)

    # Upload the manifest file to O365
    file_resp = await client.update_custom_addin_file(
        product_id=ID,
        manifest_path=file_path,
    )
    remote_file = file_resp["appDetail"]["appFileName"]
    logger.info(f"Uploaded file. File name: {remote_file}")

    # Update the add-in to use the uploaded manifest
    deploy_resp = await client.update_addin(
        product_id=ID,
        product_version=VER,
        file_name=remote_file,
    )
    request_id = deploy_resp["appManagementRequestID"]
    logger.info(f"Deploying add-in update. Request ID: {request_id}")

    # Get update status
    in_progress = True
    while in_progress:
        time.sleep(1)
        update_resp = await client.get_addin_deploy_status(request_id)
        status = update_resp["appsManagementStatus"][0]["status"]
        logger.info(f"Deployment status: {status}")

        if status == "Failed":
            error_reason: str = update_resp["appsManagementStatus"][0]["errorReason"]
            logger.error(error_reason.replace("\n", ""))
            in_progress = False
        elif status == "Success":
            print(f"Add-in updated to {VER}.")
            in_progress = False


async def main(manifest_paths: list[str]):
    global client
    await client.authorize()
    for path in manifest_paths:
        await update_addin(path)


if __name__ == "__main__":
    if len(sys.argv) <= 1 or sys.argv[1] == "help":
        print(f"Usage: {sys.argv[0]} AAD_USERNAME AAD_PASSWORD MANIFEST_PATH_1 MANIFEST_PATH_2 ...")
        sys.exit(1)

    logging.basicConfig(level=logging.INFO)
    client = M365AdminClient(
        username=os.getenv("AAD_USERNAME", sys.argv[1]),
        password=os.getenv("AAD_PASSWORD", sys.argv[2]),
    )
    asyncio.run(main(manifest_paths=sys.argv[3:]))