secure/auth/recovery.py
import json
import logging
import re
from urllib.parse import quote, quote_plus, unquote
from mail.imap_reader import imap_cfg_from_instance, wait_for_code
from secure.auth import twofa
from secure.auth.mslogin import finalize_session
from secure.auth.otp import (
AuthOutcome,
_cookie_pairs,
_get_live_data,
_jar_cookie,
)
from secure.generators import new_password
logger = logging.getLogger("secure.auth.recovery")
_RESET_URL = (
"https://account.live.com/ResetPassword.aspx"
"?wreply=https://login.live.com/oauth20_authorize.srf&mn={email}"
)
_VERIFY_URL = "https://account.live.com/API/Recovery/VerifyRecoveryCode"
_RECOVER_URL = "https://account.live.com/API/Recovery/RecoverUser"
_GET_CRED_TYPE_URL = (
"https://login.live.com/GetCredentialType.srf"
"?opid=492674B18DE5DC3A&id=38936&mkt=EN-US&lc=1033&uaid={uaid}"
)
_GET_ONE_TIME_CODE_URL = "https://login.live.com/GetOneTimeCode.srf"
_LOGIN_POST_DEFAULT = "https://login.live.com/ppsecure/post.srf"
_DEFAULT_UAID = "eceb73c1b9fd43f2b41907bd9e1ba059"
# RSA public key + scenario/flavour ids Microsoft's recovery endpoints expect.
_PUB_KEY = "25CE4D96CB3A09A69CD847C69FC6D40AF4A4DE12"
_SCID = 100103
_UIFLVR = 1001
_SERVERDATA_RE = re.compile(r"var\s+ServerData\s*=\s*(\{.*?\})(?:;|\n)", re.S)
_OTP_RE = re.compile(r"^\d{6,7}$")
def _server_data(html: str) -> dict | None:
m = _SERVERDATA_RE.search(html or "")
if not m:
return None
try:
return json.loads(m.group(1))
except Exception:
return None
def _clear_jar(http) -> None:
"""Empty aiohttp's cookie jar. Its CookieJar re-quotes account.live.com's
`amsc` value (full of +/=/: chars) and merges that corrupted copy over any
manual Cookie header, which MS rejects with a generic 500. Clearing it before
each recovery write lets the clean, manually-forwarded cookies go out intact;
the jar refills from the post-reset login for SISU/pipeline afterwards.
"""
jar = getattr(http, "cookie_jar", None)
if jar is not None:
try:
jar.clear()
except Exception:
pass
def _merge_cookies(pairs: dict, resp) -> None:
"""Accumulate Set-Cookie across a response's redirect history + final headers
into `pairs`. account.live.com's recovery writes need the cookies gathered so
far (GET *and* the VerifyRecoveryCode response), forwarded manually because
aiohttp's jar both drops the empty `IPT` cookie and re-quotes `amsc` into a
value MS rejects (see _clear_jar). Mirrors the reference's manual cookie store.
"""
for r in list(getattr(resp, "history", ())) + [resp]:
try:
headers = r.headers.getall("Set-Cookie", [])
except Exception:
headers = []
for h in headers:
name, _, rest = h.partition("=")
if name.strip():
pairs[name.strip()] = rest.split(";", 1)[0]
def _cookie_header(pairs: dict) -> str:
return "; ".join(f"{k}={v}" for k, v in pairs.items()) + "; " if pairs else ""
async def _fetch_server_data(http, email: str, proxy) -> tuple[dict | None, dict]:
url = _RESET_URL.format(email=quote_plus(email))
cookies: dict[str, str] = {}
try:
async with http.get(url, proxy=proxy) as resp:
html = await resp.text()
_merge_cookies(cookies, resp)
except Exception as exc:
logger.error("_fetch_server_data failed for %s***: %s", email[:3], exc)
return None, {}
sd = _server_data(html)
if not sd or not sd.get("sRecoveryToken") or not sd.get("apiCanary"):
return None, {}
return sd, cookies
async def _verify_recovery_code(http, email: str, code: str, sd: dict, cookies: dict, proxy) -> tuple[str | None, str | None]:
"""POST VerifyRecoveryCode. Returns (recover_token, error). Header canary is
lowercase and is the ServerData apiCanary; success is a non-empty `token`.
"""
payload = {
"recoveryCode": code,
"scid": _SCID,
"token": unquote(sd.get("sRecoveryToken", "")),
"uaid": sd.get("sUnauthSessionID", ""),
"uiflvr": _UIFLVR,
"code": code,
}
headers = {
"Content-Type": "application/json; charset=utf-8",
"X-Requested-With": "XMLHttpRequest",
"canary": sd.get("apiCanary", ""),
}
if cookies:
headers["Cookie"] = _cookie_header(cookies)
_clear_jar(http)
try:
async with http.post(_VERIFY_URL, data=json.dumps(payload), headers=headers, proxy=proxy) as resp:
_merge_cookies(cookies, resp) # RecoverUser needs the verify-response cookies too
if resp.status >= 400:
return None, "verify_failed"
try:
body = await resp.json()
except Exception:
body = None
except Exception as exc:
logger.error("_verify_recovery_code failed for %s***: %s", email[:3], exc)
return None, "verify_failed"
if not isinstance(body, dict):
return None, "verify_failed"
if body.get("token"):
return body["token"], None
err = (body.get("error") or {}).get("code") if isinstance(body.get("error"), dict) else None
if err == "6001":
return None, "tfa_enabled"
if err == "1300":
return None, "invalid_recovery_code"
return None, "verify_failed" # 500 / unknown — transient (often IP/exit), not a bad code
def _recover_payload(token: str, security_email: str | None, password: str) -> dict:
return {
"contactEmail": security_email or "",
"contactEpid": "",
"password": password,
"passwordExpiryEnabled": 0,
"publicKey": _PUB_KEY,
"token": unquote(token),
}
async def _recover_user(http, verify_token: str, canary: str, security_email: str | None, password: str, cookies: dict, proxy) -> str | None:
"""POST RecoverUser (reset password + attach `security_email` as a proof).
Header Canary is capitalised and is the ServerData apiCanary; the body token
is the VerifyRecoveryCode response token. Returns 'ok'/'same'/'tfa'/None.
"""
headers = {
"Content-Type": "application/json; charset=utf-8",
"X-Requested-With": "XMLHttpRequest",
"Canary": canary,
}
if cookies:
headers["Cookie"] = _cookie_header(cookies)
payload = _recover_payload(verify_token, security_email, password)
_clear_jar(http)
try:
async with http.post(_RECOVER_URL, data=json.dumps(payload), headers=headers, proxy=proxy) as resp:
if resp.status >= 400:
return None
try:
body = await resp.json()
except Exception:
body = None
except Exception as exc:
logger.error("_recover_user failed: %s", exc)
return None
if not isinstance(body, dict):
return None
err = (body.get("error") or {}).get("code") if isinstance(body.get("error"), dict) else None
if err == "6001":
return "tfa"
if err == "1218": # same creds already set — idempotent replay, treat as success
return "same"
if body.get("recoveryCode") or body.get("apiCanary"):
return "ok"
return None
def _cred_body(email: str, ppft: str, uaid: str) -> dict:
return {
"checkPhones": True,
"country": "",
"federationFlags": 3,
"flowToken": ppft,
"forceotclogin": False,
"isCookieBannerShown": False,
"isExternalFederationDisallowed": False,
"isFederationDisabled": False,
"isFidoSupported": True,
"isOtherIdpSupported": False,
"isRemoteConnectSupported": False,
"isRemoteNGCSupported": True,
"isSignup": False,
"originalRequest": "",
"otclogindisallowed": False,
"uaid": uaid,
"username": email,
}
def _parse_proofs(data) -> dict:
empty = {"otp_proofs": [], "no_password": False, "invalid_email": False, "tfa_eligible": False}
if not isinstance(data, dict):
return empty
creds = data.get("Credentials") if isinstance(data.get("Credentials"), dict) else {}
otp_proofs = creds.get("OtcLoginEligibleProofs") or []
ngc = creds.get("RemoteNgcParams")
return {
"otp_proofs": otp_proofs if isinstance(otp_proofs, list) else [],
"no_password": creds.get("HasPassword") == 0,
"invalid_email": data.get("IfExistsResult") == 1,
"tfa_eligible": (not otp_proofs) and (not ngc),
}
async def _get_proofs(http, email: str, proxy) -> dict:
live = await _get_live_data(http, proxy)
ppft = live.get("ppft")
uaid = live.get("uaid") or _DEFAULT_UAID
if not ppft or ppft == "null":
return _parse_proofs(None)
url = _GET_CRED_TYPE_URL.format(uaid=uaid)
headers = {
"accept": "application/json",
"Content-Type": "application/json",
"X-Requested-With": "XMLHttpRequest",
}
if live.get("cookies"):
headers["Cookie"] = live["cookies"]
try:
async with http.post(url, data=json.dumps(_cred_body(email, ppft, uaid)), headers=headers, proxy=proxy) as resp:
if resp.status != 200:
return _parse_proofs(None)
body = await resp.json()
except Exception as exc:
logger.error("_get_proofs failed for %s***: %s", email[:3], exc)
return _parse_proofs(None)
return _parse_proofs(body)
async def _trigger_otp(http, email: str, proof_data, proxy) -> bool:
"""Ask Microsoft to email a one-time code to the proof (the attached
security email). Fresh flowtoken/cookies rather than a hardcoded blob.
"""
live = await _get_live_data(http, proxy)
ppft = live.get("ppft")
if not ppft or ppft == "null":
return False
data = (
f"login={quote(email)}&flowtoken={quote(ppft)}&purpose=eOTT_OtcLogin&channel=Email"
f"&ChallengeViewSupported=1&uaid={live.get('uaid')}&AltEmailE={quote(str(proof_data))}&lcid=1033"
)
headers = {"Content-Type": "application/x-www-form-urlencoded", "Cookie": live.get("cookies", "")}
try:
async with http.post(_GET_ONE_TIME_CODE_URL, data=data, headers=headers, proxy=proxy) as resp:
if resp.status >= 400:
return False
try:
body = await resp.json()
except Exception:
body = {}
except Exception as exc:
logger.error("_trigger_otp failed for %s***: %s", email[:3], exc)
return False
return isinstance(body, dict) and body.get("State") in (200, 201)
async def _login_with_otp_proof(http, email: str, code: str, proof_data, no_password: bool, proxy) -> str | None:
"""POST the emailed code against the proof to obtain __Host-MSAAUTH.
type=27 when the account has a password, type=24 (npotc) when it doesn't.
"""
if not _OTP_RE.fullmatch(str(code)):
return None
live = await _get_live_data(http, proxy)
ppft = live.get("ppft")
if not ppft or ppft == "null":
return None
link = live.get("loginLink", _LOGIN_POST_DEFAULT)
if no_password:
payload = (
f"SentProofIDE={quote_plus(str(proof_data))}&ProofType=1&npotc={quote_plus(str(code))}"
"&ps=3&psRNGCDefaultType=&psRNGCEntropy=&psRNGCSLK=&canary=&ctx=&hpgrequestid="
f"&PPFT={quote_plus(ppft)}&PPSX=Pass&NewUser=1&FoundMSAs=&fspost=0&i21=0"
"&CookieDisclosure=0&IsFidoSupported=1&isSignupPost=0&isRecoveryAttemptPost=0&i13=0"
f"&login={quote_plus(email)}&loginfmt={quote_plus(email)}&type=24&LoginOptions=3"
"&lrt=&lrtPartition=&hisRegion=&hisScaleUnit="
)
else:
payload = (
f"login={quote_plus(email)}&loginfmt={quote_plus(email)}&type=27"
f"&SentProofIDE={quote_plus(str(proof_data))}&otc={quote_plus(str(code))}&PPFT={quote_plus(ppft)}"
)
headers = {"Content-Type": "application/x-www-form-urlencoded"}
if live.get("cookies"):
headers["Cookie"] = live["cookies"]
try:
async with http.post(link, data=payload, headers=headers, allow_redirects=True, proxy=proxy) as resp:
if resp.status < 200 or resp.status >= 400:
return None
auth = _cookie_pairs(resp).get("__Host-MSAAUTH")
resp_url = str(resp.url)
except Exception as exc:
logger.error("_login_with_otp_proof failed for %s***: %s", email[:3], exc)
return None
return auth or _jar_cookie(http, resp_url, "__Host-MSAAUTH")
async def _password_login(http, email: str, password: str, proxy) -> str | None:
"""Direct password login with the just-reset password (no 2FA). Reuses the
twofa module's vanguard/password-submit/interstitial-navigation machinery.
"""
live = await twofa._get_live_data_2fa(http, proxy)
if not live:
return None
cookies = live["cookies"]
vanguard = await twofa._vanguard_flow_token(http, email, password, cookies, proxy)
if not vanguard:
return None
pwd_html = await twofa._submit_password(
http, email, password, live["ppft"], live["login_link"], vanguard, cookies, proxy
)
if pwd_html is None:
return None
auth = cookies.get("__Host-MSAAUTH")
if auth:
return auth
return await twofa._navigate_post_auth(http, pwd_html, live["login_link"], cookies, proxy)
async def _post_reset_login(http, email: str, password: str, security_email: str | None, cfg, proxy) -> str | None:
"""Reach __Host-MSAAUTH after RecoverUser. Primary path emails a code to the
security mailbox (needs cfg IMAP); the fallback logs in with the new password.
"""
imap_cfg = imap_cfg_from_instance(cfg) if cfg is not None else None
if security_email and imap_cfg:
proofs = await _get_proofs(http, email, proxy)
if proofs.get("otp_proofs"):
no_password = proofs.get("no_password", False)
for proof in proofs["otp_proofs"]:
data = proof.get("data") if isinstance(proof, dict) else None
if not data:
continue
if not await _trigger_otp(http, email, data, proxy):
continue
code = await wait_for_code(imap_cfg, since_seen=set(), timeout=25, poll=3)
if not code:
continue
msaauth = await _login_with_otp_proof(http, email, code, data, no_password, proxy)
if msaauth:
return msaauth
return await _password_login(http, email, password, proxy)
async def authenticate(creds: dict, http, proxy, cfg=None) -> AuthOutcome:
"""Recovery-code method: reset the password with the recovery code, then log
in to mint a live session. `cfg` must carry a `security_email` — RecoverUser
attaches it as the account's contact proof (an empty one is rejected with MS
error 1284) and the post-reset login reads its inbox (needs cfg IMAP too;
without IMAP it falls back to a password login).
"""
email = creds["email"]
recovery_code = creds["recovery_code"]
security_email = getattr(cfg, "security_email", None)
if not security_email:
return AuthOutcome(None, "security_email_required")
password = new_password()
sd, cookies = await _fetch_server_data(http, email, proxy)
if not sd:
return AuthOutcome(None, "server_data_failed")
verify_token, verr = await _verify_recovery_code(http, email, recovery_code, sd, cookies, proxy)
if not verify_token:
return AuthOutcome(None, verr or "invalid_recovery_code")
recovered = await _recover_user(http, verify_token, sd.get("apiCanary", ""), security_email, password, cookies, proxy)
if recovered == "tfa":
return AuthOutcome(None, "tfa_enabled")
if recovered not in ("ok", "same"):
return AuthOutcome(None, "recover_failed")
msaauth = await _post_reset_login(http, email, password, security_email, cfg, proxy)
if not msaauth:
return AuthOutcome(None, "login_failed")
session = await finalize_session(email, http, msaauth)
# RecoverUser set this password / security email; surface them so they're
# reported even if the pipeline's password/security-email actions no-op.
session.extra["new_password"] = password
if security_email:
session.extra["added_security_email"] = security_email
return AuthOutcome(session, None)