secure/auth/otp.py
import json
import logging
import re
from dataclasses import dataclass, field
from urllib.parse import quote, quote_plus
from yarl import URL
from secure.auth.mslogin import finalize_session
from secure.types import MSSession
logger = logging.getLogger("secure.auth.otp")
_LOGIN_URL = "https://login.live.com"
_GET_ONE_TIME_CODE_URL = "https://login.live.com/GetOneTimeCode.srf?id=38936"
_GET_CRED_TYPE_URL = (
"https://login.live.com/GetCredentialType.srf"
"?opid=492674B18DE5DC3A&id=38936&mkt=EN-US&lc=1033&uaid={uaid}"
)
_LINK_RE = re.compile(
r"https://login\.live\.com/ppsecure/post\.srf\?contextid=[0-9a-zA-Z]+&opid=[0-9a-zA-Z]+&bk=[a-zA-Z0-9]+&uaid=[0-9a-zA-Z]+&pid=0"
)
_PPFT_RE = re.compile(r'value=(?:"([^"]+)"|\\\"([^"]+)\\\")')
_DEFAULT_UAID = "eceb73c1b9fd43f2b41907bd9e1ba059"
_DEFAULT_LINK = "https://login.live.com/ppsecure/post.srf"
_OTP_RE = re.compile(r"^\d{6,7}$")
@dataclass
class AuthOutcome:
session: MSSession | None
fail_reason: str | None
@dataclass(frozen=True)
class ProofInfo:
display: str
data: str
proof_type: str # "email" | "phone"
@dataclass
class AccountCheck:
ok: bool
message: str
proofs: list[ProofInfo] = field(default_factory=list)
no_password: bool = False
authenticator: bool = False
invalid_email: bool = False
error: str | None = None
@dataclass
class OtcPlan:
step: str # "choose" | "send" | "error"
proofs: list[ProofInfo]
no_password: bool = False
reason: str | None = None
message: str = ""
@dataclass
class OtcSendResult:
ok: bool
message: str
reason: str | None = None
state: int | None = None
@dataclass
class OtcSession:
step: str # "choose" | "code" | "error"
proofs: list[ProofInfo]
selected: ProofInfo | None
no_password: bool
message: str
reason: str | None = None
def _proxy_of(http):
return getattr(http, "proxy", None)
def _cookie_pairs(resp) -> dict:
try:
raw = resp.headers.getall("Set-Cookie", [])
except Exception:
raw = []
out = {}
for header in raw:
name, _, rest = header.partition("=")
if rest:
out[name.strip()] = rest.split(";", 1)[0]
return out
def _cookie_header(pairs: dict) -> str:
if not pairs:
return ""
return "; ".join(f"{k}={v}" for k, v in pairs.items()) + "; "
def _jar_cookie(http, url: str, name: str) -> str | None:
jar = getattr(http, "cookie_jar", None)
if jar is None:
return None
try:
morsel = jar.filter_cookies(URL(url)).get(name)
except Exception:
return None
return morsel.value if morsel else None
async def _get_live_data(http, proxy) -> dict:
async with http.get(_LOGIN_URL, proxy=proxy) as resp:
html = await resp.text()
pairs = _cookie_pairs(resp)
link_match = _LINK_RE.search(html)
ppft_match = _PPFT_RE.search(html)
ppft = (ppft_match.group(1) or ppft_match.group(2)) if ppft_match else "null"
return {
"loginLink": link_match.group(0) if link_match else _DEFAULT_LINK,
"ppft": ppft,
"cookies": _cookie_header(pairs),
"uaid": pairs.get("uaid", _DEFAULT_UAID),
}
def proofs_from_raw(raw) -> list[ProofInfo]:
out: list[ProofInfo] = []
if not isinstance(raw, list):
return out
for item in raw:
if not isinstance(item, dict):
continue
data = str(item.get("data") or "")
if not data:
continue
display = str(item.get("display") or "?")
proof_type = (
"email" if item.get("type") == 1 or "@" in display else "phone"
)
out.append(ProofInfo(display=display, data=data, proof_type=proof_type))
out.sort(key=lambda p: p.proof_type != "email")
return out
def email_proofs(proofs: list[ProofInfo]) -> list[ProofInfo]:
return [p for p in proofs if p.proof_type == "email"]
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 plan_otc(check: AccountCheck) -> OtcPlan:
if check.authenticator:
return OtcPlan(
step="error",
proofs=[],
reason="authenticator",
message=(
"This account uses Microsoft Authenticator. "
"Use the 2FA method instead."
),
)
if not check.ok:
return OtcPlan(
step="error",
proofs=[],
reason=check.error or "invalid",
message=check.message
or "Couldn't send a verification code to that email. "
"Double-check it and try again.",
)
emails = email_proofs(check.proofs)
if not emails:
return OtcPlan(
step="error",
proofs=list(check.proofs),
no_password=check.no_password,
reason="no_proofs",
message="No security emails found for this account.",
)
if len(emails) == 1:
return OtcPlan(step="send", proofs=emails, no_password=check.no_password)
return OtcPlan(
step="choose",
proofs=emails,
no_password=check.no_password,
message="Choose where to send the one-time code.",
)
async def check_account(http, email: str, proxy=None) -> AccountCheck:
proxy = _proxy_of(http) if proxy is None else proxy
empty = AccountCheck(ok=False, message="Connection error. Try again.", error="connection")
try:
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 empty
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"]
async with http.post(
url,
data=json.dumps(_cred_body(email, ppft, uaid)),
headers=headers,
proxy=proxy,
) as resp:
if resp.status != 200:
return empty
body = await resp.json()
except Exception as exc:
logger.error("check_account failed for %s***: %s", email[:3], exc)
return empty
if not isinstance(body, dict):
return empty
creds = body.get("Credentials") if isinstance(body.get("Credentials"), dict) else None
if body.get("errorHR") or body.get("IfExistsResult", 0) not in (0, None) or not creds:
if body.get("IfExistsResult") == 1 or not creds:
return AccountCheck(
ok=False,
message="Couldn't find that Microsoft account.",
invalid_email=True,
error="not_found",
)
return empty
if creds.get("RemoteNgcParams"):
return AccountCheck(
ok=True,
message="This account uses Microsoft Authenticator.",
authenticator=True,
)
proofs = proofs_from_raw(creds.get("OtcLoginEligibleProofs") or [])
no_password = creds.get("HasPassword") == 0
if not proofs:
return AccountCheck(
ok=False,
message="No security emails found for this account.",
no_password=no_password,
error="no_proofs",
)
return AccountCheck(
ok=True,
message=f"Found {len(proofs)} method(s)",
proofs=proofs,
no_password=no_password,
)
async def send_otc(
http, email: str, proof: ProofInfo, no_password: bool = False
) -> OtcSendResult:
proxy = _proxy_of(http)
try:
live = await _get_live_data(http, proxy)
ppft = live.get("ppft")
if not ppft or ppft == "null":
return OtcSendResult(ok=False, message="Connection error.", reason="connection")
purpose = (
"eOTT_NoPasswordAccountLoginCode" if no_password else "eOTT_OtcLogin"
)
channel = "Phone" if proof.proof_type == "phone" else "Email"
proof_param = (
f"PhoneE={quote(proof.data)}"
if proof.proof_type == "phone"
else f"AltEmailE={quote(proof.data)}"
)
data = (
f"login={quote(email)}&flowtoken={quote(ppft)}&purpose={purpose}"
f"&channel={channel}&ChallengeViewSupported=1&uaid={live.get('uaid')}"
f"&{proof_param}&lcid=1033"
)
headers = {
"Content-Type": "application/x-www-form-urlencoded",
"Cookie": live.get("cookies", ""),
}
async with http.post(
_GET_ONE_TIME_CODE_URL, data=data, headers=headers, proxy=proxy
) as resp:
if resp.status >= 400:
return OtcSendResult(
ok=False, message="Failed to send code.", reason="http"
)
try:
body = await resp.json()
except Exception:
body = {}
if not isinstance(body, dict):
return OtcSendResult(
ok=False, message="Unexpected response.", reason="unexpected"
)
state = body.get("State")
if state in (200, 201):
return OtcSendResult(ok=True, message="Code sent", state=state)
if state == 429:
return OtcSendResult(
ok=False,
message="Too many attempts. Wait a bit and try again.",
reason="rate_limited",
state=429,
)
return OtcSendResult(
ok=False, message="Failed to send code.", reason="send_failed", state=state
)
except Exception as exc:
logger.error("send_otc failed for %s***: %s", email[:3], exc)
return OtcSendResult(ok=False, message="Error sending code.", reason="error")
async def request_otp(
http, email: str, proof: ProofInfo | None = None, *, no_password: bool = False
) -> bool:
try:
if proof is None:
check = await check_account(http, email)
emails = email_proofs(check.proofs)
if check.authenticator or not check.ok or not emails:
return False
proof = emails[0]
no_password = check.no_password
return (await send_otc(http, email, proof, no_password=no_password)).ok
except Exception as exc:
logger.error("request_otp failed for %s***: %s", email[:3], exc)
return False
async def begin_otp_session(http, email: str) -> OtcSession:
check = await check_account(http, email)
plan = plan_otc(check)
if plan.step == "error":
return OtcSession(
step="error",
proofs=plan.proofs,
selected=None,
no_password=plan.no_password,
message=plan.message,
reason=plan.reason,
)
if plan.step == "choose":
return OtcSession(
step="choose",
proofs=plan.proofs,
selected=None,
no_password=plan.no_password,
message=plan.message or "Choose where to send the one-time code.",
)
proof = plan.proofs[0]
sent = await send_otc(http, email, proof, no_password=plan.no_password)
if not sent.ok:
return OtcSession(
step="error",
proofs=plan.proofs,
selected=proof,
no_password=plan.no_password,
message=sent.message
or "Couldn't send a verification code to that email. "
"Double-check it and try again.",
reason=sent.reason or "send_failed",
)
return OtcSession(
step="code",
proofs=plan.proofs,
selected=proof,
no_password=plan.no_password,
message=sent.message,
)
async def _login_otp(
http,
email: str,
otp: str,
proxy,
proof_data: str | None = None,
no_password: bool = False,
) -> str | None:
if not _OTP_RE.fullmatch(otp):
return None
proof_id = proof_data or email
try:
live = await _get_live_data(http, proxy)
ppft = live.get("ppft")
if not ppft or ppft == "null":
return None
link = live.get("loginLink", _DEFAULT_LINK)
if no_password:
payload = (
f"SentProofIDE={quote_plus(str(proof_id))}&ProofType=1"
f"&npotc={quote_plus(otp)}"
"&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_id))}&otc={quote_plus(otp)}"
f"&PPFT={quote_plus(ppft)}"
)
headers = {"Content-Type": "application/x-www-form-urlencoded"}
if live.get("cookies"):
headers["Cookie"] = live["cookies"]
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)
return auth or _jar_cookie(http, resp_url, "__Host-MSAAUTH")
except Exception as exc:
logger.error("_login_otp failed for %s***: %s", email[:3], exc)
return None
async def authenticate(creds: dict, http, proxy, cfg=None) -> AuthOutcome:
email = creds["email"]
otp = creds["otp"]
proof_data = creds.get("proof_data") or email
no_password = bool(creds.get("no_password"))
msaauth = await _login_otp(
http, email, otp, proxy, proof_data=proof_data, no_password=no_password
)
if not msaauth:
return AuthOutcome(None, "invalid_otp")
session = await finalize_session(email, http, msaauth)
return AuthOutcome(session, None)