From 6dbed22e49119b1427a0a50152b0c33b0a7acab5 Mon Sep 17 00:00:00 2001 From: Mark Veidemanis Date: Tue, 12 Apr 2022 22:05:46 +0100 Subject: [PATCH] Implement LBTC authentication --- handler/lib/agoradesk_py.py | 16 ++- handler/lib/localbitcoins_py.py | 174 ++++++++++++++------------------ 2 files changed, 86 insertions(+), 104 deletions(-) diff --git a/handler/lib/agoradesk_py.py b/handler/lib/agoradesk_py.py index 8683d56..af565bd 100644 --- a/handler/lib/agoradesk_py.py +++ b/handler/lib/agoradesk_py.py @@ -20,7 +20,10 @@ __copyright__ = "(C) 2021 https://codeberg.org/MarvinsCryptoTools/agoradesk_py" __version__ = "0.1.0" # set logging -logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", +) logging.getLogger("requests.packages.urllib3").setLevel(logging.INFO) logging.getLogger("urllib3.connectionpool").setLevel(logging.INFO) logger = util.get_logger(__name__) @@ -618,7 +621,10 @@ class AgoraDesk: return self._api_call( api_method="equation", http_method="POST", - query_values={"price_equation": price_equation, "currency": currency}, + query_values={ + "price_equation": price_equation, + "currency": currency, + }, ) # Public ad search related API Methods @@ -1019,4 +1025,8 @@ class AgoraDesk: if otp: params["otp"] = otp - return self._api_call(api_method="wallet-send/XMR", http_method="POST", query_values=params) + return self._api_call( + api_method="wallet-send/XMR", + http_method="POST", + query_values=params, + ) diff --git a/handler/lib/localbitcoins_py.py b/handler/lib/localbitcoins_py.py index 3fe936b..ffabb91 100644 --- a/handler/lib/localbitcoins_py.py +++ b/handler/lib/localbitcoins_py.py @@ -29,101 +29,16 @@ __copyright__ = "(C) 2021 https://codeberg.org/MarvinsCryptoTools/agoradesk_py" __version__ = "0.1.0" # set logging -logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", +) logging.getLogger("requests.packages.urllib3").setLevel(logging.INFO) logging.getLogger("urllib3.connectionpool").setLevel(logging.INFO) logger = util.get_logger(__name__) URI_API = "https://localbitcoins.com/api/" - - -def hmac(hmac_key, hmac_secret, server="https://localbitcoins.com"): - conn = Connection() - conn._set_hmac(server, hmac_key, hmac_secret) - return conn - - -class Connection: - def __init__(self): - self.server = None - - # HMAC stuff - self.hmac_key = None - self.hmac_secret = None - - def call(self, method, url, params=None, stream=False, files=None): - method = method.upper() - if method not in ["GET", "POST"]: - raise Exception("Invalid method {}!".format(method)) - - if method == "GET" and files: - raise Exception("You cannot send files with GET method!") - - if files and not isinstance(files, dict): - raise Exception('"files" must be a dict of file objects or file contents!') - - # If URL is absolute, then convert it - if url.startswith(self.server): - url = url[len(self.server) :] # noqa - - if self.hmac_key: - - # If nonce fails, retry several times, then give up - for retry in range(10): - - nonce = str(int(time.time() * 1000)).encode("ascii") - - # Prepare request based on method. - if method == "POST": - api_request = requests.Request("POST", self.server + url, data=params, files=files).prepare() - params_encoded = api_request.body - - # GET method - else: - api_request = requests.Request("GET", self.server + url, params=params).prepare() - params_encoded = urlparse(api_request.url).query - - # Calculate signature - message = nonce + self.hmac_key + url.encode("ascii") - if params_encoded: - if sys.version_info >= (3, 0) and isinstance(params_encoded, str): - message += params_encoded.encode("ascii") - else: - message += params_encoded - signature = hmac_lib.new(self.hmac_secret, msg=message, digestmod=hashlib.sha256).hexdigest().upper() - - # Store signature and other stuff to headers - api_request.headers["Apiauth-Key"] = self.hmac_key - api_request.headers["Apiauth-Nonce"] = nonce - api_request.headers["Apiauth-Signature"] = signature - - # Send request - session = requests.Session() - response = session.send(api_request, stream=stream) - - # If HMAC Nonce is already used, then wait a little and try again - try: - response_json = response.json() - if int(response_json.get("error", {}).get("error_code")) == 42: - time.sleep(0.1) - continue - except: # noqa - # No JSONic response, or interrupt, better just give up - pass - - return response - - raise Exception("Nonce is too small!") - - raise Exception("No HMAC connection initialized!") - - def get_access_token(self): - return self.access_token - - def _set_hmac(self, server, hmac_key, hmac_secret): - self.server = server - self.hmac_key = hmac_key.encode("ascii") - self.hmac_secret = hmac_secret.encode("ascii") +SERVER = "https://localbitcoins.com" class LocalBitcoins: @@ -135,10 +50,18 @@ class LocalBitcoins: # pylint: disable=too-many-public-methods # API provides this many methods, I can't change that - def __init__(self, api_key: Optional[str], debug: Optional[bool] = False) -> None: - self.api_key = "" - if api_key: - self.api_key = api_key + def __init__( + self, + hmac_key: Optional[str], + hmac_secret: Optional[str], + debug: Optional[bool] = False, + ) -> None: + self.hmac_key = "" + self.hmac_secret = "" + if hmac_key: + self.hmac_key = hmac_key.encode("ascii") + if hmac_secret: + self.hmac_secret = hmac_secret.encode("ascii") self.debug = debug if self.debug: @@ -147,20 +70,62 @@ class LocalBitcoins: else: logger.setLevel(logging.INFO) - logger.debug("creating instance of LocalBitcoins API with api_key %s", self.api_key) + logger.debug( + "creating instance of LocalBitcoins API with api_key %s", + self.hmac_key, + ) + + def sign_payload(self, nonce, url, params_encoded): + """ + Sign the payload with our HMAC keys. + """ + # Calculate signature + message = nonce + self.hmac_key + url.encode("ascii") + print("message", message) + if params_encoded: + if sys.version_info >= (3, 0) and isinstance(params_encoded, str): + message += params_encoded.encode("ascii") + print("after message appended", message) + else: + message += params_encoded + signature = hmac_lib.new(self.hmac_secret, msg=message, digestmod=hashlib.sha256).hexdigest().upper() + print("signature", signature) + return signature + + def encode_params(self, api_method, api_call_url, query_values): + if api_method == "POST": + api_request = requests.Request("POST", api_call_url, data=query_values).prepare() + params_encoded = api_request.body + + # GET method + else: + api_request = requests.Request("GET", api_call_url, params=query_values).prepare() + params_encoded = urlparse(api_request.url).query + return params_encoded def _api_call( self, api_method: str, http_method: Optional[str] = "GET", query_values: Optional[Dict[str, Any]] = None, + files: Optional[Any] = None, ) -> Dict[str, Any]: + api_method += "/" api_call_url = URI_API + api_method + url = api_call_url + if url.startswith(SERVER): + url = url[len(SERVER) :] # noqa + + # HMAC crypto stuff + params_encoded = self.encode_params(api_method, api_call_url, query_values) + nonce = str(int(time.time() * 1000)).encode("ascii") + signature = self.sign_payload(nonce, url, params_encoded) + headers = { - "Content-Type": "application/json", - "User-Agent": f"localbitcoins_py/{__version__} " f"https://git.zm.is/Pathogen/pluto", - "Authorization": self.api_key, + "Apiauth-Key": self.hmac_key, + "Apiauth-Nonce": nonce, + "Apiauth-Signature": signature, } logger.debug("API Call URL: %s", api_call_url) @@ -202,7 +167,7 @@ class LocalBitcoins: result["message"] = "OK" else: result["message"] = "API ERROR" - + print("RESP", result) return result except httpx.ConnectError as error: result["message"] = str(error) @@ -716,7 +681,10 @@ class LocalBitcoins: return self._api_call( api_method="equation", http_method="POST", - query_values={"price_equation": price_equation, "currency": currency}, + query_values={ + "price_equation": price_equation, + "currency": currency, + }, ) # Public ad search related API Methods @@ -883,4 +851,8 @@ class LocalBitcoins: if pincode: params["pincode"] = pincode - return self._api_call(api_method="wallet-send-pin", http_method="POST", query_values=params) + return self._api_call( + api_method="wallet-send-pin", + http_method="POST", + query_values=params, + )