Merge #1252: Add RPC API endpoint for tumbler

1d5728f Fix duplicated log issue (Daniel)
43b4eca Add api docs (Daniel)
a2d6f40 Add rpc endpoint for tumbler (Daniel)

docs/api/wallet-rpc.yaml

@@ -348,6 +348,61 @@ paths:
           $ref: '#/components/responses/409-NoConfig'
         '503':
           $ref: '#/components/responses/503-ServiceUnavailable'
+  /wallet/{walletname}/taker/schedule:
+    post:
+      security:
+        - bearerAuth: []
+      summary: create and run a schedule of transactions
+      operationId: runschedule
+      description: Creates and then starts a schedule of transactions.
+      parameters:
+        - name: walletname
+          in: path
+          description: name of wallet including .jmdat
+          required: true
+          schema:
+            type: string
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/RunScheduleRequest'
+        description: taker side schedule parameters
+      responses:
+        '202':
+          $ref: "#/components/responses/RunSchedule-202-OK"
+        '400':
+          $ref: '#/components/responses/400-BadRequest'
+        '401':
+          $ref: '#/components/responses/401-Unauthorized'
+        '404':
+          $ref: '#/components/responses/404-NotFound'
+        '409':
+          $ref: '#/components/responses/409-NoConfig'
+        '503':
+          $ref: '#/components/responses/503-ServiceUnavailable'
+    get:
+      security:
+        - bearerAuth: []
+      summary: get the schedule that is currently running
+      operationId: getschedule
+      description: Get the current transaction schedule if one is running.
+      parameters:
+        - name: walletname
+          in: path
+          description: name of the wallet including .jmdat
+          required: true
+          schema:
+            type: string
+      responses:
+        '200':
+          $ref: "#/components/responses/GetSchedule-200-OK"
+        '400':
+          $ref: '#/components/responses/400-BadRequest'
+        '401':
+          $ref: "#/components/responses/401-Unauthorized"
+        '404':
+          $ref: '#/components/responses/404-NotFound'
   /wallet/{walletname}/taker/stop:
     get:
       security:
@@ -550,7 +605,16 @@ components:
         destination:
           type: string
           example: "bcrt1qujp2x2fv437493sm25gfjycns7d39exjnpptzw"
-
+    RunScheduleRequest:
+      type: object
+      required:
+        - destinations
+      properties:
+        destinations:
+          type: array
+          items:
+            type: string
+            example: "bcrt1qujp2x2fv437493sm25gfjycns7d39exjnpptzw"
     StartMakerRequest:
       type: object
       required:
@@ -767,6 +831,17 @@ components:
       properties:
         seedphrase:
           type: string
+    GetScheduleResponse:
+      type: object
+      required:
+        - schedule
+      properties:
+        schedule:
+          type: array
+          items:
+            oneOf:
+              - type: string
+              - type: integer
     LockWalletResponse:
       type: object
       required:
@@ -907,6 +982,18 @@ components:
         application/json:
           schema:
             $ref: "#/components/schemas/FreezeResponse"
+    RunSchedule-202-OK:
+      description: "schedule started successfully"
+      content:
+        application/json:
+          schema:
+            $ref: "#/components/schemas/GetScheduleResponse"
+    GetSchedule-200-OK:
+      description: "schedule retrieved successfully"
+      content:
+        application/json:
+          schema:
+            $ref: "#/components/schemas/GetScheduleResponse"
     YieldGenReport-200-OK:
       description: "get list of coinjoins taken part in as maker (across all wallets)"
       content:

jmclient/jmclient/wallet_rpc.py

@@ -10,6 +10,7 @@
 from autobahn.twisted.websocket import listenWS
 from klein import Klein
 import jwt
+import pprint
 
 from jmbitcoin import human_readable_transaction
 from jmclient import Taker, jm_single, \
@@ -20,7 +21,11 @@
     create_wallet, get_max_cj_fee_values, \
     StorageError, StoragePasswordError, JmwalletdWebSocketServerFactory, \
     JmwalletdWebSocketServerProtocol, RetryableStorageError, \
-    SegwitWalletFidelityBonds, wallet_gettimelockaddress, NotEnoughFundsException
+    SegwitWalletFidelityBonds, wallet_gettimelockaddress, \
+    NotEnoughFundsException, get_tumble_log, get_tumble_schedule, \
+    get_schedule, get_tumbler_parser, schedule_to_text, \
+    tumbler_filter_orders_callback, tumbler_taker_finished_update, \
+    validate_address
 from jmbase.support import get_log, utxostr_to_utxo
 
 jlog = get_log()
@@ -158,6 +163,11 @@ def __init__(self, port, wss_port, tls=True):
         # keep track of client side connections so they
         # can be shut down cleanly:
         self.coinjoin_connection = None
+        # Options for generating a tumble schedule / running the tumbler.
+        # Doubles as flag for indicating whether we're currently running
+        # a tumble schedule.
+        self.tumbler_options = None
+        self.tumble_log = None
 
     def get_client_factory(self):
         return JMClientProtocolFactory(self.taker)
@@ -429,14 +439,27 @@ def dummy_restart_callback(msg):
                         token=self.cookie)
 
     def taker_finished(self, res, fromtx=False, waittime=0.0, txdetails=None):
