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