Пишем свой авторег для Telegram

import Java from 'frida-java-bridge';

Java.perform(() => {
  const CM = Java.use('org.telegram.tgnet.ConnectionsManager');
  CM.native_init.implementation = function () {
    send({ type: 'device_token', token: arguments[11] });
    return this.native_init.apply(this, arguments);
  };
});

import asyncio
import os
import re
from datetime import datetime, timezone

import frida
from opentele2.api import API
from opentele2.tl import TelegramClient
from telethon import errors, functions
from telethon.tl import types as t

PKG = "org.telegram.messenger.web"
SITEKEY = "6LflQ8EqAAAAAE3JaczP-gBVVObsFsSe2U7yZJ6O"
HERE = os.path.dirname(os.path.abspath(__file__))
SESSIONS_DIR = os.path.join(HERE, "sessions")
MAX_BATCH = 4

def get_numbers() -> list[str]:
    numbers: list[str] = []
    print(f"\nВводи номера по одному (максимум {MAX_BATCH}, пустая строка — старт):")
    while len(numbers) < MAX_BATCH:
        raw = input(f"  номер #{len(numbers) + 1} (с +): ").strip()
        if not raw:
            if numbers:
                break
            continue
        digits = re.sub(r"\D", "", raw)
        if len(digits) >= 8:
            numbers.append("+" + digits)
        else:
            print("  некорректный номер — ещё раз.")
    return numbers

def session_path(phone: str) -> str:
    os.makedirs(SESSIONS_DIR, exist_ok=True)
    return os.path.join(SESSIONS_DIR, "reg_" + re.sub(r"\D", "", phone))

def build_client(phone: str, device_token: str) -> TelegramClient:
    api = API.TelegramAndroid.copy()
    api.api_id = 4
    api.api_hash = "014b35b6184100b085b0d0572f9b5103"
    api.device_model = "samsungSM-S928B"
    api.system_version = "SDK 35"
    api.app_version = "12.8.3 (69229)"
    api.lang_pack = "android"
    api.lang_code = "en"
    api.system_lang_code = "en-us"
    client = TelegramClient(session_path(phone), api=api)
    offset = datetime.now(timezone.utc).astimezone().utcoffset()
    tz = int(offset.total_seconds()) if offset else 0
    client._init_request.params = t.JsonObject(
        value=[
            t.JsonObjectValue("device_token", t.JsonString(device_token)),
            t.JsonObjectValue(
                "data",
                t.JsonString(
                    "49C1522548EBACD46CE322B6FD47F6092BB745D0F88082145CAF35E14DCC38E1"
                ),
            ),
            t.JsonObjectValue("installer", t.JsonString("com.android.vending")),
            t.JsonObjectValue("package_id", t.JsonString(PKG)),
            t.JsonObjectValue("tz_offset", t.JsonNumber(float(tz))),
            t.JsonObjectValue("perf_cat", t.JsonNumber(3.0)),
        ]
    )
    return client

class AsyncOracle:
    def __init__(self) -> None:
        self.loop = asyncio.get_running_loop()
        agent = frida.Compiler().build("agent.js", project_root=HERE)
        device = frida.get_usb_device(timeout=8)
        app = next(
            (a for a in device.enumerate_applications() if a.identifier == PKG), None
        )
        if not app or not app.pid:
            raise RuntimeError("Открой Telegram на устройстве")
        self.session = device.attach(app.pid)
        self.futures: dict[int, asyncio.Future] = {}
        self.ready = asyncio.Event()
        self.script = self.session.create_script(agent)
        self.script.on("message", self._on_message)
        self.script.load()
        exports = getattr(self.script, "exports_sync", None) or self.script.exports
        self.push_token = os.getenv("DEVICE_TOKEN") or (exports.pushtoken() or "")
        if self.push_token:
            src = "env" if os.getenv("DEVICE_TOKEN") else "эмулятор"
            print(f"  device_token ({src}): {self.push_token[:32]}…")
        else:
            print("  [!] device_token пуст (Firebase не готов?) — доставка уйдёт в App")

    def _on_message(self, message, _data) -> None:
        if message.get("type") != "send":
            return
        payload = message["payload"]
        kind = payload.get("type")
        if kind == "ready":
            self.loop.call_soon_threadsafe(self.ready.set)
        elif kind == "token":
            fut = self.futures.get(payload["account"])
            if fut and not fut.done():
                self.loop.call_soon_threadsafe(fut.set_result, payload["token"])

    async def mint(self, slot: int) -> str:
        await self.ready.wait()
        fut: asyncio.Future = self.loop.create_future()
        self.futures[slot] = fut
        exports = getattr(self.script, "exports_sync", None) or self.script.exports
        exports.mint(slot, slot, SITEKEY)
        try:
            return await asyncio.wait_for(fut, timeout=30)
        except asyncio.TimeoutError:
            return ""

    def close(self) -> None:
        try:
            self.session.detach()
        except Exception:
            pass

def show_target(r) -> None:
    if type(r).__name__ == "SentCodePaymentRequired":
        print(
            f"  ответ: платная верификация (store_product={getattr(r, 'store_product', None)})"
        )
        return
    ty = r.type
    nxt = type(r.next_type).__name__ if getattr(r, "next_type", None) else None
    print(f"  код ушёл: {type(ty).__name__}")
    if getattr(ty, "email_pattern", None):
        print(f"    email           = {ty.email_pattern}")
    print(f"    длина кода      = {getattr(ty, 'length', None)}")
    print(f"    next_type       = {nxt}")
    print(f"    timeout         = {getattr(r, 'timeout', None)}")
    print(f"    phone_code_hash = {r.phone_code_hash}")

