python-aternos/python_aternos/atconnect.py

import re
import random
import logging
import lxml.html
from requests import Response
from cloudscraper import CloudScraper
from typing import Optional, Union

from . import atjsparse
from .aterrors import CredentialsError

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'

class AternosConnect:

	def __init__(self) -> None:

		self.session = CloudScraper()
		self.atsession = ''

	def parse_token(self) -> str:

		loginpage = self.request_cloudflare(
			f'https://aternos.org/go/', 'GET'
		).content
		pagetree = lxml.html.fromstring(loginpage)

		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]

			ctx = atjsparse.exec(token_func)
			self.token = ctx.window['AJAX_TOKEN']

		except (IndexError, TypeError):
			raise CredentialsError(
				'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:

		# a list with randlen+1 empty strings:
		# generate a string with spaces,
		# then split it by space
		rand_arr = (' ' * (randlen+1)).split(' ')

		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:

		if isinstance(num, str):
			num = int(num, frombase)

		if isinstance(num, float):
			sliced = str(num)[2:]
			num = int(sliced)

		symbols = '0123456789abcdefghijklmnopqrstuvwxyz'
		basesym = symbols[:base]
		result = ''
		while num > 0:
			rem = num % base
			result = str(basesym[rem]) + result
			num //= base
		return result

	def request_cloudflare(
		self, url:str, method:str,
		params:Optional[dict]=None, data:Optional[dict]=None,
		headers:Optional[dict]=None, reqcookies:Optional[dict]=None,
		sendtoken:bool=False, redirect:bool=True) -> Response:

		try:
			self.atsession = self.session.cookies['ATERNOS_SESSION']
		except KeyError:
			pass

		if method not in ('GET', 'POST'):
			raise NotImplementedError('Only GET and POST are available')

		params = params or {}
		data = data or {}
		headers = headers or {}
		reqcookies = reqcookies or {}
		headers['User-Agent'] = REQUA

		if sendtoken:
			params['TOKEN'] = self.token
			params['SEC'] = self.sec

		# requests.cookies.CookieConflictError bugfix
		reqcookies['ATERNOS_SESSION'] = self.atsession
		del self.session.cookies['ATERNOS_SESSION']

		logging.debug(f'Requesting({method})' + url)
		logging.debug('headers=' + str(headers))
		logging.debug('params=' + str(params))
		logging.debug('data=' + str(data))
		logging.debug('req-cookies=' + str(reqcookies))
		logging.debug('session-cookies=' + str(self.session.cookies))
		
		if method == 'POST':
			req = self.session.post(
				url, data=data, params=params,
				headers=headers, cookies=reqcookies,
				allow_redirects=redirect
			)
		else:
			req = self.session.get(
				url, params={**params, **data},
				headers=headers, cookies=reqcookies,
				allow_redirects=redirect
			)
		
		logging.info(
			f'{method} completed with {req.status_code} status'
		)

		return req