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     }
  83
  84     @Override
  85     public List<OutputType> getSupportedOutputTypes() {
  86         return List.of(OutputType.PLAIN_TEXT, OutputType.JSON);
  87     }
  88
  89     @Override
  90     public void handleCommand(
  91             final Namespace ns, final Manager m, final OutputWriter outputWriter
  92     ) throws CommandException {
  93         logger.info("Starting daemon in single-account mode for " + m.getSelfNumber());
  94         final var noReceiveStdOut = Boolean.TRUE.equals(ns.getBoolean("no-receive-stdout"));
  95         final var receiveMode = ns.<ReceiveMode>get("receive-mode");
  96         final var ignoreAttachments = Boolean.TRUE.equals(ns.getBoolean("ignore-attachments"));
  97
  98         m.setReceiveConfig(new ReceiveConfig(ignoreAttachments));
  99         addDefaultReceiveHandler(m, noReceiveStdOut ? null : outputWriter, receiveMode != ReceiveMode.ON_START);
 100
 101         final Channel inheritedChannel;
 102         try {
 103             inheritedChannel = System.inheritedChannel();
 104             if (inheritedChannel instanceof ServerSocketChannel serverChannel) {
 105                 logger.info("Using inherited socket: " + serverChannel.getLocalAddress());
 106                 runSocketSingleAccount(m, serverChannel, receiveMode == ReceiveMode.MANUAL);
 107             }
 108         } catch (IOException e) {
 109             throw new IOErrorException("Failed to use inherited socket", e);
 110         }
 111         final var socketFile = ns.<File>get("socket");
 112         if (socketFile != null) {
 113             final var address = UnixDomainSocketAddress.of(socketFile.toPath());
 114             final var serverChannel = IOUtils.bindSocket(address);
 115             runSocketSingleAccount(m, serverChannel, receiveMode == ReceiveMode.MANUAL);
 116         }
 117         final var tcpAddress = ns.getString("tcp");
 118         if (tcpAddress != null) {
 119             final var address = IOUtils.parseInetSocketAddress(tcpAddress);
 120             final var serverChannel = IOUtils.bindSocket(address);
 121             runSocketSingleAccount(m, serverChannel, receiveMode == ReceiveMode.MANUAL);
 122         }
 123         final var isDbusSystem = Boolean.TRUE.equals(ns.getBoolean("dbus-system"));
 124         if (isDbusSystem) {
 125             runDbusSingleAccount(m, true, receiveMode != ReceiveMode.ON_START);
 126         }
 127         final var isDbusSession = Boolean.TRUE.equals(ns.getBoolean("dbus"));
 128         if (isDbusSession || (
 129                 !isDbusSystem
 130                         && socketFile == null
 131                         && tcpAddress == null
 132                         && !(inheritedChannel instanceof ServerSocketChannel)
 133         )) {
 134             runDbusSingleAccount(m, false, receiveMode != ReceiveMode.ON_START);
 135         }
 136
 137         m.addClosedListener(() -> {
 138             synchronized (this) {
 139                 notifyAll();
 140             }
 141         });
 142
 143         synchronized (this) {
 144             try {
 145                 wait();
 146             } catch (InterruptedException ignored) {
 147             }
 148         }
 149     }
 150
 151     @Override
 152     public void handleCommand(
 153             final Namespace ns, final MultiAccountManager c, final OutputWriter outputWriter
 154     ) throws CommandException {
 155         logger.info("Starting daemon in multi-account mode");
 156         final var noReceiveStdOut = Boolean.TRUE.equals(ns.getBoolean("no-receive-stdout"));
 157         final var receiveMode = ns.<ReceiveMode>get("receive-mode");
 158         final var ignoreAttachments = Boolean.TRUE.equals(ns.getBoolean("ignore-attachments"));
 159
 160         final var receiveConfig = new ReceiveConfig(ignoreAttachments);
 161         c.getManagers().forEach(m -> {
 162             m.setReceiveConfig(receiveConfig);
 163             addDefaultReceiveHandler(m, noReceiveStdOut ? null : outputWriter, receiveMode != ReceiveMode.ON_START);
 164         });
 165         c.addOnManagerAddedHandler(m -> {
 166             m.setReceiveConfig(receiveConfig);
 167             addDefaultReceiveHandler(m, noReceiveStdOut ? null : outputWriter, receiveMode != ReceiveMode.ON_START);
 168         });
 169
 170         final Channel inheritedChannel;
 171         try {
 172             inheritedChannel = System.inheritedChannel();
 173             if (inheritedChannel instanceof ServerSocketChannel serverChannel) {
 174                 logger.info("Using inherited socket: " + serverChannel.getLocalAddress());
 175                 runSocketMultiAccount(c, serverChannel, receiveMode == ReceiveMode.MANUAL);
 176             }
 177         } catch (IOException e) {
 178             throw new IOErrorException("Failed to use inherited socket", e);
 179         }
 180         final var socketFile = ns.<File>get("socket");
 181         if (socketFile != null) {
 182             final var address = UnixDomainSocketAddress.of(socketFile.toPath());
 183             final var serverChannel = IOUtils.bindSocket(address);
 184             runSocketMultiAccount(c, serverChannel, receiveMode == ReceiveMode.MANUAL);
 185         }
 186         final var tcpAddress = ns.getString("tcp");
 187         if (tcpAddress != null) {
 188             final var address = IOUtils.parseInetSocketAddress(tcpAddress);
 189             final var serverChannel = IOUtils.bindSocket(address);
 190             runSocketMultiAccount(c, serverChannel, receiveMode == ReceiveMode.MANUAL);
 191         }
 192         final var isDbusSystem = Boolean.TRUE.equals(ns.getBoolean("dbus-system"));
 193         if (isDbusSystem) {
 194             runDbusMultiAccount(c, receiveMode != ReceiveMode.ON_START, true);
 195         }
 196         final var isDbusSession = Boolean.TRUE.equals(ns.getBoolean("dbus"));
 197         if (isDbusSession || (
 198                 !isDbusSystem
 199                         && socketFile == null
 200                         && tcpAddress == null
 201                         && !(inheritedChannel instanceof ServerSocketChannel)
 202         )) {
 203             runDbusMultiAccount(c, receiveMode != ReceiveMode.ON_START, false);
 204         }
 205
 206         synchronized (this) {
 207             try {
 208                 wait();
 209             } catch (InterruptedException ignored) {
 210             }
 211         }
 212     }
 213
 214     private void addDefaultReceiveHandler(Manager m, OutputWriter outputWriter, final boolean isWeakListener) {
 215         final var handler = outputWriter instanceof JsonWriter o
 216                 ? new JsonReceiveMessageHandler(m, o)
 217                 : outputWriter instanceof PlainTextWriter o
 218                         ? new ReceiveMessageHandler(m, o)
 219                         : Manager.ReceiveMessageHandler.EMPTY;
 220         m.addReceiveHandler(handler, isWeakListener);
 221     }
 222
 223     private void runSocketSingleAccount(
 224             final Manager m, final ServerSocketChannel serverChannel, final boolean noReceiveOnStart
 225     ) {
 226         runSocket(serverChannel, channel -> {
 227             final var handler = getSignalJsonRpcDispatcherHandler(channel, noReceiveOnStart);
 228             handler.handleConnection(m);
 229         });
 230     }
 231
 232     private void runSocketMultiAccount(
 233             final MultiAccountManager c, final ServerSocketChannel serverChannel, final boolean noReceiveOnStart
 234     ) {
 235         runSocket(serverChannel, channel -> {
 236             final var handler = getSignalJsonRpcDispatcherHandler(channel, noReceiveOnStart);
 237             handler.handleConnection(c);
 238         });
 239     }
 240
 241     private static final AtomicInteger threadNumber = new AtomicInteger(0);
 242
 243     private void runSocket(final ServerSocketChannel serverChannel, Consumer<SocketChannel> socketHandler) {
 244         final var thread = new Thread(() -> {
 245             while (true) {
 246                 final var connectionId = threadNumber.getAndIncrement();
 247                 final SocketChannel channel;
 248                 final String clientString;
 249                 try {
 250                     channel = serverChannel.accept();
 251                     clientString = channel.getRemoteAddress() + " " + IOUtils.getUnixDomainPrincipal(channel);
 252                     logger.info("Accepted new client connection {}: {}", connectionId, clientString);
 253                 } catch (IOException e) {
 254                     logger.error("Failed to accept new socket connection", e);
 255                     synchronized (this) {
 256                         notifyAll();
 257                     }
 258                     break;
 259                 }
 260                 final var connectionThread = new Thread(() -> {
 261                     try (final var c = channel) {
 262                         socketHandler.accept(c);
 263                     } catch (IOException e) {
 264                         logger.warn("Failed to close channel", e);
 265                     } catch (Throwable e) {
 266                         logger.warn("Connection handler failed, closing connection", e);
 267                     }
 268                     logger.info("Connection {} closed: {}", connectionId, clientString);
 269                 });
 270                 connectionThread.setName("daemon-connection-" + connectionId);
 271                 connectionThread.start();
 272             }
 273         });
 274         thread.setName("daemon-listener");
 275         thread.start();
 276     }
 277
 278     private SignalJsonRpcDispatcherHandler getSignalJsonRpcDispatcherHandler(
 279             final SocketChannel c, final boolean noReceiveOnStart
 280     ) {
 281         final var lineSupplier = IOUtils.getLineSupplier(Channels.newReader(c, StandardCharsets.UTF_8));
 282         final var jsonOutputWriter = new JsonWriterImpl(Channels.newWriter(c, StandardCharsets.UTF_8));
 283
 284         return new SignalJsonRpcDispatcherHandler(jsonOutputWriter, lineSupplier, noReceiveOnStart);
 285     }
 286
 287     private void runDbusSingleAccount(
 288             final Manager m, final boolean isDbusSystem, final boolean noReceiveOnStart
 289     ) throws CommandException {
 290         runDbus(isDbusSystem, (conn, objectPath) -> {
 291             try {
 292                 exportDbusObject(conn, objectPath, m, noReceiveOnStart).join();
 293             } catch (InterruptedException ignored) {
 294             }
 295         });
 296     }
 297
 298     private void runDbusMultiAccount(
 299             final MultiAccountManager c, final boolean noReceiveOnStart, final boolean isDbusSystem
 300     ) throws CommandException {
 301         runDbus(isDbusSystem, (connection, objectPath) -> {
 302             final var signalControl = new DbusSignalControlImpl(c, objectPath);
 303             connection.exportObject(signalControl);
 304
 305             c.addOnManagerAddedHandler(m -> {
 306                 final var thread = exportMultiAccountManager(connection, m, noReceiveOnStart);
 307                 try {
 308                     thread.join();
 309                 } catch (InterruptedException ignored) {
 310                 }
 311             });
 312             c.addOnManagerRemovedHandler(m -> {
 313                 final var path = DbusConfig.getObjectPath(m.getSelfNumber());
 314                 try {
 315                     final var object = connection.getExportedObject(null, path);
 316                     if (object instanceof DbusSignalImpl dbusSignal) {
 317                         dbusSignal.close();
 318                     }
 319                 } catch (DBusException ignored) {
 320                 }
 321             });
 322
 323             final var initThreads = c.getManagers()
 324                     .stream()
 325                     .map(m -> exportMultiAccountManager(connection, m, noReceiveOnStart))
 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 CommandException {
 340         DBusConnection.DBusBusType busType;
 341         if (isDbusSystem) {
 342             busType = DBusConnection.DBusBusType.SYSTEM;
 343         } else {
 344             busType = DBusConnection.DBusBusType.SESSION;
 345         }
 346         DBusConnection conn;
 347         try {
 348             conn = DBusConnectionBuilder.forType(busType).build();
 349             dbusRunner.run(conn, DbusConfig.getObjectPath());
 350         } catch (DBusException e) {
 351             throw new UnexpectedErrorException("Dbus command failed: " + e.getMessage(), e);
 352         } catch (UnsupportedOperationException e) {
 353             throw new UserErrorException("Failed to connect to Dbus: " + e.getMessage(), e);
 354         }
 355
 356         try {
 357             conn.requestBusName(DbusConfig.getBusname());
 358         } catch (DBusException e) {
 359             throw new UnexpectedErrorException("Dbus command failed, maybe signal-cli dbus daemon is already running: "
 360                     + e.getMessage(), e);
 361         }
 362
 363         logger.info("DBus daemon running on {} bus: {}", busType, DbusConfig.getBusname());
 364     }
 365
 366     private Thread exportMultiAccountManager(
 367             final DBusConnection conn, final Manager m, final boolean noReceiveOnStart
 368     ) {
 369         final var objectPath = DbusConfig.getObjectPath(m.getSelfNumber());
 370         return exportDbusObject(conn, objectPath, m, noReceiveOnStart);
 371     }
 372
 373     private Thread exportDbusObject(
 374             final DBusConnection conn, final String objectPath, final Manager m, final boolean noReceiveOnStart
 375     ) {
 376         final var signal = new DbusSignalImpl(m, conn, objectPath, noReceiveOnStart);
 377         final var initThread = new Thread(signal::initObjects);
 378         initThread.setName("dbus-init");
 379         initThread.start();
 380
 381         return initThread;
 382     }
 383
 384     interface DbusRunner {
 385
 386         void run(DBusConnection connection, String objectPath) throws DBusException;
 387     }
 388 }