sten.wtf / Unstable Instance

secure/pipeline/actions/enable_2fa.py

import re

from secure.generators import new_totp_secret, totp_now
from secure.types import ActionResult

_ADD_URL = "https://account.live.com/proofs/Add?uaid=b79c68e4a2f04965a2a9a4d694809ff0&mpsplit=2&apt=3&mkt=en-us"
_VERIFY_URL = "https://account.live.com/API/AddVerifyTotp"
_ENABLE_URL = "https://account.live.com/proofs/EnableTfa?mkt=en-us"
_ENABLE_UAID = "eb19fd46532e4139a1dc2aedc2b881e4"

_KEY_RE = re.compile(r'Secret key: <span class="dirltr bold">([^<]+)<')
_PROOF_RE = re.compile(r'<input type="hidden" id="ProofId" name="ProofId" value="([^"]+)"')
_RVTKN_RE = re.compile(r"rvtkn=([^\"&\s]+)")


async def _fetch_add_page(http) -> str:
    async with http.request(
        "GET", _ADD_URL, headers={"X-Requested-With": "XMLHttpRequest"}
    ) as resp:
        return await resp.text()


def _parse_proof_id(text: str) -> str | None:
    if not _KEY_RE.search(text):
        return None
    proof_match = _PROOF_RE.search(text)
    return proof_match.group(1) if proof_match else None


async def _verify_totp(http, canary, proof_id, secret) -> bool:
    code = totp_now(secret)
    async with http.request(
        "POST",
        _VERIFY_URL,
        json={"ProofId": proof_id, "TotpCode": code},
        headers={"Canary": canary or ""},
    ) as resp:
        body = await resp.json()
    return isinstance(body, dict) and bool(body.get("apiCanary"))


async def _enable_tfa(http) -> bool:
    async with http.request("GET", _ENABLE_URL) as resp:
        text = await resp.text()
    rvtkn_match = _RVTKN_RE.search(text)
    if not rvtkn_match:
        return False
    confirm_url = f"{_ENABLE_URL}&uaid={_ENABLE_UAID}&rvtkn={rvtkn_match.group(1)}"
    async with http.request("GET", confirm_url) as resp:
        return getattr(resp, "status", 0) == 200


class EnableTwoFactorAction:
    name = "enable_2fa"

    async def run(self, session, ctx) -> ActionResult:
        try:
            add_page_text = await _fetch_add_page(session.http)
        except Exception as exc:
            return ActionResult(self.name, False, detail=f"add_page_failed: {exc}")

        proof_id = _parse_proof_id(add_page_text)
        if proof_id is None:
            return ActionResult(self.name, True, detail="already_enabled")

        secret = new_totp_secret()
        try:
            verified = await _verify_totp(session.http, session.apicanary, proof_id, secret)
        except Exception as exc:
            return ActionResult(self.name, False, detail=f"verify_failed: {exc}")
        if not verified:
            return ActionResult(self.name, False, detail="verify_rejected")

        try:
            enabled = await _enable_tfa(session.http)
        except Exception as exc:
            return ActionResult(self.name, False, detail=f"enable_failed: {exc}")
        if not enabled:
            return ActionResult(self.name, False, detail="enable_failed")

        return ActionResult(self.name, True, data={"totp_secret": secret})