sten.wtf / Unstable Instance

secure/pipeline/actions/recovery_code.py

import json
import re

from secure.types import ActionResult

_GEN_URL = "https://account.live.com/API/Proofs/GenerateRecoveryCode"
_MANAGE_URL = "https://account.live.com/proofs/manage/additional?mkt=en-US"
_T0_RE = re.compile(r"var\s*t0\s*=\s*({.*?});", re.S)


def _encrypted_net_id_from_t0(t0: dict) -> str | None:
    manage = t0.get("manageProofs") or (t0.get("WLXAccount") or {}).get("manageProofs") or {}
    return manage.get("encryptedNetId") or t0.get("encryptedNetId")


async def _fetch_encrypted_net_id(http) -> str | None:
    async with http.request("GET", _MANAGE_URL) as resp:
        html = await resp.text()
    m = _T0_RE.search(html or "")
    if not m:
        return None
    try:
        t0 = json.loads(m.group(1))
    except Exception:
        return None
    return _encrypted_net_id_from_t0(t0)


class RecoveryCodeAction:
    name = "recovery_code"

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

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

        net_id = (session.extra or {}).get("encrypted_net_id")
        if not net_id:
            net_id = await _fetch_encrypted_net_id(session.http)
        if not net_id:
            log("recovery_code: missing encrypted_net_id")
            return ActionResult(self.name, False, detail="missing_encrypted_net_id")

        headers = {
            "Content-Type": "application/json",
            "X-Requested-With": "XMLHttpRequest",
            "Canary": session.apicanary,
        }
        try:
            async with session.http.request(
                "POST",
                _GEN_URL,
                data=json.dumps({"encryptedNetId": net_id}),
                headers=headers,
            ) as resp:
                if resp.status >= 400:
                    log(f"recovery_code: HTTP {resp.status}")
                    return ActionResult(self.name, False, detail=f"http_{resp.status}")
                data = await resp.json()
        except Exception as e:
            log(f"recovery_code: error: {e}")
            return ActionResult(self.name, False, detail=f"error: {e}")

        if not isinstance(data, dict):
            log("recovery_code: non-dict response body")
            return ActionResult(self.name, False, detail="bad_body")

        code = data.get("recoveryCode")
        if code:
            log("recovery_code: generated a new recovery code")
            return ActionResult(self.name, True, data={"recovery_code": code})

        err = data.get("error") if isinstance(data.get("error"), dict) else None
        code_s = str(err.get("code")) if err and err.get("code") is not None else "unknown"
        log(f"recovery_code: error code={code_s}")
        return ActionResult(self.name, False, detail=f"error_code_{code_s}")