uni-notify/app/browser/login.py

"""Автоматический вход в Microsoft 365 своим логином/паролем с 2FA-кодом, который спрашивается в Telegram.

Ничего не обходит: заполняет обычную форму входа так же, как это сделал бы человек.
Для нестандартных SSO-страниц университета присылает скриншот — дальше можно
управлять браузером командами /click, /type, /key.
"""
from __future__ import annotations

import asyncio
import logging
from typing import Awaitable, Callable
from urllib.parse import urlparse

from playwright.async_api import Page

log = logging.getLogger(__name__)

MS_LOGIN_HOSTS = ("login.microsoftonline.com", "login.live.com", "login.microsoft.com", "account.live.com")

Say = Callable[..., Awaitable[None]]
Ask = Callable[..., Awaitable[str | None]]


def is_login_url(url: str) -> bool:
    host = urlparse(url).hostname or ""
    return any(host.endswith(h) for h in MS_LOGIN_HOSTS)


async def _visible(page: Page, sel: str) -> bool:
    try:
        loc = page.locator(sel).first
        return await loc.count() > 0 and await loc.is_visible()
    except Exception:
        return False


async def _text(page: Page, sel: str) -> str:
    try:
        loc = page.locator(sel).first
        if await loc.count() and await loc.is_visible():
            return (await loc.inner_text()).strip()
    except Exception:
        pass
    return ""


async def _click(page: Page, sel: str) -> None:
    await page.locator(sel).first.click(timeout=5000)


async def login_flow(page: Page, *, target_url: str, target_host: str, email: str,
                     password: str | None, say: Say, ask: Ask, timeout: float = 420) -> bool:
    """Ведёт страницу через вход Microsoft до target_host. True — вошли."""
    loop = asyncio.get_running_loop()
    deadline = loop.time() + timeout
    if not is_login_url(page.url) and target_host not in (urlparse(page.url).hostname or ""):
        await page.goto(target_url, wait_until="domcontentloaded")

    told_number = told_push = told_unknown = False
    unknown_since: float | None = None
    otc_tries = 0
    last_step = ""

    while loop.time() < deadline:
        await asyncio.sleep(1.5)
        url = page.url
        host = urlparse(url).hostname or ""

        # --- успех ---
        if target_host in host and not is_login_url(url):
            await asyncio.sleep(3)
            if target_host in (urlparse(page.url).hostname or ""):
                return True
            continue

        # --- нестандартная страница (SSO университета и т.п.) ---
        if not is_login_url(url):
            unknown_since = unknown_since or loop.time()
            if not told_unknown and loop.time() - unknown_since > 12:
                told_unknown = True
                shot = await page.screenshot()
                await say(
                    f"Нестандартная страница входа: {host}\n"
                    "Управляй браузером: /click X Y, /type текст, /key Enter, /screen — "
                    "я продолжу, как только вернёмся на Microsoft или в почту.",
                    photo=shot,
                )
            continue
        unknown_since = None

        # --- ошибки ---
        err = await _text(page, "#usernameError, #passwordError, #idSpan_SAOTCC_Error_OTC, #service_exception_message")
        if err and err != last_step:
            last_step = err
            await say(f"Microsoft пишет: {err}")
            if "password" in err.lower() or "пароль" in err.lower() or "hasło" in err.lower():
                password = None  # спросим заново

        # --- «Оставаться в системе?» ---
        if await _visible(page, "#KmsiCheckboxField") or await _visible(page, "#KmsiDescription"):
            if await _visible(page, "#KmsiCheckboxField"):
                try:
                    await page.locator("#KmsiCheckboxField").check(timeout=3000)
                except Exception:
                    pass
            await _click(page, "#idSIButton9")
            continue

        # --- ввод кода из приложения ---
        if await _visible(page, "input[name='otc']"):
            if otc_tries >= 3:
                await say("Три неверных кода подряд — останавливаюсь. /login чтобы начать заново.")
                return False
            code = await ask("🔐 Введи 6-значный код из приложения-аутентификатора:", timeout=240)
            if not code:
                await say("Код не пришёл — вход отменён.")
                return False
            otc_tries += 1
            await page.locator("input[name='otc']").first.fill(code.strip().replace(" ", ""))
            if await _visible(page, "#idSubmit_SAOTCC_Continue"):
                await _click(page, "#idSubmit_SAOTCC_Continue")
            else:
                await page.keyboard.press("Enter")
            try:  # ждём, пока поле исчезнет (успех) — иначе это ошибка кода
                await page.locator("input[name='otc']").first.wait_for(state="hidden", timeout=12000)
            except Exception:
                pass
            continue

        # --- сопоставление чисел в Authenticator ---
        number = await _text(page, "#idRichContext_DisplaySign")
        if number and not told_number:
            told_number = True
            await say(f"📱 Открой Authenticator и выбери число <b>{number}</b>")
            continue

        # --- выбор способа подтверждения: предпочитаем код ---
        if await _visible(page, "[data-value='PhoneAppOTP']"):
            await _click(page, "[data-value='PhoneAppOTP']")
            continue

        # --- push «Approve sign in» без числа ---
        if await _visible(page, "#idDiv_SAOTCAS_Title") and not told_push:
            told_push = True
            await say("📱 Подтверди вход в приложении Authenticator (или /screen, если что-то не так).")
            continue

        # --- пароль ---
        if await _visible(page, "input[name='passwd']"):
            if not password:
                password = await ask("🔑 Пароль от универского аккаунта (сообщение сразу удалю):", secret=True, timeout=240)
                if not password:
                    await say("Пароль не пришёл — вход отменён.")
                    return False
            await page.locator("input[name='passwd']").first.fill(password)
            await _click(page, "#idSIButton9")
            await asyncio.sleep(2)
            continue

        # --- email ---
        if await _visible(page, "input[name='loginfmt']"):
            await page.locator("input[name='loginfmt']").first.fill(email)
            await _click(page, "#idSIButton9")
            await asyncio.sleep(2)
            continue

        # --- выбор аккаунта ---
        tile = page.get_by_text(email, exact=False)
        try:
            if await tile.count() and await tile.first.is_visible():
                await tile.first.click(timeout=5000)
                continue
        except Exception:
            pass

    shot = await page.screenshot()
    await say("⏱ Вход не завершился за отведённое время. Вот что на экране:", photo=shot)
    return False