async def arm(oracle: AsyncOracle, slot: int, phone: str):
    client = build_client(phone, oracle.push_token)
    try:
        await client.connect()
        if await client.is_user_authorized():
            me = await client.get_me()
            print(f"  {phone}: уже авторизованы ({me.id})")
            return {"client": client, "phone": phone, "authorized": True}

        token = await oracle.mint(slot)
        if not token:
            print(f"  {phone}: токен не пойман — пропуск")
            await client.disconnect()
            return None

        def send_code():
            return functions.auth.SendCodeRequest(
                phone,
                client.api_id,
                client.api_hash,
                t.CodeSettings(allow_firebase=True, allow_app_hash=True),
            )

        try:
            await client(send_code())
        except errors.RPCError:
            pass
        r = await client(
            functions.InvokeWithReCaptchaRequest(token=token, query=send_code())
        )

        if type(r).__name__ == "SentCodePaymentRequired":
            print(f"  {phone}: платная верификация — пропуск")
            await client.disconnect()
            return None
        print(f"  {phone}: {type(r.type).__name__}")
        return {"client": client, "phone": phone, "r": r, "pch": r.phone_code_hash}
    except errors.FloodWaitError as exc:
        print(f"  {phone}: FloodWait {exc.seconds}s — пропуск")
    except errors.RPCError as exc:
        print(f"  {phone}: {type(exc).__name__}: {str(exc)[:60]} — пропуск")
    except Exception as exc:
        print(f"  {phone}: {type(exc).__name__}: {exc} — пропуск")
    try:
        await client.disconnect()
    except Exception:
        pass
    return None

async def setup_email(client: TelegramClient, phone: str, pch: str):
    purpose = t.EmailVerifyPurposeLoginSetup(phone_number=phone, phone_code_hash=pch)
    email = input("  Требуется email. Введи свой email: ").strip()
    try:
        sent = await client(
            functions.account.SendVerifyEmailCodeRequest(purpose=purpose, email=email)
        )
        print(f"  код отправлен на {getattr(sent, 'email_pattern', email)}")
    except errors.RPCError as exc:
        print(f"  email не принят: {type(exc).__name__}")
        return None, None
    for _ in range(3):
        code = input("  Код из письма: ").strip()
        try:
            verified = await client(
                functions.account.VerifyEmailRequest(
                    purpose=purpose, verification=t.EmailVerificationCode(code=code)
                )
            )
        except errors.RPCError as exc:
            print(f"  код неверный ({type(exc).__name__}), ещё раз.")
            continue
        r = verified.sent_code
        if type(r).__name__ == "SentCodePaymentRequired":
            print("  после email — платная верификация, пропуск.")
            return None, None
        show_target(r)
        return r, r.phone_code_hash
    return None, None

async def finish_login(client: TelegramClient, phone: str, pch: str) -> bool:
    for _ in range(3):
        code = input("  Код с номера: ").strip()
        try:
            user = await client.sign_in(phone=phone, code=code, phone_code_hash=pch)
            print(f"  вошли (аккаунт уже был): {user.id}")
            return True
        except errors.PhoneCodeInvalidError:
            print("  неверный код, ещё раз.")
        except errors.PhoneCodeExpiredError:
            print("  код истёк — пропуск номера.")
            return False
        except errors.SessionPasswordNeededError:
            user = await client.sign_in(password=input("  2FA пароль: ").strip())
            print(f"  вошли с 2FA: {user.id}")
            return True
        except errors.PhoneNumberUnoccupiedError:
            first = input("  Номер свободен -> имя: ").strip() or "Test"
            res = await client(
                functions.auth.SignUpRequest(
                    phone_number=phone,
                    phone_code_hash=pch,
                    first_name=first,
                    last_name="",
                )
            )
            try:
                await client._on_login(res.user)
            except Exception:
                pass
            print(f"  зарегистрирован новый аккаунт: {res.user.id}")
            return True
    return False

async def complete(armed: dict) -> bool:
    client, phone = armed["client"], armed["phone"]
    try:
        if armed.get("authorized"):
            return True
        r, pch = armed["r"], armed["pch"]
        print(f"\n=== {phone} ===")
        show_target(r)
        kind = type(r.type).__name__

        if kind == "SentCodeTypeSetUpEmailRequired":
            r, pch = await setup_email(client, phone, pch)
            if r is None:
                return False
        elif kind == "SentCodeTypeApp":
            if not getattr(r, "next_type", None):
                print(
                    "  только App без SMS-фолбэка (нет живого device_token?) — пропуск."
                )
                return False
            try:
                r = await client(functions.auth.ResendCodeRequest(phone, pch))
                pch = r.phone_code_hash
                show_target(r)
            except errors.RPCError as exc:
                print(f"  resend не удался: {type(exc).__name__} — пропуск.")
                return False

        return await finish_login(client, phone, pch)
    except errors.RPCError as exc:
        print(f"  ошибка Telegram: {type(exc).__name__}: {str(exc)[:70]}")
        return False
    finally:
        await client.disconnect()

