nmode's Git Repositories - signal-cli/blob - src/main/java/org/asamk/signal/jsonrpc/JsonRpcReader.java

   1 package org.asamk.signal.jsonrpc;
   2
   3 import com.fasterxml.jackson.core.JsonParseException;
   4 import com.fasterxml.jackson.databind.JsonMappingException;
   5 import com.fasterxml.jackson.databind.JsonNode;
   6 import com.fasterxml.jackson.databind.ObjectMapper;
   7 import com.fasterxml.jackson.databind.node.ContainerNode;
   8 import com.fasterxml.jackson.databind.node.ObjectNode;
   9 import com.fasterxml.jackson.databind.node.ValueNode;
  10
  11 import org.asamk.signal.util.Util;
  12 import org.slf4j.Logger;
  13 import org.slf4j.LoggerFactory;
  14
  15 import java.io.IOException;
  16 import java.io.InputStream;
  17 import java.util.ArrayList;
  18 import java.util.concurrent.Executors;
  19 import java.util.concurrent.locks.ReentrantLock;
  20 import java.util.function.Consumer;
  21 import java.util.function.Supplier;
  22 import java.util.stream.StreamSupport;
  23
  24 public class JsonRpcReader {
  25
  26     private static final Logger logger = LoggerFactory.getLogger(JsonRpcReader.class);
  27
  28     private final JsonRpcSender jsonRpcSender;
  29     private final ObjectMapper objectMapper;
  30     private final InputStream input;
  31     private final Supplier<String> lineSupplier;
  32
  33     public JsonRpcReader(final JsonRpcSender jsonRpcSender, final Supplier<String> lineSupplier) {
  34         this.jsonRpcSender = jsonRpcSender;
  35         this.input = null;
  36         this.lineSupplier = lineSupplier;
  37         this.objectMapper = Util.createJsonObjectMapper();
  38     }
  39
  40     public JsonRpcReader(final JsonRpcSender jsonRpcSender, final InputStream input) {
  41         this.jsonRpcSender = jsonRpcSender;
  42         this.input = input;
  43         this.lineSupplier = null;
  44         this.objectMapper = Util.createJsonObjectMapper();
  45     }
  46
  47     public void readMessages(final RequestHandler requestHandler, final Consumer<JsonRpcResponse> responseHandler) {
  48         if (input != null) {
  49             JsonRpcMessage message = parseJsonRpcMessage(input);
  50             if (message == null) {
  51                 return;
  52             }
  53
  54             handleMessage(message, requestHandler, responseHandler);
  55             return;
  56         }
  57
  58         try (final var executor = Executors.newCachedThreadPool()) {
  59             while (!Thread.interrupted()) {
  60                 final var input = lineSupplier.get();
  61                 if (input == null) {
  62                     logger.trace("Reached end of JSON-RPC input stream.");
  63                     break;
  64                 }
  65
  66                 logger.trace("Incoming JSON-RPC message: {}", input);
  67                 final var message = parseJsonRpcMessage(input);
  68                 if (message == null) {
  69                     continue;
  70                 }
  71
  72                 executor.submit(() -> handleMessage(message, requestHandler, responseHandler));
  73             }
  74         }
  75     }
  76
  77     private void handleMessage(
  78             final JsonRpcMessage message,
  79             final RequestHandler requestHandler,
  80             final Consumer<JsonRpcResponse> responseHandler
  81     ) {
  82         switch (message) {
  83             case JsonRpcRequest jsonRpcRequest -> {
  84                 logger.debug("Received json rpc request, method: " + jsonRpcRequest.getMethod());
  85                 final var response = handleRequest(requestHandler, jsonRpcRequest);
  86                 if (response != null) {
  87                     jsonRpcSender.sendResponse(response);
  88                 }
  89             }
  90             case JsonRpcResponse jsonRpcResponse -> responseHandler.accept(jsonRpcResponse);
  91             case JsonRpcBatchMessage jsonRpcBatchMessage -> {
  92                 final var messages = jsonRpcBatchMessage.getMessages();
  93                 final var responseList = new ArrayList<JsonRpcResponse>(messages.size());
  94                 try (final var executor = Executors.newCachedThreadPool()) {
  95                     final var lock = new ReentrantLock();
  96                     messages.forEach(jsonNode -> {
  97                         final JsonRpcRequest request;
  98                         try {
  99                             request = parseJsonRpcRequest(jsonNode);
 100                         } catch (JsonRpcException e) {
 101                             final var response = JsonRpcResponse.forError(e.getError(), getId(jsonNode));
 102                             lock.lock();
 103                             try {
 104                                 responseList.add(response);
 105                             } finally {
 106                                 lock.unlock();
 107                             }
 108                             return;
 109                         }
 110
 111                         executor.submit(() -> {
 112                             final var response = handleRequest(requestHandler, request);
 113                             if (response != null) {
 114                                 lock.lock();
 115                                 try {
 116                                     responseList.add(response);
 117                                 } finally {
 118                                     lock.unlock();
 119                                 }
 120                             }
 121                         });
 122                     });
 123                 }
 124
 125                 if (!responseList.isEmpty()) {
 126                     jsonRpcSender.sendBatchResponses(responseList);
 127                 }
 128             }
 129         }
 130     }
 131
 132     private JsonRpcResponse handleRequest(final RequestHandler requestHandler, final JsonRpcRequest request) {
 133         try {
 134             final var result = requestHandler.apply(request.getMethod(), request.getParams());
 135             if (request.getId() != null) {
 136                 return JsonRpcResponse.forSuccess(result, request.getId());
 137             } else {
 138                 logger.debug("Command '{}' succeeded but client didn't specify an id, dropping response",
 139                         request.getMethod());
 140             }
 141         } catch (JsonRpcException e) {
 142             if (request.getId() != null) {
 143                 return JsonRpcResponse.forError(e.getError(), request.getId());
 144             } else {
 145                 logger.debug("Command '{}' failed but client didn't specify an id, dropping error: {}",
 146                         request.getMethod(),
 147                         e.getMessage());
 148             }
 149         }
 150         return null;
 151     }
 152
 153     private JsonRpcMessage parseJsonRpcMessage(final String input) {
 154         if (input.trim().isEmpty()) {
 155             jsonRpcSender.sendResponse(JsonRpcResponse.forError(new JsonRpcResponse.Error(JsonRpcResponse.Error.PARSE_ERROR,
 156                     "Empty input line",
 157                     null), null));
 158             return null;
 159         }
 160
 161         final JsonNode jsonNode;
 162         try {
 163             jsonNode = objectMapper.readTree(input);
 164         } catch (JsonParseException e) {
 165             jsonRpcSender.sendResponse(JsonRpcResponse.forError(new JsonRpcResponse.Error(JsonRpcResponse.Error.PARSE_ERROR,
 166                     e.getMessage(),
 167                     null), null));
 168             return null;
 169         } catch (IOException e) {
 170             throw new AssertionError(e);
 171         }
 172
 173         return parseJsonRpcMessage(jsonNode);
 174     }
 175
 176     private JsonRpcMessage parseJsonRpcMessage(final InputStream input) {
 177         final JsonNode jsonNode;
 178         try {
 179             jsonNode = objectMapper.readTree(input);
 180         } catch (JsonParseException e) {
 181             jsonRpcSender.sendResponse(JsonRpcResponse.forError(new JsonRpcResponse.Error(JsonRpcResponse.Error.PARSE_ERROR,
 182                     e.getMessage(),
 183                     null), null));
 184             return null;
 185         } catch (IOException e) {
 186             throw new AssertionError(e);
 187         }
 188
 189         return parseJsonRpcMessage(jsonNode);
 190     }
 191
 192     private JsonRpcMessage parseJsonRpcMessage(final JsonNode jsonNode) {
 193         if (jsonNode == null) {
 194             jsonRpcSender.sendResponse(JsonRpcResponse.forError(new JsonRpcResponse.Error(JsonRpcResponse.Error.INVALID_REQUEST,
 195                     "invalid request",
 196                     null), null));
 197             return null;
 198         } else if (jsonNode.isArray()) {
 199             if (jsonNode.isEmpty()) {
 200                 jsonRpcSender.sendResponse(JsonRpcResponse.forError(new JsonRpcResponse.Error(JsonRpcResponse.Error.INVALID_REQUEST,
 201                         "invalid request",
 202                         null), null));
 203                 return null;
 204             }
 205             return new JsonRpcBatchMessage(StreamSupport.stream(jsonNode.spliterator(), false).toList());
 206         } else if (jsonNode.isObject()) {
 207             if (jsonNode.has("result") || jsonNode.has("error")) {
 208                 return parseJsonRpcResponse(jsonNode);
 209             } else {
 210                 try {
 211                     return parseJsonRpcRequest(jsonNode);
 212                 } catch (JsonRpcException e) {
 213                     jsonRpcSender.sendResponse(JsonRpcResponse.forError(e.getError(), getId(jsonNode)));
 214                     return null;
 215                 }
 216             }
 217         } else {
 218             jsonRpcSender.sendResponse(JsonRpcResponse.forError(new JsonRpcResponse.Error(JsonRpcResponse.Error.INVALID_REQUEST,
 219                     "unexpected type: " + jsonNode.getNodeType().name(),
 220                     null), null));
 221             return null;
 222         }
 223     }
 224
 225     private ValueNode getId(JsonNode jsonNode) {
 226         final var id = jsonNode.get("id");
 227         return id instanceof ValueNode value ? value : null;
 228     }
 229
 230     private JsonRpcRequest parseJsonRpcRequest(final JsonNode input) throws JsonRpcException {
 231         if (input instanceof ObjectNode i && input.has("params") && input.get("params").isNull()) {
 232             // Workaround for clients that send a null params field instead of omitting it
 233             i.remove("params");
 234         }
 235         JsonRpcRequest request;
 236         try {
 237             request = objectMapper.treeToValue(input, JsonRpcRequest.class);
 238         } catch (JsonMappingException e) {
 239             throw new JsonRpcException(new JsonRpcResponse.Error(JsonRpcResponse.Error.INVALID_REQUEST,
 240                     e.getMessage(),
 241                     null));
 242         } catch (IOException e) {
 243             throw new AssertionError(e);
 244         }
 245
 246         if (!"2.0".equals(request.getJsonrpc())) {
 247             throw new JsonRpcException(new JsonRpcResponse.Error(JsonRpcResponse.Error.INVALID_REQUEST,
 248                     "only jsonrpc version 2.0 is supported",
 249                     null));
 250         }
 251
 252         if (request.getMethod() == null) {
 253             throw new JsonRpcException(new JsonRpcResponse.Error(JsonRpcResponse.Error.INVALID_REQUEST,
 254                     "method field must be set",
 255                     null));
 256         }
 257
 258         return request;
 259     }
 260
 261     private JsonRpcResponse parseJsonRpcResponse(final JsonNode input) {
 262         JsonRpcResponse response;
 263         try {
 264             response = objectMapper.treeToValue(input, JsonRpcResponse.class);
 265         } catch (JsonParseException | JsonMappingException e) {
 266             logger.debug("Received invalid jsonrpc response {}", e.getMessage());
 267             return null;
 268         } catch (IOException e) {
 269             throw new AssertionError(e);
 270         }
 271
 272         if (!"2.0".equals(response.getJsonrpc())) {
 273             logger.debug("Received invalid jsonrpc response with invalid version {}", response.getJsonrpc());
 274             return null;
 275         }
 276
 277         if (response.getResult() != null && response.getError() != null) {
 278             logger.debug("Received invalid jsonrpc response with both result and error");
 279             return null;
 280         }
 281
 282         if (response.getResult() == null && response.getError() == null) {
 283             logger.debug("Received invalid jsonrpc response without result and error");
 284             return null;
 285         }
 286
 287         if (response.getId() == null || response.getId().isNull()) {
 288             logger.debug("Received invalid jsonrpc response without id");
 289             return null;
 290         }
 291
 292         return response;
 293     }
 294
 295     public interface RequestHandler {
 296
 297         JsonNode apply(String method, ContainerNode<?> params) throws JsonRpcException;
 298     }
 299 }