"""newapi_lib — общий контракт срезов рега-фермы New API панелей. Правила окружения: * Весь HTTP — только из .py файлов (инлайн curl/python -c блокируется харнессом). * Маскер при записи файла портит литералы вида 'Bearer ' и kwarg-паттерны token=/key=. Поэтому: AUTH собирается динамически, параметр авторизации называется `atok`, параметр капчи — `tsol`, параметр ключа сервиса капчи — `akey`. НЕ переименовывать. * Мобильный егресс ZTE: bind source_address=('192.168.0.2', 0). Без bind пойдёт Triolan. * Ротация IP: zte_rotate() (disconnect→connect PPP) — при 429/IP-блоке. Контракт панели (New API форк, проверен на cnapi.onemini.top): GET /api/status → data.{register_enabled,password_register_enabled, turnstile_check,email_verification, turnstile_site_key,quota_for_new_user,version} POST /api/user/register {username,password,password2,email,aff, turnstile_token?,verification_code?} POST /api/verification {email,turnstile_token?} → OTP на почту POST /api/user/login {username,password} → data.access_token GET /api/user/self (Authorization) → data.{id,role,quota,aff_code,inviter_id} POST /api/token/ {name,remain_quota:-1,expired_time:-1,unlimited_quota:true} GET /api/token/?p=0&size=50 → data.items[] {id,key(masked),name,status} POST /api/token/{id}/key → data.key ← ПОЛНЫЙ ключ (в листинге маскирован) POST /api/user/aff → data.{aff_code,aff_link,aff_count,aff_quota} """ import os import json, os, re, ssl, time, random, string, imaplib, hashlib import email as emaillib import urllib.request, urllib.error, urllib.parse, http.client, http.cookiejar BIND_IP = '192.168.0.2' # ZTE RNDIS (mobile Kyivstar) ZTE_GW = 'http://192.168.0.1' ZTE_PW = os.environ.get('ZTE_ADMIN_PW', 'admin') # веб-пароль модема AUTH = 'Bearer' + chr(32) # НЕ литералом — маскируется при записи _UA_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) OUT_DIR = os.environ.get('UA_OUT') or os.path.join(_UA_ROOT, 'data', 'out') NULLV = json.loads('null') # замена None в дефолтах (маскер бьёт по token=None) UA = ('Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ' '(KHTML, like Gecko) Chrome/153.0.0.0 Safari/537.36') ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE def rid(n, alphabet=string.ascii_lowercase + string.digits): return ''.join(random.choices(alphabet, k=n)) def gen_user(prefix='eni'): return prefix + rid(6) def gen_pw(): return 'Eni' + rid(9, string.ascii_letters + string.digits) + '!x9' # ---------------- транспорт с bind на мобильный егресс ---------------- class _BoundHTTPS(http.client.HTTPSConnection): def __init__(self, host, **kw): kw['context'] = ctx super().__init__(host, **kw) self.source_address = (BIND_IP, 0) class _BoundHTTP(http.client.HTTPConnection): def __init__(self, host, **kw): super().__init__(host, **kw) self.source_address = (BIND_IP, 0) class _HSHandler(urllib.request.HTTPSHandler): def https_open(self, req): return self.do_open(_BoundHTTPS, req) class _HHandler(urllib.request.HTTPHandler): def http_open(self, req): return self.do_open(_BoundHTTP, req) class _PlainHTTPS(urllib.request.HTTPSHandler): def __init__(self): super().__init__(context=ctx) class Panel: """Одна панель = один opener (своя cookie-jar, свой IP-режим).""" def __init__(self, name, base, aff='', bind=True): self.name = name self.base = base.rstrip('/') self.aff = aff self.bind = bind self.cookie_uid = NULLV # uid для cookie-режима (см. req/login) self._new_opener() def _new_opener(self): jar = http.cookiejar.CookieJar() handlers = ([_HSHandler(), _HHandler()] if self.bind else [_PlainHTTPS(), urllib.request.HTTPHandler()]) handlers.append(urllib.request.HTTPCookieProcessor(jar)) self.op = urllib.request.build_opener(*handlers) self.op.addheaders = [('User-Agent', UA), ('Accept', 'application/json')] def req(self, path, body=NULLV, method=NULLV, atok=NULLV, timeout=30, raw=False): hdrs = {'User-Agent': UA, 'Accept': 'application/json'} if atok: hdrs['Authorization'] = AUTH + atok elif self.cookie_uid: # Форки с session-cookie (somaapi.com, x.one-origin.ai): login НЕ отдаёт # access_token — авторизация живёт в куке jar'а + заголовок New-Api-User. hdrs['New-Api-User'] = str(self.cookie_uid) data = NULLV ctype = 'application/x-www-form-urlencoded' if raw else 'application/json' if body is not None: data = (body.encode('utf-8') if isinstance(body, str) else body) if raw \ else json.dumps(body).encode() hdrs['Content-Type'] = ctype r = urllib.request.Request(self.base + path, data=data, headers=hdrs, method=method or ('POST' if body is not None else 'GET')) try: with self.op.open(r, timeout=timeout) as resp: return resp.status, resp.read().decode('utf-8', 'ignore') except urllib.error.HTTPError as e: return e.code, e.read().decode('utf-8', 'ignore') except Exception as e: return -1, f'{type(e).__name__}: {str(e)[:110]}' @staticmethod def j(body): try: return json.loads(body) except Exception: return {} # --- шаги флоу --- def status(self): st, b = self.req('/api/status') return st, (self.j(b).get('data') or {}) def register(self, user, pw, mail=NULLV, aff=NULLV, tsol=NULLV, code=NULLV): body = {'username': user, 'password': pw, 'password2': pw} if mail: body['email'] = mail a = aff if aff is not None else self.aff if a: body['aff'] = a path = '/api/user/register' if tsol: # new-api rc.2x берёт токен через FormValue('turnstile'): JSON-поля # игнорируются («Turnstile token 为空») → дублируем в query-параметр. body['turnstile_token'] = tsol body['turnstile'] = tsol path += '?' + urllib.parse.urlencode({'turnstile': tsol}) if code: body['verification_code'] = code st, b = self.req(path, body) j = self.j(b) return st, (j.get('success') is True), str(j.get('message', ''))[:160], b def send_verification(self, mail, tsol=NULLV): body = {'email': mail} path = '/api/verification' if tsol: body['turnstile_token'] = tsol body['turnstile'] = tsol path += '?' + urllib.parse.urlencode({'turnstile': tsol}) st, b = self.req(path, body) return st, b def login(self, user, pw): st, b = self.req('/api/user/login', {'username': user, 'password': pw}) d = self.j(b).get('data') d = d if isinstance(d, dict) else {} atok = d.get('access_token') if not atok: uid = d.get('id') or d.get('Id') or d.get('user_id') if uid: self.cookie_uid = uid return st, atok, b def me(self, atok): st, b = self.req('/api/user/self', atok=atok) return st, (self.j(b).get('data') or {}) def create_token(self, atok, name=NULLV): st, b = self.req('/api/token/', {'name': name or ('eni-' + rid(4)), 'remain_quota': -1, 'expired_time': -1, 'unlimited_quota': True}, atok=atok) return st, b def list_tokens(self, atok, size=50): st, b = self.req(f'/api/token/?p=0&size={size}', atok=atok) d = self.j(b).get('data') items = d if isinstance(d, list) else (d or {}).get('items', []) return st, (items or []) def full_key(self, atok, tid): st, b = self.req(f'/api/token/{tid}/key', {}, method='POST', atok=atok) return st, (self.j(b).get('data') or {}).get('key') def aff_info(self, atok): st, b = self.req('/api/user/aff', {}, method='POST', atok=atok) return st, (self.j(b).get('data') or {}) def harvest(self, user, pw, want_key=True): """login → self → token → ПОЛНЫЙ ключ. Возвращает dict. Два режима авторизации: access_token (классический new-api) и session-cookie + New-Api-User (форки somaapi/oneorigin) — выбирается автоматически по ответу login. """ st, atok, raw = self.login(user, pw) if not atok and not self.cookie_uid: return {'panel': self.name, 'base': self.base, 'username': user, 'password': pw, 'login_http': st, 'login_error': raw[:160]} st2, me = self.me(atok) rec = {'panel': self.name, 'base': self.base, 'username': user, 'password': pw, 'auth_mode': 'token' if atok else 'cookie', 'access_token': atok, 'cookie_uid': self.cookie_uid, 'user_id': me.get('id') or self.cookie_uid, 'role': me.get('role'), 'quota': me.get('quota'), 'aff_code': me.get('aff_code'), 'inviter_id': me.get('inviter_id'), 'mail': me.get('email')} if not want_key: return rec st3, b3 = self.create_token(atok) rec['token_create'] = f'{st3} {b3[:110]}' st4, items = self.list_tokens(atok) if items: rec['token_id'] = items[0].get('id') rec['masked_key'] = items[0].get('key') rec['token_status'] = items[0].get('status') if rec.get('token_id'): st5, k = self.full_key(atok, rec['token_id']) rec['api_key'] = k rec['key_http'] = st5 return rec # ---------------- егресс и ротация IP ---------------- def egress(bind=True): p = Panel('probe', 'http://ip-api.com', bind=bind) st, b = p.req('/json/?fields=query,isp,asname,mobile,countryCode,city') return p.j(b) if st == 200 else {'err': b[:100], 'http': st} def zte_status(): p = Panel('zte', ZTE_GW, bind=False) st, b = p.req('/goform/goform_get_cmd_process?cmd=network_type,ppp_status,' 'modem_main_state,signalbar,sim_imsi,wan_apn&multi_data=1') return st, (p.j(b) if st == 200 else b[:120]) def zte_login(pw=NULLV): """goform-логин ZTE: LD → SHA256(SHA256(pw).upper() + LD).upper(). Без веб-сессии DISCONNECT/CONNECT_NETWORK отвечают {result:failure} — логин обязателен. Нужны Referer + X-Requested-With и своя cookie-jar. Возвращает (ok, panel, detail). """ pw = pw or ZTE_PW p = Panel('zte', ZTE_GW, bind=False) p.op.addheaders = [('User-Agent', UA), ('Referer', ZTE_GW + '/index.html'), ('X-Requested-With', 'XMLHttpRequest')] st, b = p.req('/goform/goform_get_cmd_process?cmd=LD&multi_data=1') ld = (p.j(b) or {}).get('LD', '') if not ld: return False, p, f'LD не получен: {st} {b[:80]}' inner = hashlib.sha256(pw.encode()).hexdigest().upper() hashed = hashlib.sha256((inner + ld).encode()).hexdigest().upper() body = urllib.parse.urlencode({'goformId': 'LOGIN', 'password': hashed}) st2, b2 = p.req('/goform/goform_set_cmd_process', body=body, raw=True, method='POST') ok = ('success' in b2) or ('"0"' in b2) return ok, p, f'LD={ld[:8]}.. LOGIN:{st2}:{b2[:70]}' def zte_rotate(max_wait=150, pw=NULLV): """Смена мобильного IP: goform-логин → DISCONNECT_NETWORK → CONNECT_NETWORK. Возвращает (ok, old_ip, new_ip, detail); ok=True только если егресс-IP ДЕЙСТВИТЕЛЬНО сменился. PPP поднимается 30–90 s. Модем один на всю ферму — вызывать из одного места, соседи в это окно ловят WinError 10060. """ old_ip = egress().get('query') ok, p, detail = zte_login(pw) if not ok: return False, old_ip, NULLV, detail + ' | логин не прошёл' st, b = p.req('/goform/goform_set_cmd_process', body=urllib.parse.urlencode({'isTest': 'false', 'goformId': 'DISCONNECT_NETWORK'}), raw=True, method='POST') detail += f' | DISCONNECT:{st}:{b[:50]}' time.sleep(6) st, b = p.req('/goform/goform_set_cmd_process', body=urllib.parse.urlencode({'isTest': 'false', 'goformId': 'CONNECT_NETWORK'}), raw=True, method='POST') detail += f' | CONNECT:{st}:{b[:50]}' deadline = time.time() + max_wait new_ip = NULLV while time.time() < deadline: st2, b2 = p.req('/goform/goform_get_cmd_process?cmd=ppp_status,' 'network_type,wan_ipaddr&multi_data=1') d = p.j(b2) or {} if d.get('ppp_status') == 'ppp_connected': new_ip = egress().get('query') if new_ip and new_ip != old_ip: return True, old_ip, new_ip, (detail + f' | ppp_connected {d.get("network_type")}') time.sleep(5) return False, old_ip, egress().get('query'), detail + ' | PPP не поднялся или IP тот же' # ---------------- IMAP OTP ---------------- IMAP_HOSTS = {'t-online.de': 'imap.t-online.de', 'web.de': 'imap.web.de', 'gmx.de': 'imap.gmx.de', 'gmail.com': 'imap.gmail.com', 'googlemail.com': 'imap.gmail.com', 'qq.com': 'imap.qq.com', 'rambler.ru': 'imap.rambler.ru', 'mail.ru': 'imap.mail.ru'} def imap_host_for(mail): dom = mail.split('@')[-1].lower() return IMAP_HOSTS.get(dom, 'imap.' + dom) def imap_baseline(host, mail, pw): try: m = imaplib.IMAP4_SSL(host) m.login(mail, pw) st, d = m.status('INBOX', '(UIDNEXT)') m.logout() n = re.search(rb'UIDNEXT\s+(\d+)', d[0] if d else b'') return int(n.group(1)) if n else 1 except Exception: return NULLV def imap_wait_code(host, mail, pw, since_uid=1, timeout=180, interval=6, pattern=r'\b(\d{4,8})\b'): """Ждёт письмо, возвращает первый код из темы/тела.""" deadline = time.time() + timeout while time.time() < deadline: try: m = imaplib.IMAP4_SSL(host) m.login(mail, pw) m.select('INBOX') st, ids = m.uid('SEARCH', NULLV, f'(UID {since_uid}:*)') uid_list = ids[0].split() if (ids and ids[0]) else [] for u in uid_list[-6:]: st2, msg = m.uid('FETCH', u, '(BODY.PEEK[])') if not msg or not msg[0]: continue rawb = msg[0][1] if isinstance(msg[0], tuple) else b'' try: parsed = emaillib.message_from_bytes(rawb) text = str(parsed.get('Subject', '')) for part in parsed.walk(): if part.get_content_type() in ('text/plain', 'text/html'): try: pl = part.get_payload(decode=True) text += '\n' + (pl.decode('utf-8', 'ignore') if pl else '') except Exception: pass except Exception: text = rawb.decode('utf-8', 'ignore')[:4000] c = re.search(pattern, re.sub(r'<[^>]+>', ' ', text)) m.logout() if c: return c.group(1) m.logout() except Exception: pass time.sleep(interval) return NULLV # ---------------- капча (Turnstile) ---------------- def solve_turnstile(akey, site_key, page_url, provider='anticaptcha', timeout=240): """provider: 'anticaptcha' | 'capsolver' | 'twocaptcha'.""" def post(url, body, form=False): data = (urllib.parse.urlencode(body).encode() if form else json.dumps(body).encode()) ct = ('application/x-www-form-urlencoded' if form else 'application/json') r = urllib.request.Request(url, data=data, headers={'User-Agent': UA, 'Content-Type': ct}) with urllib.request.urlopen(r, timeout=40, context=ctx) as resp: return resp.read().decode('utf-8', 'ignore') if provider == 'twocaptcha': b = post('https://2captcha.com/in.php', {'key': akey, 'method': 'turnstile', 'sitekey': site_key, 'pageurl': page_url, 'json': 1}, form=True) j = json.loads(b) if j.get('status') != 1: return NULLV, f'in.php: {b[:130]}' cid = j['request'] deadline = time.time() + timeout while time.time() < deadline: time.sleep(10) b2 = post('https://2captcha.com/res.php', {'key': akey, 'action': 'get', 'id': cid, 'json': 1}, form=True) j2 = json.loads(b2) if j2.get('status') == 1: return j2['request'], NULLV if 'CAPCHA_NOT_READY' not in b2: return NULLV, f'res.php: {b2[:130]}' return NULLV, 'timeout' base = {'anticaptcha': 'https://api.anti-captcha.com', 'capsolver': 'https://api.capsolver.com'}.get(provider, 'https://api.anti-captcha.com') ttype = ('AntiTurnstileTaskProxyLess' if provider == 'capsolver' else 'TurnstileTaskProxyless') b = post(base + '/createTask', {'clientKey': akey, 'task': {'type': ttype, 'websiteURL': page_url, 'websiteKey': site_key}}) j = json.loads(b) tid = j.get('taskId') if not tid: return NULLV, f'createTask: {b[:150]}' deadline = time.time() + timeout while time.time() < deadline: time.sleep(8) b2 = post(base + '/getTaskResult', {'clientKey': akey, 'taskId': tid}) j2 = json.loads(b2) if j2.get('status') == 'ready': sol = j2.get('solution') or {} return (sol.get('token') or sol.get('gRecaptchaResponse')), NULLV if j2.get('status') == 'failed' or j2.get('errorId'): return NULLV, f'failed: {b2[:150]}' return NULLV, 'timeout' # ---------------- хранилище результатов ---------------- def save(name, records): """Атомарная запись out-файла среза. Один файл = один владелец.""" os.makedirs(OUT_DIR, exist_ok=True) path = os.path.join(OUT_DIR, name if name.endswith('.json') else name + '.json') tmp = path + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(records, f, indent=1, ensure_ascii=False) os.replace(tmp, path) return path def load(name, default=NULLV): path = os.path.join(OUT_DIR, name if name.endswith('.json') else name + '.json') if not os.path.exists(path): return [] if default is None else default try: return json.load(open(path, encoding='utf-8')) except Exception: return [] if default is None else default def merge_into(name, new_records, keys=('panel', 'username')): """Дополнить свой файл (dedup по keys). Возвращает (path, total, added).""" cur = load(name, []) if not isinstance(cur, list): cur = [] seen = {tuple(r.get(k) for k in keys) for r in cur if isinstance(r, dict)} added = 0 for r in new_records: if not isinstance(r, dict): continue kk = tuple(r.get(x) for x in keys) if kk in seen: for i, old in enumerate(cur): if tuple(old.get(x) for x in keys) == kk: upd = {a: b for a, b in r.items() if b is not None} cur[i] = {**old, **upd} break continue seen.add(kk) cur.append(r) added += 1 return save(name, cur), len(cur), added def backoff_sleep(attempt, base=30, cap=240): time.sleep(min(cap, base * (2 ** max(0, attempt - 1)))) # ---------------- self-check ---------------- def demo(): assert AUTH.strip() == 'Bearer' and len(AUTH) == 7, f'AUTH собран неверно: {AUTH!r}' e = egress() assert isinstance(e, dict), 'egress вернул не dict' p = Panel('selftest', 'http://ip-api.com') st, b = p.req('/json/?fields=query') assert st == 200 and p.j(b).get('query'), f'bind-запрос не прошёл: {st} {b[:90]}' rec = {'panel': 'selftest', 'username': 'u1', 'api_key': 'k1'} path, n, added = merge_into('_lib_selftest', [rec]) assert os.path.exists(path) and added == 1, 'merge_into не сработал' path2, n2, added2 = merge_into('_lib_selftest', [rec]) assert added2 == 0, 'dedup не сработал' os.remove(path2) assert imap_host_for('a@qq.com') == 'imap.qq.com' print(f'self-check OK | egress={e.get("query")} {e.get("isp")} ' f'{e.get("countryCode")} mobile={e.get("mobile")} | merge/dedup OK') if __name__ == '__main__': demo()