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