secure/auth/twofa.py
import json
import logging
import re
import secrets
import time
from urllib.parse import quote, urljoin
from secure.auth.mslogin import finalize_session
from secure.auth.otp import AuthOutcome, _cookie_header, _cookie_pairs, _jar_cookie
from secure.generators import totp_now
logger = logging.getLogger("secure.auth.twofa")
_LOGIN_URL = "https://login.live.com"
_ACCOUNT_URL = "https://account.live.com"
_CHECKPASSWORD_URL = f"{_LOGIN_URL}/checkpassword.srf"
_DEFAULT_LINK = f"{_LOGIN_URL}/ppsecure/post.srf"
_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"
)
_SFT_PATTERNS = [
re.compile(r'<input[^>]*name=["\']PPFT["\']\s+[^>]*value=["\']([^"\']+)["\']'),
re.compile(r"sFT:'(.*?)'"),
re.compile(r'"sFT":"([^"]*?)"'),
]
_URL_POST_RE = re.compile(r'"urlPost":"([^"]+)"')
def _extract_first(patterns, html: str) -> str | None:
if not html:
return None
for pat in patterns:
m = pat.search(html)
if m:
return m.group(1)
return None
def _extract_login_state(html: str) -> dict:
state = {"proofs": [], "url_post": None, "sft": None}
state["sft"] = _extract_first(_SFT_PATTERNS, html)
up = _URL_POST_RE.search(html or "")
if up:
state["url_post"] = up.group(1).replace("\\/", "/").replace("\\u0026", "&")
ap = re.search(r'"arrUserProofs":(\[.*?\}\])', html or "", re.DOTALL)
if ap:
try:
state["proofs"] = json.loads(ap.group(1).replace("\\/", "/"))
except Exception:
state["proofs"] = []
return state
def _extract_auto_post_form(html: str) -> dict | None:
if not html:
return None
fm = re.search(r'<form\b[^>]*\baction=["\']([^"\']+)["\'][^>]*>([\s\S]*?)</form>', html, re.I)
if not fm:
return None
action = fm.group(1).replace("&", "&")
inner = fm.group(2)
fields = {}
for tag in re.findall(r"<input\b[^>]*?>", inner, re.I):
nm = re.search(r'\bname=["\']([^"\']*)["\']', tag)
if not nm:
continue
vm = re.search(r'\bvalue=["\']([^"\']*)["\']', tag)
fields[nm.group(1)] = (vm.group(1) if vm else "").replace("&", "&")
return {"action": action, "fields": fields}
def _abs(url: str, base: str) -> str:
return urljoin(base, url)
async def _get_live_data_2fa(http, proxy) -> dict | None:
wreply = quote(f"{_ACCOUNT_URL}/proofs/Manage/additional", safe="")
url = (
f"{_LOGIN_URL}/login.srf?wa=wsignin1.0&rpsnv=200&ct={int(time.time())}"
f"&rver=7.5.2211.0&wp=SA_20MIN&wreply={wreply}&lc=1033&id=38936&mkt=en-US"
)
async with http.get(url, allow_redirects=True, proxy=proxy) as resp:
if resp.status < 200 or resp.status >= 400:
return None
html = await resp.text()
cookies = _cookie_pairs(resp)
link_match = _LINK_RE.search(html)
login_link = link_match.group(0) if link_match else None
if not login_link:
up = _URL_POST_RE.search(html)
if up:
login_link = up.group(1).replace("\\/", "/").replace("\\u0026", "&")
if not login_link:
login_link = _DEFAULT_LINK
ppft = _extract_first(_SFT_PATTERNS, html)
if not ppft:
pm = re.search(r'value=(?:"([^"]+)"|\\"([^"]+)\\")', html)
if pm:
ppft = pm.group(1) or pm.group(2)
if not ppft:
return None
return {"ppft": ppft, "login_link": login_link, "cookies": cookies}
async def _vanguard_flow_token(http, email: str, password: str, cookies: dict, proxy) -> str | None:
uaid = cookies.get("uaid")
rid = uaid or secrets.token_hex(16)
headers = {
"accept": "application/json",
"content-type": "application/json; charset=utf-8",
"client-request-id": rid,
"correlationid": rid,
"hpgact": "0",
"hpgid": "33",
}
cookie_header = _cookie_header(cookies)
if cookie_header:
headers["Cookie"] = cookie_header
body = json.dumps({"username": email, "password": password, "checkpasswordflowtoken": ""})
try:
async with http.post(_CHECKPASSWORD_URL, data=body, headers=headers, proxy=proxy) as resp:
if resp.status != 200:
return None
payload = await resp.json()
cookies.update(_cookie_pairs(resp))
except Exception as exc:
logger.error("_vanguard_flow_token failed for %s***: %s", email[:3], exc)
return None
ok = (payload.get("validationresult") or payload.get("validationResult") or "").lower() == "succeed"
if not ok:
return None
return payload.get("vanguardflowtoken") or payload.get("vanguardFlowToken")
async def _submit_password(http, email: str, password: str, ppft: str, login_link: str,
vanguard: str, cookies: dict, proxy) -> str | None:
body = (
"ps=2&psRNGCDefaultType=&psRNGCEntropy=&psRNGCSLK=&canary=&ctx=&hpgrequestid="
f"&PPFT={quote(ppft)}&PPSX=PassportRN&NewUser=1&FoundMSAs=&fspost=1&i21=0"
"&CookieDisclosure=0&IsFidoSupported=1&isSignupPost=0&isRecoveryAttemptPost=0&i13=0"
f"&login={quote(email)}&loginfmt={quote(email)}&type=11&LoginOptions=3&lrt=&lrtPartition="
f"&hisRegion=&hisScaleUnit=&cpr=0&passwd={quote(password)}"
f"&vanguardflowtoken={quote(vanguard)}"
)
headers = {"Content-Type": "application/x-www-form-urlencoded"}
cookie_header = _cookie_header(cookies)
if cookie_header:
headers["Cookie"] = cookie_header
try:
async with http.post(
login_link, data=body, headers=headers, allow_redirects=False, proxy=proxy
) as resp:
if resp.status < 200 or resp.status >= 400:
return None
html = await resp.text()
cookies.update(_cookie_pairs(resp))
return html
except Exception as exc:
logger.error("_submit_password failed for %s***: %s", email[:3], exc)
return None
async def _submit_totp(http, email: str, otp_code: str, proof_data, ppft: str, url_post: str,
cookies: dict, proxy) -> dict | None:
body = (
f"AddTD=true&SentProofIDE={quote(str(proof_data))}&GeneralVerify=false"
f"&PPFT={quote(ppft)}&canary=&sacxt=1&hpgrequestid=&hideSmsInMfaProofs=false"
f"&type=19&login={quote(email)}&infoPageShown=0&otc={quote(str(otp_code))}"
)
headers = {"Content-Type": "application/x-www-form-urlencoded"}
cookie_header = _cookie_header(cookies)
if cookie_header:
headers["Cookie"] = cookie_header
try:
async with http.post(
url_post or _DEFAULT_LINK, data=body, headers=headers, allow_redirects=True, proxy=proxy
) as resp:
if resp.status < 200 or resp.status >= 400:
return None
html = await resp.text()
cookies.update(_cookie_pairs(resp))
return {"html": html, "url": str(resp.url)}
except Exception as exc:
logger.error("_submit_totp failed for %s***: %s", email[:3], exc)
return None
async def _navigate_post_auth(http, html: str, cur_url: str, cookies: dict, proxy) -> str | None:
for _hop in range(10):
auth = cookies.get("__Host-MSAAUTH") or _jar_cookie(http, cur_url, "__Host-MSAAUTH")
if auth:
return auth
form = _extract_auto_post_form(html)
if form:
action = _abs(form["action"], cur_url)
body = "&".join(f"{quote(k)}={quote(v)}" for k, v in form["fields"].items())
headers = {"Content-Type": "application/x-www-form-urlencoded"}
cookie_header = _cookie_header(cookies)
if cookie_header:
headers["Cookie"] = cookie_header
try:
async with http.post(action, data=body, headers=headers, allow_redirects=True, proxy=proxy) as resp:
if resp.status < 200 or resp.status >= 400:
return None
html = await resp.text()
cur_url = str(resp.url)
cookies.update(_cookie_pairs(resp))
except Exception:
return None
continue
up = _URL_POST_RE.search(html or "")
sft2 = _extract_first(_SFT_PATTERNS, html or "")
if up and sft2:
post_url = _abs(up.group(1).replace("\\/", "/").replace("\\u0026", "&"), cur_url)
body = f"PPFT={quote(sft2)}&canary=&LoginOptions=3&type=28&hpgrequestid=&ctx="
headers = {
"Content-Type": "application/x-www-form-urlencoded",
"Origin": _LOGIN_URL,
"Referer": cur_url,
}
cookie_header = _cookie_header(cookies)
if cookie_header:
headers["Cookie"] = cookie_header
try:
async with http.post(post_url, data=body, headers=headers, allow_redirects=True, proxy=proxy) as resp:
if resp.status < 200 or resp.status >= 400:
return None
html = await resp.text()
cur_url = str(resp.url)
cookies.update(_cookie_pairs(resp))
except Exception:
return None
continue
return None
return cookies.get("__Host-MSAAUTH") or _jar_cookie(http, cur_url, "__Host-MSAAUTH")
async def _login_2fa_full(http, email: str, password: str, otp_code: str, proxy) -> str | None:
try:
live = await _get_live_data_2fa(http, proxy)
if not live:
return None
cookies = live["cookies"]
vanguard = await _vanguard_flow_token(http, email, password, cookies, proxy)
if not vanguard:
return None
pwd_html = await _submit_password(http, email, password, live["ppft"], live["login_link"], vanguard, cookies, proxy)
if pwd_html is None:
return None
state = _extract_login_state(pwd_html)
totp_proof = next((p for p in state["proofs"] if p.get("type") == 10), None)
if not totp_proof:
totp_proof = next(
(p for p in state["proofs"] if "authenticator" in str(p.get("display", "")).lower()), None
)
if not totp_proof:
auth = cookies.get("__Host-MSAAUTH")
if auth:
return auth
return None
sft = state["sft"] or live["ppft"]
submitted = await _submit_totp(
http, email, otp_code, totp_proof.get("data"), sft, state["url_post"] or live["login_link"], cookies, proxy
)
if not submitted:
return None
return await _navigate_post_auth(http, submitted["html"], submitted["url"], cookies, proxy)
except Exception as exc:
logger.error("_login_2fa_full failed for %s***: %s", email[:3], exc)
return None
async def authenticate(creds: dict, http, proxy, cfg=None) -> AuthOutcome:
email = creds["email"]
password = creds["password"]
secret_key = (creds.get("secret_key") or "").replace(" ", "").upper()
try:
otp_code = totp_now(secret_key)
except Exception:
return AuthOutcome(None, "invalid_secret_key")
msaauth = await _login_2fa_full(http, email, password, otp_code, proxy)
if not msaauth:
return AuthOutcome(None, "login_failed")
session = await finalize_session(email, http, msaauth)
return AuthOutcome(session, None)