async def main() -> None:
    print("=== Регистрация Telegram через reCAPTCHA-оракул ===")
    while True:
        phones = get_numbers()
        if not phones:
            print("Выход.")
            return

        try:
            oracle = AsyncOracle()
        except Exception as exc:
            print(f"Оракул недоступен: {exc}")
            print(
                "Проверь: устройство подключено, frida-server запущен, Telegram открыт."
            )
            return

        try:
            print(f"\nминтим {len(phones)} токен(ов) параллельно + шлём капчу...")
            armed = await asyncio.gather(
                *(arm(oracle, i, ph) for i, ph in enumerate(phones))
            )
        finally:
            oracle.close()

        ok = []
        for item in armed:
            if item and await complete(item):
                ok.append(item["phone"])

        if ok:
            print(f"\nГотово. Успешно: {', '.join(ok)}")
            print(f"Сессии в: {SESSIONS_DIR}")
            return
        print("\n— ни один номер не прошёл, введи другие —")

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nВыход.")

import Java from 'frida-java-bridge';

function installHook() {
  const CM = Java.use('org.telegram.tgnet.ConnectionsManager');
  CM.native_receivedCaptchaResult.implementation = function (currentAccount, requestTokens, token) {
    send({ type: 'token', account: currentAccount, token: token });
  };
}

rpc.exports = {
  mint(currentAccount, requestToken, key_id) {
    Java.scheduleOnMainThread(() => {
      try {
        Java.use('org.telegram.messenger.CaptchaController')
          .request(currentAccount, requestToken, 'signup', key_id);
      } catch (e) {
        send({ type: 'error', error: 'request: ' + e });
      }
    });
  },
  pushtoken() {
    let out = '';
    Java.perform(() => {
      try {
        const SC = Java.use('org.telegram.messenger.SharedConfig');
        const ps = SC.pushString.value;
        const pss = SC.pushStringStatus.value;
        out = (ps && ps.length) ? ps : (pss || '');
      } catch (e) {
        send({ type: 'error', error: 'pushtoken: ' + e });
      }
    });
    return out;
  }
};

Java.perform(() => {
  installHook();
  send({ type: 'ready' });
});

import asyncio
import os
import random
import re
import string
import time
from datetime import datetime, timezone

from dotenv import load_dotenv

from hero_sms import Activation, HeroSMS, NoBalanceError, NoNumbersError
from mail_tm import Account, MailTimeoutError, MailTM

load_dotenv()
API_KEY = os.getenv("HEROSMS_API_KEY", "")
SERVICE = os.getenv("HEROSMS_SERVICE", "tg")
COUNTRY = int(os.getenv("HEROSMS_COUNTRY", "187"))

async def get_numbers(
    client: HeroSMS, service: str = SERVICE, country: int = COUNTRY, **kwargs
) -> Activation:
    return await client.get_number(service, country, **kwargs)

async def get_sms(
    client: HeroSMS,
    activation: Activation,
    *,
    timeout: float = 300.0,
    poll_interval: float = 5.0,
) -> str:
    return await client.get_sms(
        activation, timeout=timeout, poll_interval=poll_interval
    )

async def create_email(mail: MailTM) -> Account:
    return await mail.create_account()

async def get_email_code(
    mail: MailTM,
    account: Account,
    *,
    timeout: float = 300.0,
    poll_interval: float = 5.0,
    from_contains: str | None = None,
) -> str:
    return await mail.get_code(
        account.token,
        timeout=timeout,
        poll_interval=poll_interval,
        from_contains=from_contains,
        address=account.address,
    )

os.environ.setdefault("OPENTELE_NO_FETCH", "1")
from opentele2.api import API  # noqa: E402
from opentele2.tl import TelegramClient  # noqa: E402
from telethon import errors, functions  # noqa: E402
from telethon.tl import types as t  # noqa: E402

PARALLEL = 1
DEFAULT_NAME = "Kitsune"
API_ID = 4
API_HASH = "014b35b6184100b085b0d0572f9b5103"
DEVICE_MODEL = "samsungSM-S928B"
SYSTEM_VERSION = "SDK 35"
APP_VERSION = "12.8.3 (69229)"
LANG_PACK = "android"
LANG_CODE = "en"
SYSTEM_LANG_CODE = "en-us"
INSTALLER = "com.android.vending"
PERF_CAT = 3
CERT_FINGERPRINT = "49C1522548EBACD46CE322B6FD47F6092BB745D0F88082145CAF35E14DCC38E1"
PACKAGE_ID = "org.telegram.messenger.web"
SESSIONS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "sessions")

def plausible_recaptcha_token() -> str:
    alphabet = string.ascii_letters + string.digits + "-_"
    body = "".join(random.choice(alphabet) for _ in range(random.randint(3500, 4600)))
    return "0cAFcWeA" + body

def device_token() -> str:
    return f"__FIREBASE_GENERATING_SINCE_{int(time.time())}__"

