Add admin rpc server (#72)

* Add admin rpc server

* Use the admin rpc to get the balance of the server in itest

docker/test/Dockerfile

@@ -17,6 +17,7 @@ RUN pip3 install -r requirements-itest.txt
 
 Run mkdir /app
 COPY ./squeakserver/common/rpc/squeak_server.proto /app
+COPY ./squeakserver/admin/rpc/squeak_admin.proto /app
 RUN cp -r googleapis /app
 RUN cp rpc.proto /app/lnd.proto
 
@@ -24,6 +25,7 @@ WORKDIR /app
 
 RUN python3 -m grpc_tools.protoc --proto_path=googleapis:. --python_out=. --grpc_python_out=. lnd.proto
 RUN python3 -m grpc_tools.protoc --proto_path=googleapis:. --python_out=. --grpc_python_out=. squeak_server.proto
+RUN python3 -m grpc_tools.protoc --proto_path=googleapis:. --python_out=. --grpc_python_out=. squeak_admin.proto
 
 # Copy the lighting client
 COPY "squeakserver/common/lnd_lightning_client.py" .

install-rpc.sh

@@ -17,3 +17,6 @@ python3 -m grpc_tools.protoc --proto_path=googleapis:. --python_out=. --grpc_pyt
 
 # install squeak server protocol
 python3 -m grpc_tools.protoc --proto_path=googleapis:. --python_out=. --grpc_python_out=. squeakserver/common/rpc/squeak_server.proto
+
+# install squeak admin server protocol
+python3 -m grpc_tools.protoc --proto_path=googleapis:. --python_out=. --grpc_python_out=. squeakserver/admin/rpc/squeak_admin.proto

itests/config.ini

@@ -20,6 +20,10 @@ rpc_host=0.0.0.0
 rpc_port=8774
 price=100
 
+[admin]
+rpc_host=0.0.0.0
+rpc_port=8994
+
 [postgresql]
 host=db
 database=squeakserver

itests/run_itest.sh

@@ -34,4 +34,4 @@ echo "Running test.sh...."
 docker-compose run test ./test.sh
 
 echo "Shutting down itest..."
-docker-compose down
+# docker-compose down

itests/test.py

@@ -21,6 +21,8 @@ import lnd_pb2_grpc as lnrpc
 import grpc
 import squeak_server_pb2
 import squeak_server_pb2_grpc
+import squeak_admin_pb2
+import squeak_admin_pb2_grpc
 
 from lnd_lightning_client import LNDLightningClient
 
@@ -97,6 +99,10 @@ def bxor(b1, b2): # use xor for bytes
     return bytes(result)
 
 
+def string_to_hex(s):
+    return bytes.fromhex(s)
+
+
 def run():
     # Set the network to simnet for itest.
     SelectParams("mainnet")
@@ -104,7 +110,8 @@ def run():
     # NOTE(gRPC Python Team): .close() is possible on a channel and should be
     # used in circumstances in which the with statement does not fit the needs
     # of the code.
-    with grpc.insecure_channel('sqkserver:8774') as server_channel:
+    with grpc.insecure_channel('sqkserver:8774') as server_channel,\
+         grpc.insecure_channel('sqkserver:8994') as admin_channel:
 
         # load lnd client
         lnd_lightning_client = load_lightning_client()
@@ -114,6 +121,7 @@ def run():
 
         # Make the stubs
         server_stub = squeak_server_pb2_grpc.SqueakServerStub(server_channel)
+        admin_stub = squeak_admin_pb2_grpc.SqueakAdminStub(admin_channel)
 
         # Post a squeak with a direct request to the server
         signing_key = generate_signing_key()
@@ -198,19 +206,25 @@ def run():
         print("Server list peers response: " + str(list_peers_response))
 
         # Open channel to the server lightning node
-        open_channel_response = lnd_lightning_client.open_channel_sync(buy_response.offer.pubkey, 1000000)
+        pubkey_bytes = string_to_hex(buy_response.offer.pubkey)
+        open_channel_response = lnd_lightning_client.open_channel(pubkey_bytes, 1000000)
         print("Server open channel response: " + str(open_channel_response))
+        for update in open_channel_response:
+            if update.HasField('chan_open'):
+                channel_point = update.chan_open.channel_point
+                print("Channel now open: " + str(channel_point))
+                break
 
         # List channels
         list_channels_response = lnd_lightning_client.list_channels()
         print("Server list channels response: " + str(list_channels_response))
 
-        # Sleep for 60 seconds to confirm the channel open transaction
-        time.sleep(60)
+        # # Sleep for 60 seconds to confirm the channel open transaction
+        # time.sleep(60)
 
-        # List channels
-        list_channels_response = lnd_lightning_client.list_channels()
-        print("Server list channels response: " + str(list_channels_response))
+        # # List channels
+        # list_channels_response = lnd_lightning_client.list_channels()
+        # print("Server list channels response: " + str(list_channels_response))
 
         # Pay the invoice
         payment = lnd_lightning_client.pay_invoice_sync(buy_response.offer.payment_request)
@@ -234,6 +248,22 @@ def run():
         assert get_response_squeak.GetDecryptedContentStr() == 'hello from itest!'
         print("Finished checking squeak.")
 
+        # Check the server balance
+        get_balance_response = admin_stub.GetBalance(squeak_admin_pb2.GetBalanceRequest())
+        print("Get balance response balance: " + str(get_balance_response.balance))
+        assert get_balance_response.balance == 0
+
+        # Close the channel
+        time.sleep(10)
+        for update in lnd_lightning_client.close_channel(channel_point):
+            if update.HasField('chan_close'):
+                print("Channel closed.")
+                break
+
+        # Check the server balance
+        get_balance_response = admin_stub.GetBalance(squeak_admin_pb2.GetBalanceRequest())
+        print("Get balance response balance: " + str(get_balance_response.balance))
+        # assert get_balance_response.balance > 0
 
 
 if __name__ == '__main__':

squeakserver/admin/rpc/__init__.py

@@ -0,0 +1 @@
+__version__ = '0.1.0'

squeakserver/admin/rpc/squeak_admin.proto

@@ -0,0 +1,25 @@
+syntax = "proto3";
+
+option java_multiple_files = true;
+option java_package = "io.github.yzernik.squeakserver";
+option java_outer_classname = "SqueakAdminProto";
+option objc_class_prefix = "SQK";
+
+package squeakserver;
+
+// Interface exported by the server.
+service SqueakAdmin {
+
+  /** sqkadmin: `getsqueak`
+  */
+  rpc GetBalance (GetBalanceRequest) returns (GetBalanceReply) {}
+
+}
+
+message GetBalanceRequest {
+}
+
+message GetBalanceReply {
+    /// The wallet balance
+    int64 balance = 1;
+}

squeakserver/admin/squeak_admin_server_handler.py

@@ -0,0 +1,37 @@
+import logging
+import threading
+
+from squeak.core.encryption import generate_initialization_vector
+from squeak.core.encryption import CEncryptedDecryptionKey
+from squeak.core.signing import CSigningKey
+from squeak.core.signing import CSqueakAddress
+
+from squeakserver.server.buy_offer import BuyOffer
+from squeakserver.common.lnd_lightning_client import LNDLightningClient
+from squeakserver.server.lightning_address import LightningAddressHostPort
+from squeakserver.server.postgres_db import PostgresDb
+from squeakserver.server.util import generate_offer_preimage
+from squeakserver.server.util import bxor
+from squeakserver.server.util import get_hash
+
+
+logger = logging.getLogger(__name__)
+
+
+class SqueakAdminServerHandler(object):
+    """Handles admin server commands.
+    """
+
+    def __init__(
+            self,
+            lightning_client: LNDLightningClient,
+            postgres_db: PostgresDb,
+    ) -> None:
+        self.lightning_client = lightning_client
+        self.postgres_db = postgres_db
+
+    def handle_get_balance(self):
+        logger.info("Handle get balance")
+        wallet_balance = self.lightning_client.get_wallet_balance()
+        logger.info("Wallet balance: {}".format(wallet_balance))
+        return wallet_balance.total_balance

squeakserver/admin/squeak_admin_server_servicer.py

@@ -0,0 +1,36 @@
+import logging
+from concurrent import futures
+
+import grpc
+
+from squeak.core import CSqueak
+
+from squeakserver.admin.rpc import squeak_admin_pb2
+from squeakserver.admin.rpc import squeak_admin_pb2_grpc
+from squeakserver.server.util import get_hash
+
+
+logger = logging.getLogger(__name__)
+
+
+class SqueakAdminServerServicer(squeak_admin_pb2_grpc.SqueakAdminServicer):
+    """Provides methods that implement functionality of squeak admin server."""
+
+    def __init__(self, host, port, handler):
+        self.host = host
+        self.port = port
+        self.handler = handler
+
+    def GetBalance(self, request, context):
+        total_balance = self.handler.handle_get_balance()
+        return squeak_admin_pb2.GetBalanceReply(
+            balance=total_balance,
+        )
+
+    def serve(self):
+        server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
+        squeak_admin_pb2_grpc.add_SqueakAdminServicer_to_server(
+            self, server)
+        server.add_insecure_port('{}:{}'.format(self.host, self.port))
+        server.start()
+        server.wait_for_termination()

squeakserver/common/lnd_lightning_client.py

@@ -122,3 +122,27 @@ class LNDLightningClient():
         """
         list_peers_request = self.ln_module.ListPeersRequest()
         return self.stub.ListPeers(list_peers_request, metadata=[('macaroon', self.macaroon)])
+
+    def open_channel(self, pubkey, local_amount):
+        """ Open a channel
+
+        args:
+        pubkey (bytes) -- The identity pubkey of the Lightning node
+        local_amount -- The number of satoshis the wallet should commit to the channel
+        """
+        open_channel_request = self.ln_module.OpenChannelRequest(
+            node_pubkey=pubkey,
+            local_funding_amount=local_amount,
+        )
+        return self.stub.OpenChannel(open_channel_request, metadata=[('macaroon', self.macaroon)])
+
+    def close_channel(self, channel_point):
+        """ Close a channel
+
+        args:
+        channel_point (str) -- The outpoint (txid:index) of the funding transaction.
+        """
+        close_channel_request = self.ln_module.CloseChannelRequest(
+            channel_point=channel_point,
+        )
+        return self.stub.CloseChannel(close_channel_request, metadata=[('macaroon', self.macaroon)])

squeakserver/server/main.py

@@ -14,6 +14,8 @@ from squeak.core.signing import CSigningKey
 import squeakserver.common.rpc.lnd_pb2 as ln
 import squeakserver.common.rpc.lnd_pb2_grpc as lnrpc
 
+from squeakserver.admin.squeak_admin_server_servicer import SqueakAdminServerServicer
+from squeakserver.admin.squeak_admin_server_handler import SqueakAdminServerHandler
 from squeakserver.common.lnd_lightning_client import LNDLightningClient
 from squeakserver.server.lightning_address import LightningAddressHostPort
 from squeakserver.server.squeak_server_servicer import SqueakServerServicer
@@ -22,6 +24,9 @@ from squeakserver.server.db_params import parse_db_params
 from squeakserver.server.postgres_db import PostgresDb
 
 
+logger = logging.getLogger(__name__)
+
+
 def load_lightning_client(config) -> LNDLightningClient:
     if int(config['server']['price']) == 0:
         return None
@@ -47,7 +52,6 @@ def load_lightning_host_port(config) -> LNDLightningClient:
         lnd_port,
     )
 
-
 def load_rpc_server(config, handler) -> SqueakServerServicer:
     return SqueakServerServicer(
         config['server']['rpc_host'],
@@ -55,24 +59,16 @@ def load_rpc_server(config, handler) -> SqueakServerServicer:
         handler,
     )
 
+def load_admin_rpc_server(config, handler) -> SqueakAdminServerServicer:
+    return SqueakAdminServerServicer(
+        config['admin']['rpc_host'],
+        config['admin']['rpc_port'],
+        handler,
+    )
 
 def load_price(config):
     return int(config['server']['price'])
 
-
-def start_rpc_server(handler):
-    print('Calling start_rpc_server...', flush=True)
-    server = SqueakServerServicer(handler)
-    # thread = threading.Thread(
-    #     target=server.serve,
-    #     args=(),
-    # )
-    # thread.daemon = True
-    # thread.start()
-    # return server, thread
-    server.serve()
-
-
 def load_handler(lightning_host_port, lightning_client, postgres_db, price):
     return SqueakServerHandler(
         lightning_host_port,
@@ -81,6 +77,11 @@ def load_handler(lightning_host_port, lightning_client, postgres_db, price):
         price,
     )
 
+def load_admin_handler(lightning_client, postgres_db):
+    return SqueakAdminServerHandler(
+        lightning_client,
+        postgres_db,
+    )
 
 def load_db_params(config):
     return parse_db_params(config)
@@ -95,6 +96,14 @@ def sigterm_handler(_signo, _stack_frame):
     # Raises SystemExit(0):
     sys.exit(0)
 
+def start_admin_rpc_server(rpc_server):
+    logger.info('Calling start_admin_rpc_server...')
+    thread = threading.Thread(
+        target=rpc_server.serve,
+        args=(),
+    )
+    thread.daemon = True
+    thread.start()
 
 def parse_args():
     parser = argparse.ArgumentParser(
@@ -127,13 +136,13 @@ def parse_args():
 
 
 def main():
-    print("Running main() in server...", flush=True)
+    logger.info("Running main() in server...")
     logging.basicConfig(level=logging.ERROR)
     args = parse_args()
 
     # Set the log level
     level = args.log_level.upper()
-    print("level: " + level)
+    logger.info("level: " + level)
     logging.getLogger().setLevel(level)
 
     # Get the config object
@@ -147,49 +156,39 @@ def main():
 #     db_factory = load_db_factory(config)
 #     with db_factory.make_conn() as conn:
 #         initialize_db(conn)
-#     print("Initialized the database.")
+#     logger.info("Initialized the database.")
 
 
 def run_server(config):
-    print('network:', config['DEFAULT']['network'], flush=True)
+    logger.info('network: ' + config['DEFAULT']['network'])
     # SelectParams(config['DEFAULT']['network'])
     SelectParams("mainnet")
 
     # load the db params
     db_params = load_db_params(config)
-    print('db params: ' + str(db_params), flush=True)
+    logger.info('db params: ' + str(db_params))
 
     # load postgres db
     postgres_db = load_postgres_db(config)
-    print('postgres_db: ' + str(postgres_db), flush=True)
+    logger.info('postgres_db: ' + str(postgres_db))
     postgres_db.get_version()
     postgres_db.init()
 
-    print('starting lightning client here...', flush=True)
+    logger.info('starting lightning client here...')
     price = load_price(config)
     lightning_client = load_lightning_client(config)
     lightning_host_port = load_lightning_host_port(config)
-    # db_factory = load_db_factory(config)
-    handler = load_handler(lightning_host_port, lightning_client, postgres_db, price)
+
+    # start admin rpc server
+    admin_handler = load_admin_handler(lightning_client, postgres_db)
+    admin_rpc_server = load_admin_rpc_server(config, admin_handler)
+    start_admin_rpc_server(admin_rpc_server)
 
     # start rpc server
-    # start_rpc_server(handler)
-
+    handler = load_handler(lightning_host_port, lightning_client, postgres_db, price)
     server = load_rpc_server(config, handler)
     server.serve()
 
-    # rpc_server, rpc_server_thread = start_rpc_server(handler)
-    # print("rpc server started...", flush=True)
-
-    # signal.signal(signal.SIGTERM, sigterm_handler)
-
-    # print("sleeping....", flush=True)
-    # try:
-    #     while True:
-    #         time.sleep(1)
-    # finally:
-    #     print("Shutting down...", flush=True)
-
 
 if __name__ == '__main__':
     main()