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

   1 package org.asamk.signal.commands;
   2
   3 import net.sourceforge.argparse4j.impl.Arguments;
   4 import net.sourceforge.argparse4j.inf.Namespace;
   5 import net.sourceforge.argparse4j.inf.Subparser;
   6
   7 import org.asamk.signal.DbusConfig;
   8 import org.asamk.signal.OutputType;
   9 import org.asamk.signal.ReceiveMessageHandler;
  10 import org.asamk.signal.commands.exceptions.CommandException;
  11 import org.asamk.signal.commands.exceptions.IOErrorException;
  12 import org.asamk.signal.commands.exceptions.UnexpectedErrorException;
  13 import org.asamk.signal.commands.exceptions.UserErrorException;
  14 import org.asamk.signal.dbus.DbusSignalControlImpl;
  15 import org.asamk.signal.dbus.DbusSignalImpl;
  16 import org.asamk.signal.http.HttpServerHandler;
  17 import org.asamk.signal.json.JsonReceiveMessageHandler;
  18 import org.asamk.signal.jsonrpc.SignalJsonRpcDispatcherHandler;
  19 import org.asamk.signal.manager.Manager;
  20 import org.asamk.signal.manager.MultiAccountManager;
  21 import org.asamk.signal.manager.api.ReceiveConfig;
  22 import org.asamk.signal.output.JsonWriter;
  23 import org.asamk.signal.output.JsonWriterImpl;
  24 import org.asamk.signal.output.OutputWriter;
  25 import org.asamk.signal.output.PlainTextWriter;
  26 import org.asamk.signal.util.IOUtils;
  27 import org.freedesktop.dbus.connections.impl.DBusConnection;
  28 import org.freedesktop.dbus.connections.impl.DBusConnectionBuilder;
  29 import org.freedesktop.dbus.exceptions.DBusException;
  30 import org.slf4j.Logger;
  31 import org.slf4j.LoggerFactory;
  32
  33 import java.io.File;
  34 import java.io.IOException;
  35 import java.net.UnixDomainSocketAddress;
  36 import java.nio.channels.Channel;
  37 import java.nio.channels.Channels;
  38 import java.nio.channels.ServerSocketChannel;
  39 import java.nio.channels.SocketChannel;
  40 import java.nio.charset.StandardCharsets;
  41 import java.util.List;
  42 import java.util.concurrent.atomic.AtomicInteger;
  43 import java.util.function.Consumer;
  44
  45 public class DaemonCommand implements MultiLocalCommand, LocalCommand {
  46
  47     private final static Logger logger = LoggerFactory.getLogger(DaemonCommand.class);
  48
  49     @Override
  50     public String getName() {
  51         return "daemon";
  52     }
  53
  54     @Override
  55     public void attachToSubparser(final Subparser subparser) {
  56         final var defaultSocketPath = new File(new File(IOUtils.getRuntimeDir(), "signal-cli"), "socket");
  57         subparser.help("Run in daemon mode and provide an experimental dbus or JSON-RPC interface.");
  58         subparser.addArgument("--dbus")
  59                 .action(Arguments.storeTrue())
  60                 .help("Expose a DBus interface on the user bus (the default, if no other options are given).");
  61         subparser.addArgument("--dbus-system", "--system")
  62                 .action(Arguments.storeTrue())
  63                 .help("Expose a DBus interface on the system bus.");
  64         subparser.addArgument("--socket")
  65                 .nargs("?")
  66                 .type(File.class)
  67                 .setConst(defaultSocketPath)
  68                 .help("Expose a JSON-RPC interface on a UNIX socket (default $XDG_RUNTIME_DIR/signal-cli/socket).");
  69         subparser.addArgument("--tcp")
  70                 .nargs("?")
  71                 .setConst("localhost:7583")
  72                 .help("Expose a JSON-RPC interface on a TCP socket (default localhost:7583).");
  73         subparser.addArgument("--http")
  74                 .nargs("?")
  75                 .setConst("localhost:8080")
  76                 .help("Expose a JSON-RPC interface as http endpoint (default localhost:8080).");
  77         subparser.addArgument("--no-receive-stdout")
  78                 .help("Don’t print received messages to stdout.")
  79                 .action(Arguments.storeTrue());
  80         subparser.addArgument("--receive-mode")
  81                 .help("Specify when to start receiving messages.")
  82                 .type(Arguments.enumStringType(ReceiveMode.class))
  83                 .setDefault(ReceiveMode.ON_START);
  84         subparser.addArgument("--ignore-attachments")
  85                 .help("Don’t download attachments of received messages.")
  86                 .action(Arguments.storeTrue());
  87         subparser.addArgument("--ignore-stories")
  88                 .help("Don’t receive story messages from the server.")
  89                 .action(Arguments.storeTrue());
  90         subparser.addArgument("--send-read-receipts")
  91                 .help("Send read receipts for all incoming data messages (in addition to the default delivery receipts)")
  92                 .action(Arguments.storeTrue());
  93     }
  94
  95     @Override
  96     public List<OutputType> getSupportedOutputTypes() {
  97         return List.of(OutputType.PLAIN_TEXT, OutputType.JSON);
  98     }
  99
 100     @Override
 101     public void handleCommand(
 102             final Namespace ns, final Manager m, final OutputWriter outputWriter
 103     ) throws CommandException {
 104         logger.info("Starting daemon in single-account mode for " + m.getSelfNumber());
 105         final var noReceiveStdOut = Boolean.TRUE.equals(ns.getBoolean("no-receive-stdout"));
 106         final var receiveMode = ns.<ReceiveMode>get("receive-mode");
 107         final var ignoreAttachments = Boolean.TRUE.equals(ns.getBoolean("ignore-attachments"));
 108         final var ignoreStories = Boolean.TRUE.equals(ns.getBoolean("ignore-stories"));
 109         final var sendReadReceipts = Boolean.TRUE.equals(ns.getBoolean("send-read-receipts"));
 110
 111         m.setReceiveConfig(new ReceiveConfig(ignoreAttachments, ignoreStories, sendReadReceipts));
 112         addDefaultReceiveHandler(m, noReceiveStdOut ? null : outputWriter, receiveMode != ReceiveMode.ON_START);
 113
 114         final Channel inheritedChannel;
 115         try {
 116             inheritedChannel = System.inheritedChannel();
 117             if (inheritedChannel instanceof ServerSocketChannel serverChannel) {
 118                 logger.info("Using inherited socket: " + serverChannel.getLocalAddress());
 119                 runSocketSingleAccount(m, serverChannel, receiveMode == ReceiveMode.MANUAL);
 120             }
 121         } catch (IOException e) {
 122             throw new IOErrorException("Failed to use inherited socket", e);
 123         }
 124         final var socketFile = ns.<File>get("socket");
 125         if (socketFile != null) {
 126             final var address = UnixDomainSocketAddress.of(socketFile.toPath());
 127             final var serverChannel = IOUtils.bindSocket(address);
 128             runSocketSingleAccount(m, serverChannel, receiveMode == ReceiveMode.MANUAL);
 129         }
 130         final var tcpAddress = ns.getString("tcp");
 131         if (tcpAddress != null) {
 132             final var address = IOUtils.parseInetSocketAddress(tcpAddress);
 133             final var serverChannel = IOUtils.bindSocket(address);
 134             runSocketSingleAccount(m, serverChannel, receiveMode == ReceiveMode.MANUAL);
 135         }
 136         final var httpAddress = ns.getString("http");
 137         if (httpAddress != null) {
 138             final var address = IOUtils.parseInetSocketAddress(httpAddress);
 139             final var handler = new HttpServerHandler(address, m);
 140             try {
 141                 handler.init();
 142             } catch (IOException ex) {
 143                 throw new IOErrorException("Failed to initialize HTTP Server", ex);
 144             }
 145         }
 146         final var isDbusSystem = Boolean.TRUE.equals(ns.getBoolean("dbus-system"));
 147         if (isDbusSystem) {
 148             runDbusSingleAccount(m, true, receiveMode != ReceiveMode.ON_START);
 149         }
 150         final var isDbusSession = Boolean.TRUE.equals(ns.getBoolean("dbus"));
 151         if (isDbusSession || (
 152                 !isDbusSystem
 153                         && socketFile == null
 154                         && tcpAddress == null
 155                         && httpAddress == null
 156                         && !(inheritedChannel instanceof ServerSocketChannel)
 157         )) {
 158             runDbusSingleAccount(m, false, receiveMode != ReceiveMode.ON_START);
 159         }
 160
 161         m.addClosedListener(() -> {
 162             synchronized (this) {
 163                 notifyAll();
 164             }
 165         });
 166
 167         synchronized (this) {
 168             try {
 169                 wait();
 170             } catch (InterruptedException ignored) {
 171             }
 172         }
 173     }
 174
 175     @Override
 176     public void handleCommand(
 177             final Namespace ns, final MultiAccountManager c, final OutputWriter outputWriter
 178     ) throws CommandException {
 179         logger.info("Starting daemon in multi-account mode");
 180         final var noReceiveStdOut = Boolean.TRUE.equals(ns.getBoolean("no-receive-stdout"));
 181         final var receiveMode = ns.<ReceiveMode>get("receive-mode");
 182         final var ignoreAttachments = Boolean.TRUE.equals(ns.getBoolean("ignore-attachments"));
 183         final var ignoreStories = Boolean.TRUE.equals(ns.getBoolean("ignore-stories"));
 184         final var sendReadReceipts = Boolean.TRUE.equals(ns.getBoolean("send-read-receipts"));
 185
 186         final var receiveConfig = new ReceiveConfig(ignoreAttachments, ignoreStories, sendReadReceipts);
 187         c.getManagers().forEach(m -> {
 188             m.setReceiveConfig(receiveConfig);
 189             addDefaultReceiveHandler(m, noReceiveStdOut ? null : outputWriter, receiveMode != ReceiveMode.ON_START);
 190         });
 191         c.addOnManagerAddedHandler(m -> {
 192             m.setReceiveConfig(receiveConfig);
 193             addDefaultReceiveHandler(m, noReceiveStdOut ? null : outputWriter, receiveMode != ReceiveMode.ON_START);
 194         });
 195
 196         final Channel inheritedChannel;
 197         try {
 198             inheritedChannel = System.inheritedChannel();
 199             if (inheritedChannel instanceof ServerSocketChannel serverChannel) {
 200                 logger.info("Using inherited socket: " + serverChannel.getLocalAddress());
 201                 runSocketMultiAccount(c, serverChannel, receiveMode == ReceiveMode.MANUAL);
 202             }
 203         } catch (IOException e) {
 204             throw new IOErrorException("Failed to use inherited socket", e);
 205         }
 206         final var socketFile = ns.<File>get("socket");
 207         if (socketFile != null) {
 208             final var address = UnixDomainSocketAddress.of(socketFile.toPath());
 209             final var serverChannel = IOUtils.bindSocket(address);
 210             runSocketMultiAccount(c, serverChannel, receiveMode == ReceiveMode.MANUAL);
 211         }
 212         final var tcpAddress = ns.getString("tcp");
 213         if (tcpAddress != null) {
 214             final var address = IOUtils.parseInetSocketAddress(tcpAddress);
 215             final var serverChannel = IOUtils.bindSocket(address);
 216             runSocketMultiAccount(c, serverChannel, receiveMode == ReceiveMode.MANUAL);
 217         }
 218         final var httpAddress = ns.getString("http");
 219         if (httpAddress != null) {
 220             final var address = IOUtils.parseInetSocketAddress(httpAddress);
 221             final var handler = new HttpServerHandler(address, c);
 222             try {
 223                 handler.init();
 224             } catch (IOException ex) {
 225                 throw new IOErrorException("Failed to initialize HTTP Server", ex);
 226             }
 227         }
 228         final var isDbusSystem = Boolean.TRUE.equals(ns.getBoolean("dbus-system"));
 229         if (isDbusSystem) {
 230             runDbusMultiAccount(c, receiveMode != ReceiveMode.ON_START, true);
 231         }
 232         final var isDbusSession = Boolean.TRUE.equals(ns.getBoolean("dbus"));
 233         if (isDbusSession || (
 234                 !isDbusSystem
 235                         && socketFile == null
 236                         && tcpAddress == null
 237                         && httpAddress == null
 238                         && !(inheritedChannel instanceof ServerSocketChannel)
 239         )) {
 240             runDbusMultiAccount(c, receiveMode != ReceiveMode.ON_START, false);
 241         }
 242
 243         synchronized (this) {
 244             try {
 245                 wait();
 246             } catch (InterruptedException ignored) {
 247             }
 248         }
 249     }
 250
 251     private void addDefaultReceiveHandler(Manager m, OutputWriter outputWriter, final boolean isWeakListener) {
 252         final var handler = outputWriter instanceof JsonWriter o
 253                 ? new JsonReceiveMessageHandler(m, o)
 254                 : outputWriter instanceof PlainTextWriter o
 255                         ? new ReceiveMessageHandler(m, o)
 256                         : Manager.ReceiveMessageHandler.EMPTY;
 257         m.addReceiveHandler(handler, isWeakListener);
 258     }
 259
 260     private void runSocketSingleAccount(
 261             final Manager m, final ServerSocketChannel serverChannel, final boolean noReceiveOnStart
 262     ) {
 263         runSocket(serverChannel, channel -> {
 264             final var handler = getSignalJsonRpcDispatcherHandler(channel, noReceiveOnStart);
 265             handler.handleConnection(m);
 266         });
 267     }
 268
 269     private void runSocketMultiAccount(
 270             final MultiAccountManager c, final ServerSocketChannel serverChannel, final boolean noReceiveOnStart
 271     ) {
 272         runSocket(serverChannel, channel -> {
 273             final var handler = getSignalJsonRpcDispatcherHandler(channel, noReceiveOnStart);
 274             handler.handleConnection(c);
 275         });
 276     }
 277
 278     private static final AtomicInteger threadNumber = new AtomicInteger(0);
 279
 280     private void runSocket(final ServerSocketChannel serverChannel, Consumer<SocketChannel> socketHandler) {
 281         final var thread = new Thread(() -> {
 282             while (true) {
 283                 final var connectionId = threadNumber.getAndIncrement();
 284                 final SocketChannel channel;
 285                 final String clientString;
 286                 try {
 287                     channel = serverChannel.accept();
 288                     clientString = channel.getRemoteAddress() + " " + IOUtils.getUnixDomainPrincipal(channel);
 289                     logger.info("Accepted new client connection {}: {}", connectionId, clientString);
 290                 } catch (IOException e) {
 291                     logger.error("Failed to accept new socket connection", e);
 292                     synchronized (this) {
 293                         notifyAll();
 294                     }
 295                     break;
 296                 }
 297                 final var connectionThread = new Thread(() -> {
 298                     try (final var c = channel) {
 299                         socketHandler.accept(c);
 300                     } catch (IOException e) {
 301                         logger.warn("Failed to close channel", e);
 302                     } catch (Throwable e) {
 303                         logger.warn("Connection handler failed, closing connection", e);
 304                     }
 305                     logger.info("Connection {} closed: {}", connectionId, clientString);
 306                 });
 307                 connectionThread.setName("daemon-connection-" + connectionId);
 308                 connectionThread.start();
 309             }
 310         });
 311         thread.setName("daemon-listener");
 312         thread.start();
 313     }
 314
 315     private SignalJsonRpcDispatcherHandler getSignalJsonRpcDispatcherHandler(
 316             final SocketChannel c, final boolean noReceiveOnStart
 317     ) {
 318         final var lineSupplier = IOUtils.getLineSupplier(Channels.newReader(c, StandardCharsets.UTF_8));
 319         final var jsonOutputWriter = new JsonWriterImpl(Channels.newWriter(c, StandardCharsets.UTF_8));
 320
 321         return new SignalJsonRpcDispatcherHandler(jsonOutputWriter, lineSupplier, noReceiveOnStart);
 322     }
 323
 324     private void runDbusSingleAccount(
 325             final Manager m, final boolean isDbusSystem, final boolean noReceiveOnStart
 326     ) throws CommandException {
 327         runDbus(isDbusSystem, (conn, objectPath) -> {
 328             try {
 329                 exportDbusObject(conn, objectPath, m, noReceiveOnStart).join();
 330             } catch (InterruptedException ignored) {
 331             }
 332         });
 333     }
 334
 335     private void runDbusMultiAccount(
 336             final MultiAccountManager c, final boolean noReceiveOnStart, final boolean isDbusSystem
 337     ) throws CommandException {
 338         runDbus(isDbusSystem, (connection, objectPath) -> {
 339             final var signalControl = new DbusSignalControlImpl(c, objectPath);
 340             connection.exportObject(signalControl);
 341
 342             c.addOnManagerAddedHandler(m -> {
 343                 final var thread = exportMultiAccountManager(connection, m, noReceiveOnStart);
 344                 try {
 345                     thread.join();
 346                 } catch (InterruptedException ignored) {
 347                 }
 348             });
 349             c.addOnManagerRemovedHandler(m -> {
 350                 final var path = DbusConfig.getObjectPath(m.getSelfNumber());
 351                 try {
 352                     final var object = connection.getExportedObject(null, path);
 353                     if (object instanceof DbusSignalImpl dbusSignal) {
 354                         dbusSignal.close();
 355                     }
 356                 } catch (DBusException ignored) {
 357                 }
 358             });
 359
 360             final var initThreads = c.getManagers()
 361                     .stream()
 362                     .map(m -> exportMultiAccountManager(connection, m, noReceiveOnStart))
 363                     .toList();
 364
 365             for (var t : initThreads) {
 366                 try {
 367                     t.join();
 368                 } catch (InterruptedException ignored) {
 369                 }
 370             }
 371         });
 372     }
 373
 374     private void runDbus(
 375             final boolean isDbusSystem, DbusRunner dbusRunner
 376     ) throws CommandException {
 377         DBusConnection.DBusBusType busType;
 378         if (isDbusSystem) {
 379             busType = DBusConnection.DBusBusType.SYSTEM;
 380         } else {
 381             busType = DBusConnection.DBusBusType.SESSION;
 382         }
 383         DBusConnection conn;
 384         try {
 385             conn = DBusConnectionBuilder.forType(busType).build();
 386             dbusRunner.run(conn, DbusConfig.getObjectPath());
 387         } catch (DBusException e) {
 388             throw new UnexpectedErrorException("Dbus command failed: " + e.getMessage(), e);
 389         } catch (UnsupportedOperationException e) {
 390             throw new UserErrorException("Failed to connect to Dbus: " + e.getMessage(), e);
 391         }
 392
 393         try {
 394             conn.requestBusName(DbusConfig.getBusname());
 395         } catch (DBusException e) {
 396             throw new UnexpectedErrorException("Dbus command failed, maybe signal-cli dbus daemon is already running: "
 397                     + e.getMessage(), e);
 398         }
 399
 400         logger.info("DBus daemon running on {} bus: {}", busType, DbusConfig.getBusname());
 401     }
 402
 403     private Thread exportMultiAccountManager(
 404             final DBusConnection conn, final Manager m, final boolean noReceiveOnStart
 405     ) {
 406         final var objectPath = DbusConfig.getObjectPath(m.getSelfNumber());
 407         return exportDbusObject(conn, objectPath, m, noReceiveOnStart);
 408     }
 409
 410     private Thread exportDbusObject(
 411             final DBusConnection conn, final String objectPath, final Manager m, final boolean noReceiveOnStart
 412     ) {
 413         final var signal = new DbusSignalImpl(m, conn, objectPath, noReceiveOnStart);
 414         final var initThread = new Thread(signal::initObjects);
 415         initThread.setName("dbus-init");
 416         initThread.start();
 417
 418         return initThread;
 419     }
 420
 421     interface DbusRunner {
 422
 423         void run(DBusConnection connection, String objectPath) throws DBusException;
 424     }
 425 }