This repository has been archived on 2024-07-30. You can view files and clone it, but cannot push or open issues or pull requests.
python-aternos/python_aternos/atconnect.py

152 lines
3.5 KiB
Python
Raw Normal View History

2021-10-08 18:35:20 +03:00
import re
import time
import random
import lxml.html
from requests import Response
2021-10-08 18:35:20 +03:00
from cloudscraper import CloudScraper
from typing import Optional, Union
2021-10-08 18:35:20 +03:00
from . import aterrors
2021-11-01 17:04:19 +03:00
from . import atjsparse
2021-10-27 19:42:41 +03:00
REQGET = 0
REQPOST = 1
REQUA = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:68.0) Gecko/20100101 Goanna/4.8 Firefox/68.0 PaleMoon/29.4.0.2'
2021-10-08 18:35:20 +03:00
class AternosConnect:
2021-10-08 18:35:20 +03:00
2021-11-01 17:04:19 +03:00
def __init__(self) -> None:
pass
def parse_token(self, response:Optional[Union[str,bytes]]=None) -> str:
if response == None:
loginpage = self.request_cloudflare(
f'https://aternos.org/go/', REQGET
).content
pagetree = lxml.html.fromstring(loginpage)
else:
pagetree = lxml.html.fromstring(response)
try:
pagehead = pagetree.head
text = pagehead.text_content()
js_code = re.findall(r'\(\(\)(.*?)\)\(\);', text)
token_func = js_code[1] if len(js_code) > 1 else js_code[0]
print('*** Function:', token_func)
2021-11-01 17:04:19 +03:00
ctx = atjsparse.exec(token_func)
self.token = ctx.window['AJAX_TOKEN']
except (IndexError, TypeError):
raise aterrors.AternosCredentialsError(
'Unable to parse TOKEN from the page'
)
return self.token
def generate_sec(self) -> str:
randkey = self.generate_aternos_rand()
randval = self.generate_aternos_rand()
self.sec = f'{randkey}:{randval}'
self.session.cookies.set(
f'ATERNOS_SEC_{randkey}', randval,
domain='aternos.org'
)
return self.sec
def generate_aternos_rand(self, randlen:int=16) -> str:
rand_arr = []
for i in range(randlen+1):
rand_arr.append('')
rand = random.random()
rand_alphanum = self.convert_num(rand, 36) + ('0' * 17)
return (rand_alphanum[:18].join(rand_arr)[:randlen])
def convert_num(
self, num:Union[int,float,str],
base:int, frombase:int=10) -> str:
2021-11-01 17:04:19 +03:00
if isinstance(num, str):
num = int(num, frombase)
2021-11-01 17:04:19 +03:00
if isinstance(num, float):
sliced = str(num)[2:]
num = int(sliced)
symbols = '0123456789abcdefghijklmnopqrstuvwxyz'
basesym = symbols[:base]
2021-11-01 17:04:19 +03:00
result = ''
while num > 0:
rem = num % base
result = str(basesym[rem]) + result
2021-11-01 17:04:19 +03:00
num //= base
return result
def request_cloudflare(
self, url:str, method:int, retries:int=10,
params:Optional[dict]=None, data:Optional[dict]=None,
headers:Optional[dict]=None, reqcookies:Optional[dict]=None,
sendtoken:bool=False, redirect:bool=True) -> Response:
2021-11-01 17:04:19 +03:00
cftitle = '<title>Please Wait... | Cloudflare</title>'
if params == None:
params = {}
2021-11-01 17:04:19 +03:00
if headers == None:
headers = {}
headers['User-Agent'] = REQUA
if sendtoken:
url += f'?TOKEN={self.token}&SEC={self.sec}'
2021-11-01 17:04:19 +03:00
try:
cookies = self.session.cookies
except AttributeError:
cookies = None
countdown = retries
while True:
2021-11-01 17:04:19 +03:00
self.session = CloudScraper()
if cookies != None:
self.session.cookies = cookies
2021-11-01 17:04:19 +03:00
if reqcookies != None:
for cookiekey in reqcookies:
self.session.cookies.set(cookiekey, reqcookies[cookiekey])
time.sleep(1)
2021-11-01 17:04:19 +03:00
if method == REQPOST:
req = self.session.post(
url, data=data, params=params,
headers=headers, cookies=reqcookies,
allow_redirects=redirect
2021-11-01 17:04:19 +03:00
)
else:
req = self.session.get(
url, params=params,
headers=headers, cookies=reqcookies,
allow_redirects=redirect
)
if not cftitle in req.text:
break
if not countdown > 0:
raise aterrors.CloudflareError(
'The retries limit has been reached'
2021-11-01 17:04:19 +03:00
)
countdown -= 1
2021-11-01 17:04:19 +03:00
return req