def build_client(phone: str):
    digits = re.sub(r"\D", "", phone)
    os.makedirs(SESSIONS_DIR, exist_ok=True)
    session_path = os.path.join(SESSIONS_DIR, f"reg_{digits}")
    api = API.TelegramAndroid.copy()
    api.api_id = API_ID
    api.api_hash = API_HASH
    api.device_model = DEVICE_MODEL
    api.system_version = SYSTEM_VERSION
    api.app_version = APP_VERSION
    api.lang_pack = LANG_PACK
    api.lang_code = LANG_CODE
    api.system_lang_code = SYSTEM_LANG_CODE
    client = TelegramClient(session_path, api=api)
    offset = datetime.now(timezone.utc).astimezone().utcoffset()
    tz = int(offset.total_seconds()) if offset else 0
    client._init_request.params = t.JsonObject(
        value=[
            t.JsonObjectValue("device_token", t.JsonString(device_token())),
            t.JsonObjectValue("data", t.JsonString(CERT_FINGERPRINT)),
            t.JsonObjectValue("installer", t.JsonString(INSTALLER)),
            t.JsonObjectValue("package_id", t.JsonString(PACKAGE_ID)),
            t.JsonObjectValue("tz_offset", t.JsonNumber(float(tz))),
            t.JsonObjectValue("perf_cat", t.JsonNumber(float(PERF_CAT))),
        ]
    )
    return (client, session_path)

def code_settings():
    # allow_firebase=True желательно ВСЕГДА.
    return t.CodeSettings(allow_firebase=True)

def type_name(r) -> str:
    return (
        type(r).__name__
        if type(r).__name__ == "SentCodePaymentRequired"
        else type(r.type).__name__
    )

async def register(hero: HeroSMS, act: Activation) -> str:
    phone = f"+{act.phone}"

    def p(msg: str) -> None:
        print(f"[{phone}] {msg}")

    client, session_path = build_client(phone)
    await client.connect()
    try:
        if await client.is_user_authorized():
            me = await client.get_me()
            p(f"уже авторизованы этой сессией: id={me.id}")
            return "ok"

        def send_code():
            return functions.auth.SendCodeRequest(
                phone, client.api_id, client.api_hash, code_settings()
            )

        await client(functions.help.GetConfigRequest())
        try:
            r = await client(
                functions.InvokeWithReCaptchaRequest(
                    token=plausible_recaptcha_token(), query=send_code()
                )
            )
        except errors.FloodWaitError as e:
            p(f"FLOOD_WAIT {e.seconds}s -> невалиден")
            return "invalid"
        except errors.RPCError as e:
            p(f"reCAPTCHA/sendCode отклонён: {type(e).__name__} -> невалиден")
            return "invalid"
        if type_name(r) == "SentCodePaymentRequired":
            p("платная верификация -> невалиден")
            return "invalid"
        pch = r.phone_code_hash

        async def complete_login(code: str, pch: str) -> str:
            try:
                user = await client.sign_in(phone=phone, code=code, phone_code_hash=pch)
                p(f"вошли (аккаунт уже был): id={user.id}")
                return "ok"
            except errors.PhoneNumberUnoccupiedError:
                res = await client(
                    functions.auth.SignUpRequest(
                        phone_number=phone,
                        phone_code_hash=pch,
                        first_name=DEFAULT_NAME,
                        last_name="",
                    )
                )
                user = res.user
                try:
                    await client._on_login(user)
                except Exception:
                    pass
                p(f"ЗАРЕГИСТРИРОВАН новый аккаунт: id={user.id}")
                try:
                    cur = await client.GetCurrentSession()
                    if cur:
                        p(
                            f"сессия: {cur.app_name} {cur.app_version} | "
                            f"api_id={cur.api_id} | official={cur.official_app}"
                        )
                except Exception:
                    pass
                p(f"сессия сохранена: {session_path}.session")
                return "ok"
            except errors.SessionPasswordNeededError:
                p("аккаунт под 2FA (пароля нет) -> невалиден")
                return "invalid"
            except errors.RPCError as e:
                p(f"sign_in/signUp: {type(e).__name__} -> невалиден")
                return "invalid"

        if type_name(r) == "SentCodeTypeSetUpEmailRequired":
            p("нужен login-email — завожу одноразовую почту")
            async with MailTM() as mail:
                acc = await create_email(mail)
                p(f"почта: {acc.address}")
                purpose = t.EmailVerifyPurposeLoginSetup(
                    phone_number=phone, phone_code_hash=pch
                )
                try:
                    await client(
                        functions.account.SendVerifyEmailCodeRequest(
                            purpose=purpose, email=acc.address
                        )
                    )
                except errors.RPCError as e:
                    p(f"sendVerifyEmailCode: {type(e).__name__} -> невалиден")
                    return "invalid"
                try:
                    msg = await mail.wait_for_message(
                        acc.token,
                        match=lambda m: m.code is not None,
                        timeout=180,
                        address=acc.address,
                    )
                except MailTimeoutError:
                    p("код подтверждения почты не пришёл -> невалиден")
                    return "invalid"
                p(f"подтверждаю почту кодом {msg.code}")
                try:
                    verified = await client(
                        functions.account.VerifyEmailRequest(
                            purpose=purpose,
                            verification=t.EmailVerificationCode(code=msg.code),
                        )
                    )
                except errors.RPCError as e:
                    p(f"verifyEmail: {type(e).__name__} -> невалиден")
                    return "invalid"
            sent = getattr(verified, "sent_code", None)
            if sent is None or type_name(sent) == "SentCodePaymentRequired":
                p("логин-код бесплатно недоступен (payment/none) -> невалиден")
                return "invalid"
            r, pch = (sent, sent.phone_code_hash)
            p(f"почта подтверждена -> логин-код придёт как {type_name(r)}")
        p(f"капча пройдена: {type_name(r)}")
        if type_name(r) == "SentCodeTypeApp" and getattr(r, "next_type", None):
            r = await client(functions.auth.ResendCodeRequest(phone, pch))
            if type_name(r) == "SentCodePaymentRequired":
                p("после resend платная -> невалиден")
                return "invalid"
            pch = r.phone_code_hash
            p(f"resend -> {type_name(r)}")
        if type_name(r) != "SentCodeTypeSms":
            p(f"не SMS ({type_name(r)}) -> невалиден")
            return "invalid"
        code = await get_sms(hero, act)
        if not code:
            p("SMS-код не получен -> невалиден")
            return "invalid"
        return await complete_login(code, pch)
    finally:
        await client.disconnect()

