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