rostrum 14.0.1

"""
Copyright 2011 Jeff Garzik

AuthServiceProxy has the following improvements over python-jsonrpc's
ServiceProxy class:

- HTTP connections persist for the life of the AuthServiceProxy object
(if server supports HTTP/1.1)
- sends protocol 'version', per JSON-RPC 1.1
- sends proper, incrementing 'id'
- sends Basic HTTP authentication headers
- parses all JSON numbers that look like floats as Decimal
- uses standard Python json lib

Previous copyright, from python-jsonrpc/jsonrpc/proxy.py:

Copyright (c) 2007 Jan-Klaas Kollhof

This file is part of jsonrpc.

jsonrpc is free software; you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation; either version 2.1 of the License, or
(at your option) any later version.

This software is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public License
along with this software; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
"""

import http.client as httplib
import base64
import decimal
import json
import logging

USER_AGENT = "AuthServiceProxy/0.1"

HTTP_TIMEOUT = 120

log = logging.getLogger("BitcoinRPC")


class JSONRPCException(Exception):
    def __init__(self, rpc_error):
        Exception.__init__(self)
        self.error = rpc_error

    def __str__(self):
        return f"{self.error['code']}: {self.error['message']}"


def encode_decimal(o):
    if isinstance(o, decimal.Decimal):
        return str(o)
    raise TypeError(repr(o) + " is not JSON serializable")


# pylint: disable=too-many-instance-attributes,too-many-arguments
class AuthServiceProxy:
    __id_count = 0

    hostname: str
    port: int
    cookie: bytes

    # ensure_ascii: escape unicode as \uXXXX, passed to json.dumps
    # pylint: disable=too-many-positional-arguments
    def __init__(
        self,
        hostname,
        port,
        cookie,
        service_name=None,
        timeout=HTTP_TIMEOUT,
        connection=None,
        ensure_ascii=True,
    ):
        self.hostname = hostname
        self.port = port
        self.cookie = cookie

        self.__timeout = timeout
        self._service_name = service_name
        self.ensure_ascii = ensure_ascii  # can be toggled on the fly by tests
        self.__auth_header = b"Basic " + base64.b64encode(cookie)

        if connection:
            # Callables re-use the connection of the original proxy
            self.__conn = connection
        else:
            self.__conn = httplib.HTTPConnection(self.hostname, port, timeout=timeout)

    def reconnect(self):
        self.__conn = httplib.HTTPConnection(
            self.hostname, self.port, timeout=self.__timeout
        )

    def __getattr__(self, name):
        if name.startswith("__") and name.endswith("__"):
            # Python internal stuff
            raise AttributeError
        if self._service_name is not None:
            name = f"{self._service_name}.{name}"
        return AuthServiceProxy(
            self.hostname, self.port, self.cookie, name, connection=self.__conn
        )

    def _request(self, method, path, postdata):
        """
        Do a HTTP request, with retry if we get disconnected (e.g. due to a timeout).
        This is a workaround for https://bugs.python.org/issue3566 which is fixed in Python 3.5.
        """
        headers = {
            "Host": self.hostname,
            "User-Agent": USER_AGENT,
            "Authorization": self.__auth_header,
            "Content-type": "application/json",
        }
        while True:
            try:
                self.__conn.request(method, path, postdata, headers)
                return self._get_response()
            except httplib.CannotSendRequest as e:
                log.error(
                    "%s (port %s): Cannot send request: %s",
                    self.hostname,
                    self.port,
                    str(e),
                )
                self.reconnect()
            except httplib.BadStatusLine as e:
                if e.line == "''":  # if connection was closed, try again
                    self.__conn.close()
                    self.__conn.request(method, path, postdata, headers)
                    return self._get_response()

                raise
            except (BrokenPipeError, ConnectionResetError):
                # Python 3.5+ raises this instead of BadStatusLine when the
                # connection was reset
                self.__conn.close()
                self.__conn.request(method, path, postdata, headers)
                return self._get_response()

    def __call__(self, *args):
        AuthServiceProxy.__id_count += 1

        log.debug(
            "-%s-> %s %s",
            AuthServiceProxy.__id_count,
            self._service_name,
            json.dumps(args, default=encode_decimal, ensure_ascii=self.ensure_ascii),
        )
        postdata = json.dumps(
            {
                "version": "1.1",
                "method": self._service_name,
                "params": args,
                "id": AuthServiceProxy.__id_count,
            },
            default=encode_decimal,
            ensure_ascii=self.ensure_ascii,
        )
        response = self._request("POST", "/", postdata.encode("utf-8"))
        if response["error"] is not None:
            raise JSONRPCException(response["error"])
        if "result" not in response:
            raise JSONRPCException({"code": -343, "message": "missing JSON-RPC result"})

        return response["result"]

    def _batch(self, rpc_call_list):
        postdata = json.dumps(
            list(rpc_call_list), default=encode_decimal, ensure_ascii=self.ensure_ascii
        )
        log.debug("--> %s", postdata)
        return self._request("POST", "/", postdata.encode("utf-8"))

    def _get_response(self):
        http_response = self.__conn.getresponse()
        if http_response is None:
            raise JSONRPCException(
                {"code": -342, "message": "missing HTTP response from server"}
            )

        content_type = http_response.getheader("Content-Type")
        if content_type != "application/json":
            raise JSONRPCException(
                {
                    "code": -342,
                    "message": f"non-JSON HTTP response with '{http_response.status} {http_response.reason}' from server",
                }
            )

        responsedata = http_response.read().decode("utf8")
        response = json.loads(responsedata, parse_float=decimal.Decimal)
        if "error" in response and response["error"] is None:
            log.debug(
                "<-%s- %s",
                response["id"],
                json.dumps(
                    response["result"],
                    default=encode_decimal,
                    ensure_ascii=self.ensure_ascii,
                ),
            )
        else:
            log.debug("<-- %s", responsedata)
        return response