sten.wtf / Unstable Instance

secure/pipeline/actions/alias.py

import json

from secure.generators import new_alias
from secure.types import ActionResult

_CREATE_URL = "https://account.live.com/API/CreateAlias2"
_MAKE_PRIMARY_URL = "https://account.live.com/API/MakePrimary"

_CREATE_OK_CODES = {"1179", "1178", "500"}
_PRIMARY_OK_CODES = {"500"}


def _pick_canary(data: dict) -> str | None:
    if not isinstance(data, dict):
        return None
    return data.get("apiCanary") or data.get("canary")


def _headers(canary: str) -> dict:
    return {
        "Content-Type": "application/json",
        "X-Requested-With": "XMLHttpRequest",
        "canary": canary,
    }


class AliasAction:
    name = "alias"

    async def run(self, session, ctx) -> ActionResult:
        log = ctx.log if ctx and ctx.log else (lambda msg: None)
        alias = new_alias()

        if not session.apicanary:
            log("alias: missing apicanary")
            return ActionResult(self.name, False, detail="missing_apicanary")

        create_body = {
            "propertyBag": {
                "displayNamePii": alias,
                "namePii": alias,
                "isMobile": "false",
                "mobileCountryCode": "US",
                "market": "en-US",
            },
            "sendVerificationCode": False,
            "channelType": "sms",
            "shouldGetEncryptedAlias": False,
            "uiflvr": 1001,
            "uaid": "83e297d2177a42c9b08776ad5cacd3c1",
            "scid": 100141,
            "hpgid": 200958,
        }
        try:
            async with session.http.request(
                "POST",
                _CREATE_URL,
                data=json.dumps(create_body),
                headers=_headers(session.apicanary),
            ) as resp:
                status = resp.status
                if status == 429:
                    log("alias: create rate limited (429)")
                    return ActionResult(self.name, False, detail="alias_rate_limited")
                if status == 504:
                    log("alias: create HTTP 504, presumptive success")
                else:
                    data = await resp.json()
                    err = data.get("error") if isinstance(data, dict) else None
                    if err:
                        code = str(err.get("code")) if isinstance(err, dict) and err.get("code") is not None else None
                        if code not in _CREATE_OK_CODES:
                            detail = "alias_rate_limited" if code == "1213" else f"create_error_code_{code}"
                            log(f"alias: create failed status={status} code={code}")
                            return ActionResult(self.name, False, detail=detail)
                    else:
                        canary = _pick_canary(data)
                        if canary:
                            session.apicanary = canary
                        elif not (isinstance(data, dict) and data.get("alias")):
                            log(f"alias: create failed status={status} (no canary/alias/error in body)")
                            return ActionResult(self.name, False, detail=f"create_failed_status_{status}")
        except Exception as e:
            log(f"alias: create error: {e}")
            return ActionResult(self.name, False, detail=f"create_error: {e}")

        primary_body = {
            "aliasName": alias.lower(),
            "emailChecked": True,
            "removeOldPrimary": False,
        }
        try:
            async with session.http.request(
                "POST",
                _MAKE_PRIMARY_URL,
                data=json.dumps(primary_body),
                headers=_headers(session.apicanary),
            ) as resp:
                if resp.status >= 400:
                    log(f"alias: makePrimary HTTP {resp.status}")
                    return ActionResult(self.name, False, detail=f"make_primary_http_{resp.status}")
                data = await resp.json()
        except Exception as e:
            log(f"alias: makePrimary error: {e}")
            return ActionResult(self.name, False, detail=f"make_primary_error: {e}")

        fresh = _pick_canary(data)
        if fresh:
            session.apicanary = fresh
        else:
            err = data.get("error") if isinstance(data, dict) else None
            code = str(err.get("code")) if isinstance(err, dict) and err.get("code") is not None else None
            if code is not None and code not in _PRIMARY_OK_CODES:
                log(f"alias: makePrimary error code={code}")
                return ActionResult(self.name, False, detail=f"make_primary_error_code_{code}")

        log(f"alias: set {alias} as primary")
        return ActionResult(self.name, True, data={"alias": alias})