secure/pipeline/actions/profile.py
import asyncio
import logging
from secure.types import ActionResult, MSSession, PipelineContext
logger = logging.getLogger("secure.pipeline.actions.profile")
PROFILE_URL = "https://api.minecraftservices.com/minecraft/profile"
NAMECHANGE_URL = "https://api.minecraftservices.com/minecraft/profile/namechange"
ENTITLEMENTS_URL = (
"https://api.minecraftservices.com/entitlements/license"
"?requestId=c24114ab-1814-4d5c-9b1f-e8825edaec1f"
)
_MAX_RETRIES = 3
_MC_PRODUCTS = ("product_minecraft", "game_minecraft")
_GAMEPASS_MARKS = ("GAMEPASS", "GAME_PASS", "XGP")
def _auth(session: MSSession) -> dict:
return {"Authorization": f"Bearer {session.access_token}"}
def _skin_url(data: dict) -> str | None:
skins = data.get("skins") or []
active = next((s for s in skins if s.get("state") == "ACTIVE"), None)
skin = active or (skins[0] if skins else None)
return (skin or {}).get("url")
def _capes(data: dict) -> str | None:
capes = data.get("capes") or []
aliases = [c.get("alias") for c in capes if isinstance(c, dict) and c.get("alias")]
return ", ".join(aliases) if aliases else None
def _ownership_label(data: dict) -> str:
items = (data or {}).get("items") or []
has_purchase = has_gamepass = False
for item in items:
if not isinstance(item, dict) or item.get("name") not in _MC_PRODUCTS:
continue
source = str(item.get("source") or "").upper()
if "PURCHASE" in source:
has_purchase = True
elif any(mark in source for mark in _GAMEPASS_MARKS):
has_gamepass = True
if has_purchase:
return "Purchased"
if has_gamepass:
return "Xbox Game Pass"
return "Not purchased"
async def _fetch_name_change_allowed(session: MSSession) -> bool | None:
try:
async with session.http.request(
"GET", NAMECHANGE_URL, headers=_auth(session)
) as resp:
if resp.status != 200:
return None
data = await resp.json()
if isinstance(data, dict) and "nameChangeAllowed" in data:
return bool(data.get("nameChangeAllowed"))
except Exception as exc:
logger.info("namechange fetch failed: %s", exc)
return None
async def _fetch_ownership(session: MSSession) -> str | None:
try:
async with session.http.request(
"GET", ENTITLEMENTS_URL, headers=_auth(session)
) as resp:
if resp.status == 404:
return "Not purchased"
if resp.status != 200:
return None
data = await resp.json()
return _ownership_label(data if isinstance(data, dict) else {})
except Exception as exc:
logger.info("entitlements fetch failed: %s", exc)
return None
class ProfileAction:
name = "profile"
async def run(self, session: MSSession, ctx: PipelineContext) -> ActionResult:
for attempt in range(_MAX_RETRIES):
try:
async with session.http.request(
"GET",
PROFILE_URL,
headers=_auth(session),
) as resp:
if resp.status == 200:
try:
data = await resp.json()
except Exception:
data = {}
if not isinstance(data, dict) or not data.get("name"):
return ActionResult("profile", False, detail="no_mc_profile")
name_change = await _fetch_name_change_allowed(session)
ownership = await _fetch_ownership(session)
return ActionResult(
"profile",
True,
data={
"mc_username": data.get("name"),
"mc_uuid": data.get("id", ""),
"skin_url": _skin_url(data),
"mc_capes": _capes(data),
"mc_ownership": ownership,
"mc_name_change_allowed": name_change,
},
)
if resp.status == 404:
return ActionResult("profile", False, detail="no_mc_profile")
if resp.status in (400, 401, 403):
return ActionResult("profile", False, detail="failed")
if resp.status == 429 and attempt < _MAX_RETRIES - 1:
await asyncio.sleep(1)
continue
except Exception as exc:
logger.error("profile fetch failed for %s***: %s", session.email[:3], exc)
if attempt < _MAX_RETRIES - 1:
await asyncio.sleep(1)
return ActionResult("profile", False, detail="failed")