sten.wtf / Unstable Instance

secure/pipeline/actions/password.py

import json
import re

from secure.generators import new_password
from secure.types import ActionResult

_TICKET_URL = "https://account.live.com/password/Change"
_CHANGE_URL = "https://account.live.com/API/ChangePassword"
_UAID = "7a54e2e7491e45feaa0568c19e1ba059"

_TICKET_PATTERNS = (
    r'<input[^>]+id=["\']iPostedTicket["\'][^>]+name=["\']t["\'][^>]+value=["\']([^"\']+)["\']',
    r'<input[^>]+name=["\']t["\'][^>]+value=["\']([^"\']+)["\']',
)


def _parse_ticket(html: str) -> str | None:
    for pat in _TICKET_PATTERNS:
        m = re.search(pat, html or "")
        if m:
            return m.group(1)
    return None


async def _get_password_ticket(http) -> str | None:
    async with http.request("GET", _TICKET_URL) as resp:
        html = await resp.text()
    return _parse_ticket(html)


def _change_payload(ticket: str, password: str, uaid: str) -> dict:
    return {
        "userTicket": ticket,
        "token": None,
        "password": password,
        "expiryEnabled": False,
        "uiflvr": 1001,
        "uaid": uaid,
        "scid": 100104,
        "hpgid": 200710,
    }


def _change_headers(canary: str | None, uaid: str) -> dict:
    return {
        "accept": "application/json",
        "canary": canary or "",
        "content-type": "application/x-www-form-urlencoded; charset=UTF-8",
        "hpgid": "200710",
        "scid": "100104",
        "uaid": uaid,
        "uiflvr": "1001",
        "x-requested-with": "XMLHttpRequest",
    }


class PasswordAction:
    name = "password"

    async def run(self, session, ctx) -> ActionResult:
        password = new_password()
        try:
            ticket = await _get_password_ticket(session.http)
            if not ticket:
                return ActionResult(self.name, False, detail="password_ticket_failed")

            async with session.http.request(
                "POST",
                _CHANGE_URL,
                data=json.dumps(_change_payload(ticket, password, _UAID)),
                headers=_change_headers(session.apicanary, _UAID),
            ) as resp:
                if resp.status >= 400:
                    return ActionResult(
                        self.name, False, detail=f"change_password_http_{resp.status}"
                    )
        except Exception as exc:
            ctx.log(f"password action failed for {session.email[:3]}***: {exc}")
            return ActionResult(self.name, False, detail=f"{type(exc).__name__}: {exc}")

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