async def worker(wid: int, hero: HeroSMS) -> None:
    while True:
        try:
            act = await get_numbers(hero)
        except NoNumbersError:
            print(f"[worker {wid}] свободных номеров нет — стоп.")
            return
        except NoBalanceError:
            print(f"[worker {wid}] недостаточно баланса — стоп.")
            return
        num = f"+{act.phone}"
        outcome = await register(hero, act)
        if outcome == "ok":
            print(f"[worker {wid}] [OK] успех на {num}")
            return
        if outcome == "email":
            print(f"[worker {wid}] {num}: login-email не удался — беру другой номер.")
            continue
        print(f"[worker {wid}] {num}: невалиден — беру другой номер.")

async def entrypoint() -> None:
    print("=== Регистрация Telegram (Android) БЕЗ эмулятора ===")
    if not API_KEY:
        print("Не задан HEROSMS_API_KEY (см. .env / .env.example)")
        return
    async with HeroSMS(API_KEY) as hero:
        await asyncio.gather(*(worker(i + 1, hero) for i in range(PARALLEL)))

if __name__ == "__main__":
    try:
        asyncio.run(entrypoint())
    except KeyboardInterrupt:
        print("\nпрервано.")

import aiohttp, asyncio, time, json

class ApiError(Exception):
    pass

class NoNumbersError(ApiError):
    pass

class NoBalanceError(ApiError):
    pass

_E = {"NO_NUMBERS": NoNumbersError, "NO_BALANCE": NoBalanceError, "NO_MONEY": NoBalanceError}

class Activation:
    def __init__(self, id, phone):
        self.id = id
        self.phone = phone

class HeroSMS:
    def __init__(self, api_key, base_url="https://hero-sms.com/stubs/handler_api.php"):
        self.k = api_key
        self.u = base_url
        self.s = None

    async def __aenter__(self):
        self.s = aiohttp.ClientSession()
        return self

    async def __aexit__(self, *e):
        await self.s.close()

    async def _r(self, action, **p):
        q = {"api_key": self.k, "action": action}
        q.update({k: v for k, v in p.items() if v is not None})
        async with self.s.get(self.u, params=q) as r:
            return self._p(r.status, await r.text())

    def _p(self, st, t):
        t = t.strip()
        try:
            d = json.loads(t)
        except Exception:
            d = None
        if isinstance(d, dict):
            c = str(d.get("title") or "").upper()
            if st >= 400 or c in _E:
                raise _E.get(c, ApiError)(c or "ERR")
            return d
        return t

    async def get_number(self, service, country=0, **kw):
        d = await self._r("getNumberV2", service=service, country=country, **kw)
        return Activation(
            str(d.get("activationId") or d.get("id")),
            str(d.get("phoneNumber") or d.get("number") or "").lstrip("+"),
        )

    async def get_sms(self, a, timeout=300, poll_interval=5):
        i = a.id if isinstance(a, Activation) else a
        e = time.monotonic() + timeout
        while 1:
            t = await self._r("getStatus", id=i)
            if isinstance(t, str) and t.startswith("STATUS_OK"):
                return t.split(":", 1)[1]
            if isinstance(t, dict):
                c = t.get("code") or t.get("smsCode")
                if c:
                    return str(c)
            if time.monotonic() >= e:
                return ""
            await asyncio.sleep(poll_interval)

import aiohttp, asyncio, re, secrets, string, time, json

_RE = re.compile(r"(?<!\d)(\d{5,8})(?!\d)")

class MailTMError(Exception):
    pass

class MailTimeoutError(MailTMError):
    pass

class Account:
    def __init__(self, address, password, id, token):
        self.address = address
        self.password = password
        self.id = id
        self.token = token

class MailMessage:
    def __init__(self, subject, intro, text, html):
        self.subject = subject
        self.intro = intro
        self.text = text
        self.html = html

    @property
    def code(self):
        for c in (self.subject, self.intro, self.text, *self.html):
            m = _RE.search(c or "")
            if m:
                return m.group(1)
        return None

def _rand(n):
    return "".join(secrets.choice(string.ascii_lowercase + string.digits) for _ in range(n))

