sten.wtf / Unstable Instance

secure/pipeline/actions/security_email.py

import re
from urllib.parse import urlencode

from mail.imap_reader import wait_for_code
from secure.types import ActionResult

_MANAGE_URL = "https://account.live.com/proofs/Manage/additional?uaid=07876c46b7514e38b8c1196f84d73982"
_ADD_URL = "https://account.live.com/proofs/Add?mkt=en-us&apt=2&mpsplit=2"
_VERIFY_SUCCESS_PATH = "/proofs/Manage/additional"
_ADD_PROOF_MARKER = "Enter the code we sent to"

_FORM_HEADERS = {"Content-Type": "application/x-www-form-urlencoded"}

_CANARY_RE = (
    re.compile(r'id="canary"[^>]*value="([^"]+)"'),
    re.compile(r'value="([^"]+)"[^>]*id="canary"'),
)
_OTT_RE = (
    re.compile(r'name="proof"[^>]*value="(OTT\|\|[^"]+)"'),
    re.compile(r'value="(OTT\|\|[^"]+)"[^>]*name="proof"'),
)
_FORM_ACTION_RE = (
    re.compile(r'<form[^>]*action="([^"]*(?:Verify|proofs)[^"]*)"', re.I),
    re.compile(r'<form[^>]*action="([^"]+)"'),
)


def _first_match(patterns, html):
    for pat in patterns:
        m = pat.search(html)
        if m:
            return m.group(1)
    return None


def _default_proof_opts(email: str) -> str:
    lead = email[0] if email else "c"
    return f"OTT||{email}||Email||0||{lead}"


def _errcode(text: str) -> str | None:
    if "errcode=" not in text:
        return None
    idx = text.find("errcode=")
    end = text.find("&", idx)
    if end == -1:
        end = text.find('"', idx)
    if end == -1:
        end = idx + 20
    return text[idx + 8 : end]


async def _fetch_canary(http) -> str | None:
    async with http.request("GET", _MANAGE_URL) as resp:
        html = await resp.text()
    return _first_match(_CANARY_RE, html)


async def _add_proof(http, email: str, canary: str) -> tuple[bool, str | None, str | None, str | None, str | None]:
    data = urlencode({"iProofOptions": "Email", "EmailAddress": email, "canary": canary, "action": "AddProof"})
    async with http.request("POST", _ADD_URL, data=data, headers=_FORM_HEADERS) as resp:
        if resp.status != 200:
            return False, "otp_send_failed", None, None, None
        html = await resp.text()
        resp_url = str(getattr(resp, "url", "") or "")

    if _ADD_PROOF_MARKER not in html:
        return False, "unexpected_add_page", None, None, None

    fresh = _first_match(_CANARY_RE, html)
    proof = _first_match(_OTT_RE, html)
    redirect = _first_match(_FORM_ACTION_RE, html)
    if not redirect and "proofs" in resp_url.lower():
        redirect = resp_url
    if redirect is None:
        return False, "otp_send_failed", fresh, proof, None
    return True, None, fresh, proof, redirect


async def _confirm_proof(http, email: str, otp: str, canary: str, proof_options: str | None, redirect_url: str) -> tuple[bool, str | None]:
    opts = proof_options or _default_proof_opts(email)
    data = urlencode({"iProofOptions": opts, "iOttText": otp, "action": "VerifyProof", "canary": canary, "GeneralVerify": "0"})
    async with http.request("POST", redirect_url, data=data, headers=_FORM_HEADERS, allow_redirects=False) as resp:
        location = resp.headers.get("location", "")
        if _VERIFY_SUCCESS_PATH in location:
            return True, None
        text = await resp.text()
    return False, _errcode(text)


class SecurityEmailAction:

    name = "security_email"

    async def run(self, session, ctx) -> ActionResult:
        email = ctx.security_email
        if not email or not ctx.imap:
            return ActionResult(self.name, False, detail="not_configured")

        canary = await _fetch_canary(session.http)
        if not canary:
            return ActionResult(self.name, False, detail="canary_not_found")

        sent, add_detail, fresh_canary, proof_options, redirect_url = await _add_proof(session.http, email, canary)
        if not sent:
            return ActionResult(self.name, False, detail=add_detail)

        ctx.log(f"waiting for security-email verification code for {email}...")
        imap_cfg = dict(ctx.imap)
        if ctx.proxy:
            imap_cfg.setdefault("proxy", ctx.proxy)
        code = await wait_for_code(imap_cfg, since_seen=set())
        if not code:
            return ActionResult(self.name, False, detail="verification_timeout")

        ctx.log(f"verification code received for {email}")
        ok, err = await _confirm_proof(session.http, email, code, fresh_canary or canary, proof_options, redirect_url)
        if not ok:
            detail = f"otp_verify_failed ({err})" if err else "otp_verify_failed"
            return ActionResult(self.name, False, detail=detail)

        return ActionResult(self.name, True, data={"security_email": email})