-        # This is a slimmed down version compared with what is seen in
-        # the CLI code, since that code encompasses schedules with multiple
-        # entries; for now, the RPC only supports single joins.
-        # TODO this may be updated.
-        # It is also different in that the event loop must not shut down
-        # when processing finishes.
-
-        # reset our state on completion, we are no longer coinjoining:
+        if not self.tumbler_options:
+            # We were doing a single coinjoin -- stop taker.
+            self.stop_taker(res)
+        else:
+            # We're running the tumbler.
+            assert self.tumble_log is not None
+
+            logsdir = os.path.join(os.path.dirname(jm_single().config_location), "logs")
+            sfile = os.path.join(logsdir, self.tumbler_options['schedulefile'])
+
+            tumbler_taker_finished_update(self.taker, sfile, self.tumble_log, self.tumbler_options, res, fromtx, waittime, txdetails)
+
+            if not fromtx:
+                # The tumbling schedule's final transaction is done.
+                self.stop_taker(res)
+                self.tumbler_options = None
+            elif fromtx != "unconfirmed":
+                # A non-final transaction in the tumbling schedule is done -- continue schedule after wait time.
+                reactor.callLater(waittime*60, self.clientfactory.getClient().clientStart)
+
+    def stop_taker(self, res):
         self.taker = None
 
         if not res:
@@ -469,6 +492,9 @@ def filter_orders_callback(self,orderfees, cjamount):
         """
         return True
 
+    def filter_orders_callback_tumbler(self, orders_fees, cjamount):
+        return tumbler_filter_orders_callback(orders_fees, cjamount, self.taker)
+
     def check_daemon_ready(self):
         # daemon must be up before coinjoins start.
         daemon_serving_host, daemon_serving_port = get_daemon_serving_params()
@@ -1032,3 +1058,113 @@ def getseed(self, request, walletname):
                 raise InvalidRequestFormat()
             seedphrase, _ = self.services["wallet"].get_mnemonic_words()
             return make_jmwalletd_response(request, seedphrase=seedphrase)
+
+        @app.route('/wallet/<string:walletname>/taker/schedule', methods=['POST'])
+        def start_tumbler(self, request, walletname):
+            self.check_cookie(request)
+
+            if not self.services["wallet"]:
+                raise NoWalletFound()
+            if not self.wallet_name == walletname:
+                raise InvalidRequestFormat()
+
+            # -- Options parsing -----------------------------------------------
+
+            (options, args) = get_tumbler_parser().parse_args([])
+            self.tumbler_options = vars(options)
+
+            # At the moment only the destination addresses can be set.
+            # For now, all other options are hardcoded to the defaults of
+            # the tumbler.py script.
+            request_data = self.get_POST_body(request, ["destination_addresses"])
+            if not request_data:
+                raise InvalidRequestFormat()
+
+            destaddrs = request_data["destination_addresses"]
+            for daddr in destaddrs:
+                success, _ = validate_address(daddr)
+                if not success:
+                    raise InvalidRequestFormat()
+
+            # Setting max_cj_fee based on global config.
+            # We won't respect it being set via tumbler_options for now.
+            def dummy_user_callback(rel, abs):
+                raise ConfigNotPresent()
+
+            max_cj_fee = get_max_cj_fee_values(jm_single().config,
+                                               None,
+                                               user_callback=dummy_user_callback)
+
+            jm_single().mincjamount = self.tumbler_options['mincjamount']
+
+            # -- Schedule generation -------------------------------------------
+
+            # Always generates a new schedule. No restart support for now.
+            schedule = get_tumble_schedule(self.tumbler_options,
+                                           destaddrs,
+                                           self.services["wallet"].get_balance_by_mixdepth())
+
+            logsdir = os.path.join(os.path.dirname(jm_single().config_location),
+                                   "logs")
+            sfile = os.path.join(logsdir, self.tumbler_options['schedulefile'])
+            with open(sfile, "wb") as f:
+                f.write(schedule_to_text(schedule))
+
+            if self.tumble_log is None:
+                self.tumble_log = get_tumble_log(logsdir)
+
+            self.tumble_log.info("TUMBLE STARTING")
+            self.tumble_log.info("With this schedule: ")
+            self.tumble_log.info(pprint.pformat(schedule))
+
+            # -- Running the Taker ---------------------------------------------
+
+            # For simplicity, we're not doing any fee estimation for now.
+            # We might want to add fee estimation (see scripts/tumbler.py) to
+            # prevent users from overspending on fees when tumbling with small
+            # amounts.
+
+            if not self.activate_coinjoin_state(CJ_TAKER_RUNNING):
+                raise ServiceAlreadyStarted()
+
+            self.taker = Taker(self.services["wallet"],
+                               schedule,
+                               max_cj_fee=max_cj_fee,
+                               order_chooser=self.tumbler_options['order_choose_fn'],
+                               callbacks=(self.filter_orders_callback_tumbler, None,
+                                          self.taker_finished),
+                               tdestaddrs=destaddrs)
+            self.clientfactory = self.get_client_factory()
+
+            self.taker.testflag = True
+
+            dhost, dport = self.check_daemon_ready()
+
+            _, self.coinjoin_connection = start_reactor(dhost,
+                                                        dport,
+                                                        self.clientfactory,
+                                                        rs=False)
+
+            return make_jmwalletd_response(request, status=202, schedule=schedule)
+
+        @app.route('/wallet/<string:walletname>/taker/schedule', methods=['GET'])
+        def get_tumbler_schedule(self, request, walletname):
+            self.check_cookie(request)
+
+            if not self.services["wallet"]:
+                raise NoWalletFound()
+            if not self.wallet_name == walletname:
+                raise InvalidRequestFormat()
+
+            if not self.tumbler_options:
+                return make_jmwalletd_response(request, status=404)
+
+
+            logsdir = os.path.join(os.path.dirname(jm_single().config_location), "logs")
+            sfile = os.path.join(logsdir, self.tumbler_options['schedulefile'])
+            res, schedule = get_schedule(sfile)
+
+            if not res:
+                return make_jmwalletd_response(request, status=500)
+
+            return make_jmwalletd_response(request, schedule=schedule)