From ddcda18ff5e463c5546a72efa47bc275010a6e1d Mon Sep 17 00:00:00 2001 From: Oscar Fernando Davis Flores Date: Sat, 29 Jun 2024 21:06:21 -0700 Subject: [PATCH] I'm stupid --- PKG-INFO | 2 +- pymino/__init__.py | 2 +- pymino/ext/utilities/request_handler.py | 400 ++++++++++++++++++++---- setup.cfg | 2 +- 4 files changed, 336 insertions(+), 70 deletions(-) diff --git a/PKG-INFO b/PKG-INFO index 0a649b5..7618480 100644 --- a/PKG-INFO +++ b/PKG-INFO @@ -1,6 +1,6 @@ Metadata-Version: 2.1 Name: pymino -Version: 1.2.7.0 +Version: 1.2.7.1 Summary: Easily create a bot for Amino Apps using a modern easy to use synchronous library. Home-page: https://github.com/forevercynical/pymino Author: forevercynical diff --git a/pymino/__init__.py b/pymino/__init__.py index 714481d..30dc4f8 100644 --- a/pymino/__init__.py +++ b/pymino/__init__.py @@ -6,7 +6,7 @@ __author__ = 'cynical' __license__ = 'MIT' __copyright__ = 'Copyright 2023 Cynical' -__version__ = '1.2.7.0' +__version__ = '1.2.7.1' __description__ = 'A Python wrapper for the aminoapps.com API' from .bot import Bot diff --git a/pymino/ext/utilities/request_handler.py b/pymino/ext/utilities/request_handler.py index 294a8a0..0f155c6 100644 --- a/pymino/ext/utilities/request_handler.py +++ b/pymino/ext/utilities/request_handler.py @@ -1,88 +1,354 @@ -from hmac import new -from hashlib import sha1 -from base64 import b64encode -from secrets import token_hex -from typing import Union -import requests - -class Generator: +from uuid import uuid4 +from json import loads, dumps +from colorama import Fore, Style +from typing import Optional, Union, Tuple, Callable + +from .generate import Generator +from ..entities.handlers import orjson_exists +from requests import Session as Http, Response as HttpResponse + +from ..entities import ( + Forbidden, + BadGateway, + APIException, + ServiceUnavailable + ) + +from requests.exceptions import ( + ConnectionError, + ReadTimeout, + SSLError, + ProxyError, + ConnectTimeout + ) + +if orjson_exists(): + from orjson import ( + loads as orjson_loads, + dumps as orjson_dumps + ) + +class RequestHandler: + """ + `RequestHandler` - A class that handles all requests + + `**Parameters**`` + - `bot` - The main bot class. + - `generator` - The generator class. + - `proxy` - The proxy to use for requests. + + """ def __init__( self, - prefix: Union[str, int], - device_key: str, - signature_key: str, - service_key: str + bot, + generator: Generator, + proxy: Optional[str] = None ) -> None: - self.PREFIX = bytes.fromhex(str(prefix)) - self.DEVICE_KEY = bytes.fromhex(device_key) - self.SIGNATURE_KEY = bytes.fromhex(signature_key) - self.SERVICE_KEY = service_key + self.bot = bot + self.generate = generator + self.api_url: str = "http://service.aminoapps.com/api/v1" + self.http_handler: Http = Http() + self.sid: Optional[str] = None + self.device: Optional[str] = None + self.userId: Optional[str] = None + self.orjson: bool = orjson_exists() + + self.proxy = { + "http": proxy, + "https": proxy + } if proxy is not None else None + + self.response_map = { + 403: Forbidden, + 502: BadGateway, + 503: ServiceUnavailable + } + + def service_url(self, url: str) -> str: + """ + `service_url` - Appends the endpoint to the service url + + `**Parameters**`` + - `url` - The endpoint to append to the service url. + + `**Returns**`` + - `str` - The service url. + + """ + return f"{self.api_url}{url}" if url.startswith("/") else url + + def service_headers(self) -> dict: + """Returns the service headers""" + return { + "NDCLANG": "en", + "ACCEPT-LANGUAGE": "en-US", + "User-Agent": "Dalvik/2.1.0 (Linux; U; Android 12; com.narvii.amino.master/3.5.35071)", + "HOST": "service.aminoapps.com", + "CONNECTION": "Keep-Alive", + "ACCEPT-ENCODING": "gzip, deflate, br", + "NDCAUTH": f"sid={self.sid}", + "AUID": self.userId or str(uuid4()) + } + + def fetch_request(self, method: str) -> Callable: + """ + `fetch_request` - Returns the request method + + `**Parameters**`` + - `method` - The request method to return. + + `**Returns**`` + - `Callable` - The request method. + + """ + request_methods = { + "GET": self.http_handler.get, + "POST": self.http_handler.post, + "DELETE": self.http_handler.delete, + } + return request_methods[method] + + def send_request( + self, + method: str, + url: str, + data: Union[dict, bytes, None], + headers: dict, + content_type: Optional[str] + ) -> Tuple[int, str]: + """ + `send_request` - Sends a request + + `**Parameters**`` + - `method` - The request method to use. + - `url` - The url to send the request to. + - `data` - The data to send with the request. + - `headers` - The headers to send with the request. + - `content_type` - The content type of the data. + + `**Returns**`` + - `Tuple[int, str]` - The status code and response from the request. + + """ + try: + response: HttpResponse = self.fetch_request(method)( + url, data=data, headers=headers, proxies=self.proxy + ) + return response.status_code, response.text + except ( + ConnectionError, + ReadTimeout, + SSLError, + ProxyError, + ConnectTimeout, + ) as e: + self.bot._log(f"Failed to send request: {e}") + return self.handler(method, url, data, content_type) + + def handler( + self, + method: str, + url: str, + data: Union[dict, bytes, None] = None, + content_type: Optional[str] = None, + is_login_required: bool = True + ) -> dict: + """ + `handler` - Handles all requests + + `**Parameters**`` + - `method` - The request method to use. + - `url` - The url to send the request to. + - `data` - The data to send with the request. + - `content_type` - The content type of the data. + - `is_login_required` - Whether or not the request requires a login. + + `**Returns**`` + - `dict` - The response from the request. + + """ + url = self.service_url(url) + + url, headers, binary_data = self.service_handler(url, {**data, "uid": self.userId} if data is not None else None, content_type) + + if all([method=="POST", data is None]): + headers["CONTENT-TYPE"] = "application/octet-stream" + + if not is_login_required: + headers.pop("NDCAUTH") + headers.pop("AUID") + + try: + status_code, content = self.send_request( + method, url, binary_data, headers, content_type + ) + except TypeError: # NOTE: Not sure if this is even needed. + return self.handler(method, url, data, content_type) - def device_id(self) -> str: + self.print_response(method=method, url=url, status_code=status_code) + + response = self.handle_response(status_code=status_code, response=content) + + if response is None: + return self.handler(method, url, data, content_type) + + return response + + def service_handler( + self, + url: str, + data: Union[dict, bytes, None] = None, + content_type: Optional[str] = None + ) -> Tuple[str, dict, Union[dict, bytes, None]]: """ - `generate_device_id` Generates a device ID based on a specific string. + `service_handler` - Signs the request and returns the service url, headers and data + + `**Parameters**`` + - `url` - The url to send the request to. + - `data` - The data to send with the request. + - `content_type` - The content type of the data. + + `**Returns**`` + - `Tuple[str, dict, Union[dict, bytes, None]]` - The service url, headers and data. - `**Returns**` - - `str` - Returns a device ID as a string. """ - encoded_data = sha1(str(token_hex(20)).encode('utf-8')).hexdigest() + + headers = {"NDCDEVICEID": self.device or self.generate.device_id(), **self.service_headers()} - digest = new( - self.DEVICE_KEY, - self.PREFIX + bytes.fromhex(encoded_data), - sha1).hexdigest() + if data or content_type: + headers, data = self.fetch_signature(data, headers, content_type) - return f"{bytes.hex(self.PREFIX)}{encoded_data}{digest}".upper() + return url, headers, self.ensure_utf8(data) - def signature(self, data: str) -> str: + def ensure_utf8(self, data: Union[dict, bytes, None]) -> Union[dict, bytes, None]: """ - `signature` Generates a signature based on a specific string. + `ensure_utf8` - Ensures the data is utf-8 encoded + + `**Parameters**`` + - `data` - The data to encode. + + `**Returns**`` + - `Union[dict, bytes, None]` - The encoded data. - `**Parameters**` - - `data` - Data to generate a signature from - `**Returns**` - - `str` - Returns a signature as a string. """ - signature = [self.PREFIX[0]] - signature.extend(new( - self.SIGNATURE_KEY, - str(data).encode("utf-8"), sha1).digest()) + if data is None: return data - return b64encode(bytes(signature)).decode("utf-8") - - def update_device(self, device: str) -> str: + def handle_dict(data: dict): + return {key: self.ensure_utf8(value) for key, value in data.items()} + + def handle_str(data: str): + return data.encode("utf-8") + + handlers = { + dict: handle_dict, + str: handle_str + } + + return handlers.get(type(data), lambda x: x)(data) + + def fetch_signature( + self, + data: Union[dict, bytes, None], + headers: dict, + content_type: str = None + ) -> Tuple[dict, Union[dict, bytes, None]]: """ - Update a device ID to new prefix. + `fetch_signature` - Fetches the signature and returns the data and updated headers + + `**Parameters**`` + - `data` - The data to send with the request. + - `headers` - The headers to send with the request. + - `content_type` - The content type of the data. + + `**Returns**`` + - `Tuple[dict, Union[dict, bytes, None]]` - The headers and data. - :param device: The device ID to update. - :type device: str - :return: The updated device ID as a string. - :rtype: str """ - encoded_data = sha1(str(bytes.fromhex(device[2:42])).encode('utf-8')).hexdigest() - digest = new( - self.DEVICE_KEY, - self.PREFIX + bytes.fromhex(encoded_data), - sha1).hexdigest() + if not isinstance(data, bytes): + data = orjson_dumps(data).decode("utf-8") if self.orjson else dumps(data) + + headers.update({ + "CONTENT-LENGTH": f"{len(data)}", + "CONTENT-TYPE": content_type or "application/json; charset=utf-8", + "NDC-MSG-SIG": ( + self.generate.signature(data) + ), + "NDC-MESSAGE-SIGNATURE": ( + self.generate.sign_data(data, self.userId, self.sid, self.device) + ) + }) + return headers, data - return f"{bytes.hex(self.PREFIX)}{encoded_data}{digest}".upper() - - def sign_data(self, data: str, auid: str, sid: str, deviceid: str) -> str: - if any(not i for i in (data, auid, sid, deviceid)): - return None - response = requests.post( - "https://friendify.ninja/api/v1/g/s/security/public_key", - headers={ - "SID": sid, - "NDCDEVICEID": deviceid, - "AUID": auid, - "key": self.SERVICE_KEY - }, - data=str(data).encode("utf-8") - ) + def raise_error(self, response: dict) -> None: + """ + `raise_error` - Raises an error if an error is in the response + + `**Parameters**`` + - `response` - The response from the request. + + `**Returns**`` + - `None` - Raises an error if the status code is in the response map. + - `404` - Returns 404 if the status code is 105 and the email and password is set. + + """ + if all( + [ + response.get("api:statuscode", 200) == 105, + hasattr(self, "email"), + hasattr(self, "password"), + ] + ): + self.bot.run(self.email, self.password, use_cache=False) + return 404 + + self.bot._log(f"Exception: {response}") + raise APIException(response) + + def handle_response(self, status_code: int, response: str) -> dict: + """ + `handle_response` - Handles the response and returns the response as a dict + + `**Parameters**`` + - `status_code` - The status code of the response. + - `response` - The response to handle. - if response.status_code != 200: - raise Exception(response.text) - return response.text \ No newline at end of file + `**Returns**`` + - `dict` - The response as a dict. + + """ + if status_code in self.response_map: + raise self.response_map[status_code] + + try: + response = orjson_loads(response) if self.orjson else loads(response) + except Exception: + response = loads(response) + + if status_code != 200: + check_response = self.raise_error(response) + if check_response == 404: + return None + + return response + + def print_response(self, method: str, url: str, status_code: int): + """ + `print_response` - Prints the response if debug is enabled + + `**Parameters**`` + - `method` - The request method used. + - `url` - The url the request was sent to. + - `status_code` - The status code of the response. + - `response` - The response to print. + + """ + if self.bot.debug: + color = Fore.RED if status_code != 200 else { + "GET": Fore.BLUE, + "POST": Fore.GREEN, + "DELETE": Fore.MAGENTA, + "LITE": Fore.YELLOW + }.get(method, Fore.RED) + print(f"{color}{Style.BRIGHT}{method}{Style.RESET_ALL} - {url}") \ No newline at end of file diff --git a/setup.cfg b/setup.cfg index 94fe646..928c16e 100644 --- a/setup.cfg +++ b/setup.cfg @@ -1,6 +1,6 @@ [metadata] name = pymino -version = 1.2.7.0 +version = 1.2.7.1 author = forevercynical author_email = me@cynical.gg description = Easily create a bot for Amino Apps using a modern easy to use synchronous library.