sten.wtf / Unstable Instance

secure/auth/mslogin.py

import base64
import json
import logging
import re
from urllib.parse import unquote, urljoin

import aiohttp

from secure.types import MSSession

logger = logging.getLogger("secure.auth.mslogin")

_SISU_CONNECT_URL = (
    "https://sisu.xboxlive.com/connect/XboxLive/"
    "?state=login"
    "&cobrandId=8058f65d-ce06-4c30-9559-473c9275a65d"
    "&tid=896928775"
    "&ru=https://www.minecraft.net/en-us/login"
    "&aid=1142970254"
)
_MAX_SISU_HOPS = 6
_REDIRECT_STATUSES = (301, 302, 303, 307, 308)
_MC_RELYING_PARTY = "rp://api.minecraftservices.com/"
_ACCESS_TOKEN_RE = re.compile(r"accessToken=([^&]+)")

_MC_LOGIN_URL = "https://api.minecraftservices.com/authentication/login_with_xbox"

_USER_AGENT = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36"
)

_BASE_HEADERS = {
    "User-Agent": _USER_AGENT,
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
    "Accept-Language": "en-US,en;q=0.9",
    "Accept-Encoding": "gzip, deflate, br",
    "Connection": "keep-alive",
    "Upgrade-Insecure-Requests": "1",
}


def _bind_session_proxy(session: aiohttp.ClientSession, proxy: str | None) -> None:
    """Every verb on this session must use `proxy` (pipeline actions omit it)."""
    session.proxy = proxy
    if not proxy:
        return
    orig = session._request

    async def _request(method, url, **kwargs):
        if kwargs.get("proxy") is None:
            kwargs["proxy"] = proxy
        return await orig(method, url, **kwargs)

    session._request = _request


async def open_http(proxy: str | None) -> aiohttp.ClientSession:
    proxy = proxy or None
    kwargs = {
        "headers": _BASE_HEADERS,
        "timeout": aiohttp.ClientTimeout(total=30),
        "trust_env": False,
    }
    try:
        session = aiohttp.ClientSession(proxy=proxy, **kwargs)
    except TypeError:
        session = aiohttp.ClientSession(**kwargs)
    _bind_session_proxy(session, proxy)
    return session


def _proxy_of(http) -> str | None:
    return getattr(http, "proxy", None)


def _extract_xsts_uhs(token_value: str | None) -> tuple[str | None, str | None]:
    if not token_value:
        return None, None
    raw = unquote(token_value)
    raw += "=" * ((4 - len(raw) % 4) % 4)
    try:
        entries = json.loads(base64.b64decode(raw))
    except Exception:
        return None, None
    if not isinstance(entries, list):
        return None, None
    for entry in entries:
        if not isinstance(entry, dict) or entry.get("Item1") != _MC_RELYING_PARTY:
            continue
        item2 = entry.get("Item2") or {}
        token = item2.get("Token")
        try:
            uhs = item2["DisplayClaims"]["xui"][0]["uhs"]
        except (KeyError, IndexError, TypeError):
            uhs = None
        return (token, uhs) if token and uhs else (None, None)
    return None, None


async def _sisu_xbl(http) -> tuple[str | None, str | None]:
    url = _SISU_CONNECT_URL
    for _ in range(_MAX_SISU_HOPS):
        try:
            async with http.get(
                url, allow_redirects=False, proxy=_proxy_of(http)
            ) as resp:
                if resp.status not in _REDIRECT_STATUSES:
                    return None, None
                location = resp.headers.get("Location")
        except Exception:
            return None, None
        if not location:
            return None, None
        if "#" in location and "accessToken=" in location:
            fragment = location.split("#", 1)[1]
            match = _ACCESS_TOKEN_RE.search(fragment)
            return _extract_xsts_uhs(match.group(1) if match else None)
        url = urljoin(url, location)
    return None, None


async def _minecraft_token(http, xsts_token, user_hash) -> str | None:
    if not xsts_token or not user_hash:
        return None
    payload = {
        "identityToken": f"XBL3.0 x={user_hash};{xsts_token}",
        "ensureLegacyEnabled": True,
    }
    try:
        async with http.post(
            _MC_LOGIN_URL,
            json=payload,
            headers={"Content-Type": "application/json", "Accept": "application/json"},
            proxy=_proxy_of(http),
        ) as resp:
            if resp.status != 200:
                return None
            body = await resp.json()
            return body.get("access_token")
    except Exception:
        return None


async def finalize_session(email: str, http, msaauth) -> MSSession:
    xsts_token, user_hash = await _sisu_xbl(http)
    mc_token = await _minecraft_token(http, xsts_token, user_hash)
    return MSSession(
        email=email,
        http=http,
        access_token=mc_token,
        xsts_token=xsts_token,
        user_hash=user_hash,
    )