class MailTM:
    def __init__(self, base_url="https://api.mail.tm"):
        self.u = base_url
        self.s = None

    async def __aenter__(self):
        self.s = aiohttp.ClientSession()
        return self

    async def __aexit__(self, *e):
        await self.s.close()

    async def _r(self, method, path, token=None, body=None):
        h = {"Accept": "application/json"}
        if token:
            h["Authorization"] = "Bearer " + token
        async with self.s.request(method, self.u + path, json=body, headers=h) as r:
            t = await r.text()
            if r.status >= 400:
                raise MailTMError(f"[{r.status}] {t[:80]}")
            try:
                return json.loads(t)
            except Exception:
                return None

    async def create_account(self):
        doms = await self._r("GET", "/domains")
        dom = (doms if isinstance(doms, list) else doms["hydra:member"])[0]["domain"]
        a = _rand(14) + "@" + dom
        pw = _rand(16)
        acc = await self._r("POST", "/accounts", body={"address": a, "password": pw})
        tok = (await self._r("POST", "/token", body={"address": a, "password": pw}))["token"]
        return Account(a, pw, str(acc.get("id", "")), tok)

    async def _msgs(self, token):
        d = await self._r("GET", "/messages", token=token)
        return d if isinstance(d, list) else d.get("hydra:member", [])

    async def wait_for_message(self, token, match=None, timeout=300, poll_interval=5, address=""):
        e = time.monotonic() + timeout
        while 1:
            for s in await self._msgs(token):
                d = await self._r("GET", "/messages/" + s["id"], token=token)
                h = d.get("html") or []
                m = MailMessage(
                    str(d.get("subject", "")),
                    str(d.get("intro", "")),
                    str(d.get("text", "") or ""),
                    h if isinstance(h, list) else [h],
                )
                if match and not match(m):
                    continue
                return m
            if time.monotonic() >= e:
                raise MailTimeoutError(address)
            await asyncio.sleep(poll_interval)

    async def get_code(self, token, timeout=300, poll_interval=5, from_contains=None, address=""):
        m = await self.wait_for_message(
            token,
            match=lambda x: x.code is not None,
            timeout=timeout,
            poll_interval=poll_interval,
            address=address,
        )
        return m.code

HEROSMS_API_KEY=5d35cb58d2A70e10bd14746693f595b0
HEROSMS_SERVICE=tg
HEROSMS_COUNTRY=187 # Это USA.

import os
import threading

import frida

HERE = os.path.dirname(os.path.abspath(__file__))
APP = "Telegram"  # ph.telegra.Telegraph

def get_ios(timeout=15):
    try:
        return frida.get_usb_device(timeout)
    except Exception as e:
        raise RuntimeError(
            "iOS-устройство по USB не найдено "
            "(usbmux / Apple Devices запущены? кабель на месте?)"
        ) from e

def get_apns() -> str:
    agent = frida.Compiler().build("agent.js", project_root=HERE)
    dev = get_ios()
    session = dev.attach(APP)

    holder = {}
    got = threading.Event()

    def on_message(m, _data):
        if m.get("type") == "send":
            p = m["payload"]
            tp = p.get("type")
            if tp == "token" and p.get("hex"):
                holder["hex"] = p["hex"]
                got.set()
            elif tp == "hooked":
                print(
                    "  [hooked]",
                    p.get("cls"),
                    "-> форсирую registerForRemoteNotifications",
                )
            elif tp == "nosel":
                print("  [nosel] делегат без селектора:", p.get("cls"))
            elif tp == "env":
                print("  [env]", p.get("sel"), "=", p.get("val"))
            elif tp == "ready":
                print("  [ready] жду APNS-токен ...")
        elif m.get("type") == "error":
            print("  [frida]", m.get("description"))

    script = session.create_script(agent)
    script.on("message", on_message)
    script.load()
    got.wait(timeout=30)
    try:
        session.detach()
    except Exception:
        pass
    return holder.get("hex", "")

if __name__ == "__main__":
    hexv = get_apns()
    if not hexv:
        print(
            "\nТокен не пойман. Открой Telegram на айфоне (attach нужен запущенный "
            "процесс), проверь frida-server и usbmux, и повтори."
        )
    else:
        print("\nAPNS :", hexv)

import ObjC from 'frida-objc-bridge';

let done = false;

function hexData(d) {
  try { const n = d.length(); const b = d.bytes(); let h = ''; for (let i = 0; i < n; i++) h += ('0' + b.add(i).readU8().toString(16)).slice(-2); return h; } catch (e) { return null; }
}

const iv = setInterval(() => {
  if (done) return;
  try {
    const app = ObjC.classes.UIApplication;
    if (!app) return;
    const shared = app.sharedApplication();
    if (!shared || shared.isNull()) return;
    const del = shared.delegate();
    if (!del || del.isNull()) return;
    const cls = del.$class;
    const sel = '- application:didRegisterForRemoteNotificationsWithDeviceToken:';
    const m = cls[sel];
    if (m) {
      Interceptor.attach(m.implementation, {
        onEnter(a) { const hx = hexData(new ObjC.Object(a[3])); send({ type: 'token', src: 'didRegister', hex: hx }); }
      });
      done = true;
      send({ type: 'hooked', cls: del.$className });
      try { ObjC.schedule(ObjC.mainQueue, () => { shared.registerForRemoteNotifications(); }); } catch (e) {}
    } else {
      send({ type: 'nosel', cls: del.$className });
      done = true;
    }
  } catch (e) {}
}, 100);

setTimeout(() => {
  try {
    const ctxs = ObjC.chooseSync(ObjC.classes.MTApiEnvironment);
    for (const e of ctxs) {
      for (const sel of ['deviceToken', 'apnsToken', 'token']) {
        try { if (e.respondsToSelector_(ObjC.selector(sel))) { const v = e[sel](); send({ type: 'env', sel, val: v ? v.toString() : null }); } } catch (er) {}
      }
    }
  } catch (e) {}
}, 4000);

send({ type: 'ready' });

