Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def start_jsonrpc(self, config: SimpleConfig, fd):
self.app = web.Application()
self.app.router.add_post("/", self.handle)
self.rpc_user, self.rpc_password = get_rpc_credentials(config)
self.methods = jsonrpcserver.methods.Methods()
self.methods.add(self.ping)
self.methods.add(self.gui)
self.methods.add(self.daemon)
self.cmd_runner = Commands(self.config, None, self.network, self)
for cmdname in known_commands:
self.methods.add(getattr(self.cmd_runner, cmdname))
self.methods.add(self.run_cmdline)
self.host = config.get('rpchost', '127.0.0.1')
self.port = config.get('rpcport', 0)
self.runner = web.AppRunner(self.app)
await self.runner.setup()
site = web.TCPSite(self.runner, self.host, self.port)
await site.start()
socket = site._server.sockets[0]
os.write(fd, bytes(repr((socket.getsockname(), time.time())), 'utf8'))
os.close(fd)
def dispatch():
try:
req = json.loads(request.get_data().decode())
except JSONDecodeError:
return Response(
PARSE_ERROR_RESPONSE,
400,
mimetype='application/json'
)
else:
response = methods.dispatch(req)
return Response(str(response),
response.http_status,
mimetype='application/json')
def dispatch():
req = json.loads(request.get_data().decode())
response = methods.dispatch(req)
return Response(str(response),
response.http_status,
mimetype='application/json')
def __init__(self):
self.methods = jsonrpcserver.methods.Methods()
def __init__(self, daemon, fd):
rpc_user, rpc_password = get_rpc_credentials(daemon.config)
AuthenticatedServer.__init__(self, rpc_user, rpc_password)
self.daemon = daemon
self.fd = fd
self.config = daemon.config
self.host = self.config.get('rpchost', '127.0.0.1')
self.port = self.config.get('rpcport', 0)
self.app = web.Application()
self.app.router.add_post("/", self.handle)
self.methods = jsonrpcserver.methods.Methods()
self.methods.add(self.ping)
self.methods.add(self.gui)
self.cmd_runner = Commands(config=self.config, network=self.daemon.network, daemon=self.daemon)
for cmdname in known_commands:
self.methods.add(getattr(self.cmd_runner, cmdname))
self.methods.add(self.run_cmdline)