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