import asyncio
import base64
import os
import random
import re
import string
from datetime import datetime, timezone

from dotenv import load_dotenv

from hero_sms import Activation, HeroSMS, NoBalanceError, NoNumbersError
from mail_tm import Account, MailTimeoutError, MailTM

load_dotenv()
API_KEY = os.getenv("HEROSMS_API_KEY", "")
SERVICE = os.getenv("HEROSMS_SERVICE", "tg")
COUNTRY = int(os.getenv("HEROSMS_COUNTRY", "187"))

async def get_numbers(
    client: HeroSMS, service: str = SERVICE, country: int = COUNTRY, **kwargs
) -> Activation:
    return await client.get_number(service, country, **kwargs)

async def get_sms(
    client: HeroSMS,
    activation: Activation,
    *,
    timeout: float = 300.0,
    poll_interval: float = 5.0,
) -> str:
    return await client.get_sms(
        activation, timeout=timeout, poll_interval=poll_interval
    )

async def create_email(mail: MailTM) -> Account:
    return await mail.create_account()

async def get_email_code(
    mail: MailTM,
    account: Account,
    *,
    timeout: float = 300.0,
    poll_interval: float = 5.0,
    from_contains: str | None = None,
) -> str:
    return await mail.get_code(
        account.token,
        timeout=timeout,
        poll_interval=poll_interval,
        from_contains=from_contains,
        address=account.address,
    )

os.environ.setdefault("OPENTELE_NO_FETCH", "1")
from opentele2.api import API  # noqa: E402
from opentele2.tl import TelegramClient  # noqa: E402
from telethon import errors, functions  # noqa: E402
from telethon.tl import types as t  # noqa: E402

PARALLEL = 1
DEFAULT_NAME = "Kitsune"
API_ID = 8
API_HASH = "7245de8e747a0d6fbe11f7cc14fcc0bb"
DEVICE_MODEL = "iPhone X"
SYSTEM_VERSION = "16.7.16"
APP_VERSION = "12.8.1"
LANG_PACK = "ios"
LANG_CODE = "en"
SYSTEM_LANG_CODE = "en-US"
BUNDLE_ID = "ph.telegra.Telegraph"
DEVICE_TOKEN_ENVIRONMENT = "production"
DEVICE_TOKEN_TYPE = "apns"
# Реальный APNS-токен. Мусор или его отсутствие = невалид.
APNS = os.environ.get(
    "APNS", "e247cca7e83f45d19c512f3ccbe026efd33c7dfe7d0a968c1a2ac2a7c18bee7e"
)
APNS_B64 = base64.b64encode(bytes.fromhex(APNS)).decode()
SESSIONS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "sessions")

def plausible_recaptcha_token() -> str:
    alphabet = string.ascii_letters + string.digits + "-_"
    body = "".join(random.choice(alphabet) for _ in range(random.randint(3500, 4600)))
    return "0cAFcWeA5lXxhXRt" + body

def build_client(phone: str):
    digits = re.sub(r"\D", "", phone)
    os.makedirs(SESSIONS_DIR, exist_ok=True)
    session_path = os.path.join(SESSIONS_DIR, f"reg_{digits}")
    api = API.TelegramIOS.copy()
    api.api_id = API_ID
    api.api_hash = API_HASH
    api.device_model = DEVICE_MODEL
    api.system_version = SYSTEM_VERSION
    api.app_version = APP_VERSION
    api.lang_pack = LANG_PACK
    api.lang_code = LANG_CODE
    api.system_lang_code = SYSTEM_LANG_CODE
    client = TelegramClient(session_path, api=api)
    offset = datetime.now(timezone.utc).astimezone().utcoffset()
    tz = int(offset.total_seconds()) if offset else 0
    client._init_request.params = t.JsonObject(
        value=[
            t.JsonObjectValue("device_token", t.JsonString(APNS_B64)),
            t.JsonObjectValue(
                "device_token_environment", t.JsonString(DEVICE_TOKEN_ENVIRONMENT)
            ),
            t.JsonObjectValue("tz_offset", t.JsonNumber(float(tz))),
            t.JsonObjectValue("device_token_type", t.JsonString(DEVICE_TOKEN_TYPE)),
            t.JsonObjectValue("bundleId", t.JsonString(BUNDLE_ID)),
        ]
    )
    return (client, session_path)

def code_settings():
    # ВАЖНО: для iOS новые поля
    return t.CodeSettings(
        allow_missed_call=True,
        logout_tokens=[],
        allow_firebase=True,
        token=APNS,
        app_sandbox=False,
    )

def type_name(r) -> str:
    return (
        type(r).__name__
        if type(r).__name__ == "SentCodePaymentRequired"
        else type(r.type).__name__
    )

async def register(hero: HeroSMS, act: Activation) -> str:
    phone = f"+{act.phone}"

    def p(msg: str) -> None:
        print(f"[{phone}] {msg}")

    client, session_path = build_client(phone)
    await client.connect()
    try:
        if await client.is_user_authorized():
            me = await client.get_me()
            p(f"уже авторизованы этой сессией: id={me.id}")
            return "ok"

        def send_code():
            return functions.auth.SendCodeRequest(
                phone, client.api_id, client.api_hash, code_settings()
            )

        await client(functions.help.GetConfigRequest())
        try:
            r = await client(
                functions.InvokeWithReCaptchaRequest(
                    token=plausible_recaptcha_token(), query=send_code()
                )
            )
        except errors.FloodWaitError as e:
            p(f"FLOOD_WAIT {e.seconds}s -> невалиден")
            return "invalid"
        except errors.RPCError as e:
            p(f"reCAPTCHA/sendCode отклонён: {type(e).__name__} -> невалиден")
            return "invalid"
        if type_name(r) == "SentCodePaymentRequired":
            p("платная верификация -> невалиден")
            return "invalid"
        pch = r.phone_code_hash

        async def complete_login(code: str, pch: str) -> str:
            try:
                user = await client.sign_in(phone=phone, code=code, phone_code_hash=pch)
                p(f"вошли (аккаунт уже был): id={user.id}")
                return "ok"
            except errors.PhoneNumberUnoccupiedError:
                res = await client(
                    functions.auth.SignUpRequest(
                        phone_number=phone,
                        phone_code_hash=pch,
                        first_name=DEFAULT_NAME,
                        last_name="",
                    )
                )
                user = res.user
                try:
                    await client._on_login(user)
                except Exception:
                    pass
                p(f"ЗАРЕГИСТРИРОВАН новый аккаунт: id={user.id}")
                try:
                    cur = await client.GetCurrentSession()
                    if cur:
                        p(
                            f"сессия: {cur.app_name} {cur.app_version} | "
                            f"api_id={cur.api_id} | official={cur.official_app}"
                        )
                except Exception:
                    pass
                p(f"сессия сохранена: {session_path}.session")
                return "ok"
            except errors.SessionPasswordNeededError:
                p("аккаунт под 2FA (пароля нет) -> невалиден")
                return "invalid"
            except errors.RPCError as e:
                p(f"sign_in/signUp: {type(e).__name__} -> невалиден")
                return "invalid"

        if type_name(r) == "SentCodeTypeSetUpEmailRequired":
            p("нужен login-email — завожу одноразовую почту")
            async with MailTM() as mail:
                acc = await create_email(mail)
                p(f"почта: {acc.address}")
                purpose = t.EmailVerifyPurposeLoginSetup(
                    phone_number=phone, phone_code_hash=pch
                )
                try:
                    await client(
                        functions.account.SendVerifyEmailCodeRequest(
                            purpose=purpose, email=acc.address
                        )
                    )
                except errors.RPCError as e:
                    p(f"sendVerifyEmailCode: {type(e).__name__} -> невалиден")
                    return "invalid"
                try:
                    msg = await mail.wait_for_message(
                        acc.token,
                        match=lambda m: m.code is not None,
                        timeout=180,
                        address=acc.address,
                    )
                except MailTimeoutError:
                    p("код подтверждения почты не пришёл -> невалиден")
                    return "invalid"
                p(f"подтверждаю почту кодом {msg.code}")
                try:
                    verified = await client(
                        functions.account.VerifyEmailRequest(
                            purpose=purpose,
                            verification=t.EmailVerificationCode(code=msg.code),
                        )
                    )
                except errors.RPCError as e:
                    p(f"verifyEmail: {type(e).__name__} -> невалиден")
                    return "invalid"
            sent = getattr(verified, "sent_code", None)
            if sent is None or type_name(sent) == "SentCodePaymentRequired":
                p("логин-код бесплатно недоступен (payment/none) -> невалиден")
                return "invalid"
            r, pch = (sent, sent.phone_code_hash)
            p(f"почта подтверждена -> логин-код придёт как {type_name(r)}")
        p(f"капча пройдена: {type_name(r)}")
        if type_name(r) == "SentCodeTypeApp" and getattr(r, "next_type", None):
            r = await client(functions.auth.ResendCodeRequest(phone, pch))
            if type_name(r) == "SentCodePaymentRequired":
                p("после resend платная -> невалиден")
                return "invalid"
            pch = r.phone_code_hash
            p(f"resend -> {type_name(r)}")
        if type_name(r) != "SentCodeTypeSms":
            p(f"не SMS ({type_name(r)}) -> невалиден")
            return "invalid"
        code = await get_sms(hero, act)
        if not code:
            p("SMS-код не получен -> невалиден")
            return "invalid"
        return await complete_login(code, pch)
    finally:
        await client.disconnect()

async def worker(wid: int, hero: HeroSMS) -> None:
    while True:
        try:
            act = await get_numbers(hero)
        except NoNumbersError:
            print(f"[worker {wid}] свободных номеров нет — стоп.")
            return
        except NoBalanceError:
            print(f"[worker {wid}] недостаточно баланса — стоп.")
            return
        num = f"+{act.phone}"
        outcome = await register(hero, act)
        if outcome == "ok":
            print(f"[worker {wid}] [OK] успех на {num}")
            return
        if outcome == "email":
            print(f"[worker {wid}] {num}: login-email не удался — беру другой номер.")
            continue
        print(f"[worker {wid}] {num}: невалиден — беру другой номер.")

async def entrypoint() -> None:
    print("=== Регистрация Telegram (iOS) БЕЗ эмулятора ===")
    if not API_KEY:
        print("Не задан HEROSMS_API_KEY (см. .env / .env.example)")
        return
    async with HeroSMS(API_KEY) as hero:
        await asyncio.gather(*(worker(i + 1, hero) for i in range(PARALLEL)))

if __name__ == "__main__":
    try:
        asyncio.run(entrypoint())
    except KeyboardInterrupt:
        print("\nпрервано.")