xmpp: move to library project

1 files changed, 432 insertions, 0 deletions

diff --git a/juick-server-xmpp/src/main/java/com/juick/server/XMPPServer.java b/juick-server-xmpp/src/main/java/com/juick/server/XMPPServer.java
new file mode 100644
index 00000000..c291202d
--- /dev/null
+++ b/juick-server-xmpp/src/main/java/com/juick/server/XMPPServer.java
@@ -0,0 +1,432 @@
+/*
+ * Copyright (C) 2008-2017, Juick
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package com.juick.server;
+
+import com.juick.server.xmpp.s2s.*;
+import com.juick.service.UserService;
+import com.juick.server.xmpp.extensions.JuickMessage;
+import com.juick.xmpp.extensions.StreamError;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.xmlpull.v1.XmlPullParserException;
+import rocks.xmpp.addr.Jid;
+import rocks.xmpp.core.stanza.model.Stanza;
+import rocks.xmpp.util.XmppUtils;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import javax.inject.Inject;
+import javax.net.ssl.*;
+import javax.xml.bind.JAXBException;
+import javax.xml.bind.Unmarshaller;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.XMLStreamWriter;
+import java.io.*;
+import java.net.*;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.SecureRandom;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * @author ugnich
+ */
+@Component
+public class XMPPServer implements ConnectionListener, AutoCloseable {
+    private static final Logger logger = LoggerFactory.getLogger(XMPPServer.class);
+
+    private static final int TIMEOUT_MINUTES = 15;
+
+    @Inject
+    public ExecutorService service;
+    @Value("${hostname:localhost}")
+    private Jid jid;
+    @Value("${s2s_port:5269}")
+    private int s2sPort;
+    @Value("${keystore:juick.p12}")
+    public String keystore;
+    @Value("${keystore_password:secret}")
+    public String keystorePassword;
+    @Value("${broken_ssl_hosts:}")
+    public String[] brokenSSLhosts;
+    @Value("${banned_hosts:}")
+    public String[] bannedHosts;
+
+    private final List<ConnectionIn> inConnections = new CopyOnWriteArrayList<>();
+    private final Map<ConnectionOut, Optional<Socket>> outConnections = new ConcurrentHashMap<>();
+    private final List<CacheEntry> outCache = new CopyOnWriteArrayList<>();
+    private final List<StanzaListener> stanzaListeners = new CopyOnWriteArrayList<>();
+    private final AtomicBoolean closeFlag = new AtomicBoolean(false);
+
+    SSLContext sc;
+    private TrustManager[] trustAllCerts = new TrustManager[]{
+            new X509TrustManager() {
+                public void checkClientTrusted(java.security.cert.X509Certificate[] certs, String authType) {
+                }
+
+                public void checkServerTrusted(java.security.cert.X509Certificate[] certs, String authType) {
+                }
+                public java.security.cert.X509Certificate[] getAcceptedIssuers() {
+                    return null;
+                }
+            }
+    };
+    private boolean tlsConfigured = false;
+
+
+    private ServerSocket listener;
+
+    @Inject
+    private BasicXmppSession session;
+    @Inject
+    private UserService userService;
+
+    @PostConstruct
+    public void init() throws KeyStoreException {
+        closeFlag.set(false);
+        KeyStore ks = KeyStore.getInstance("PKCS12");
+        try (InputStream ksIs = new FileInputStream(keystore)) {
+            ks.load(ksIs, keystorePassword.toCharArray());
+            KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory
+                    .getDefaultAlgorithm());
+            kmf.init(ks, keystorePassword.toCharArray());
+            sc = SSLContext.getInstance("TLSv1.2");
+            sc.init(kmf.getKeyManagers(), trustAllCerts, new SecureRandom());
+            tlsConfigured = true;
+        } catch (Exception e) {
+            logger.warn("tls unavailable");
+        }
+        service.submit(() -> {
+            try {
+                listener = new ServerSocket(s2sPort);
+                logger.info("s2s listener ready");
+                while (!listener.isClosed()) {
+                    if (Thread.currentThread().isInterrupted()) break;
+                    Socket socket = listener.accept();
+                    ConnectionIn client = new ConnectionIn(this, socket);
+                    addConnectionIn(client);
+                    service.submit(client);
+                }
+            } catch (SocketException e) {
+                // shutdown
+            } catch (IOException | XmlPullParserException e) {
+                logger.warn("xmpp exception", e);
+            }
+        });
+    }
+
+    @Override
+    public void close() throws Exception {
+        if (listener != null && !listener.isClosed()) {
+            listener.close();
+        }
+        outConnections.forEach((c, s) -> {
+            c.logoff();
+            outConnections.remove(c);
+        });
+        inConnections.forEach(c -> {
+            c.closeConnection();
+            inConnections.remove(c);
+        });
+        service.shutdown();
+        logger.info("XMPP server destroyed");
+    }
+
+    public void addConnectionIn(ConnectionIn c) {
+        c.setListener(this);
+        inConnections.add(c);
+    }
+
+    public void addConnectionOut(ConnectionOut c, Optional<Socket> socket) {
+        c.setListener(this);
+        outConnections.put(c, socket);
+    }
+
+    public void removeConnectionIn(ConnectionIn c) {
+        inConnections.remove(c);
+    }
+
+    public void removeConnectionOut(ConnectionOut c) {
+        outConnections.remove(c);
+    }
+
+    public String getFromCache(Jid to) {
+        final String[] cache = new String[1];
+        outCache.stream().filter(c -> c.hostname != null && c.hostname.equals(to)).findFirst().ifPresent(c -> {
+            cache[0] = c.xml;
+            outCache.remove(c);
+        });
+        return cache[0];
+    }
+
+    public Optional<ConnectionOut> getConnectionOut(Jid hostname, boolean needReady) {
+        return outConnections.keySet().stream().filter(c -> c.to != null &&
+                c.to.equals(hostname) && (!needReady || c.streamReady)).findFirst();
+    }
+
+    public Optional<ConnectionIn> getConnectionIn(String streamID) {
+        return inConnections.stream().filter(c -> c.streamID != null && c.streamID.equals(streamID)).findFirst();
+    }
+
+    public void sendOut(Stanza s) {
+        try {
+            StringWriter stanzaWriter = new StringWriter();
+            XMLStreamWriter xmppStreamWriter = XmppUtils.createXmppStreamWriter(
+                    session.getConfiguration().getXmlOutputFactory().createXMLStreamWriter(stanzaWriter));
+            session.createMarshaller().marshal(s, xmppStreamWriter);
+            xmppStreamWriter.flush();
+            xmppStreamWriter.close();
+            String xml = stanzaWriter.toString();
+            logger.debug("s2s (out): {}", xml);
+            sendOut(Jid.of(s.getTo().getDomain()), xml);
+        } catch (XMLStreamException | JAXBException e1) {
+            logger.info("jaxb exception", e1);
+        }
+    }
+
+    public void sendOut(Jid hostname, String xml) {
+        boolean haveAnyConn = false;
+
+        ConnectionOut connOut = null;
+        for (ConnectionOut c : outConnections.keySet()) {
+            if (c.to != null && c.to.equals(hostname)) {
+                if (c.streamReady) {
+                    connOut = c;
+                    break;
+                } else {
+                    haveAnyConn = true;
+                    break;
+                }
+            }
+        }
+        if (connOut != null) {
+            connOut.send(xml);
+            return;
+        }
+
+        boolean haveCache = false;
+        for (CacheEntry c : outCache) {
+            if (c.hostname != null && c.hostname.equals(hostname)) {
+                c.xml += xml;
+                c.updated = Instant.now();
+                haveCache = true;
+                break;
+            }
+        }
+        if (!haveCache) {
+            outCache.add(new CacheEntry(hostname, xml));
+        }
+
+        if (!haveAnyConn && !closeFlag.get()) {
+            try {
+                createDialbackConnection(hostname.toEscapedString(), null, null);
+            } catch (Exception e) {
+                logger.warn("dialback error", e);
+            }
+        }
+    }
+
+    void createDialbackConnection(String to, String checkSID, String dbKey) throws Exception {
+        ConnectionOut connectionOut = new ConnectionOut(getJid(), Jid.of(to), null, null, checkSID, dbKey);
+        addConnectionOut(connectionOut, Optional.empty());
+        service.submit(() -> {
+            try {
+                Socket socket = new Socket();
+                socket.connect(DNSQueries.getServerAddress(to));
+                connectionOut.setInputStream(socket.getInputStream());
+                connectionOut.setOutputStream(socket.getOutputStream());
+                addConnectionOut(connectionOut, Optional.of(socket));
+                connectionOut.addChildParser(new JuickMessage());
+                connectionOut.connect();
+            } catch (IOException e) {
+                userService.getActiveJIDs().stream().filter(j -> Jid.of(j).getDomain().equals(to))
+                        .forEach(j -> {
+                            userService.setActiveStatusForJID(j, UserService.ActiveStatus.Inactive);
+                            logger.info("{} is inactive now", j);
+                        });
+            }
+        });
+    }
+
+    public void startDialback(Jid from, String streamId, String dbKey) throws Exception {
+        Optional<ConnectionOut> c = getConnectionOut(from, false);
+        if (c.isPresent()) {
+            c.get().sendDialbackVerify(streamId, dbKey);
+        } else {
+            createDialbackConnection(from.toEscapedString(), streamId, dbKey);
+        }
+    }
+
+    public void addStanzaListener(StanzaListener listener) {
+        stanzaListeners.add(listener);
+    }
+
+    public void onStanzaReceived(String xmlValue) {
+        Stanza stanza = parse(xmlValue);
+        stanzaListeners.forEach(l -> l.stanzaReceived(stanza));
+    }
+
+    public BasicXmppSession getSession() {
+        return session;
+    }
+
+    public List<ConnectionIn> getInConnections() {
+        return inConnections;
+    }
+
+    public Map<ConnectionOut, Optional<Socket>> getOutConnections() {
+        return outConnections;
+    }
+
+    @Override
+    public boolean isTlsAvailable() {
+        return tlsConfigured;
+    }
+
+    @Override
+    public void starttls(ConnectionIn connection) {
+        logger.debug("stream {} securing", connection.streamID);
+        connection.sendStanza("<proceed xmlns=\"" + Connection.NS_TLS + "\" />");
+        try {
+            connection.setSocket(sc.getSocketFactory().createSocket(connection.getSocket(), connection.getSocket().getInetAddress().getHostAddress(),
+                    connection.getSocket().getPort(), true));
+            ((SSLSocket) connection.getSocket()).setUseClientMode(false);
+            ((SSLSocket) connection.getSocket()).startHandshake();
+            connection.setSecured(true);
+            logger.debug("stream {} secured", connection.streamID);
+            connection.restartParser();
+        } catch (XmlPullParserException | IOException sex) {
+            logger.warn("stream {} ssl error {}", connection.streamID, sex);
+            connection.sendStanza("<failed xmlns\"" + Connection.NS_TLS + "\" />");
+            removeConnectionIn(connection);
+            connection.closeConnection();
+        }
+    }
+
+    @Override
+    public void proceed(ConnectionOut connection) {
+        try {
+            Socket socket = outConnections.get(connection).get();
+            socket = sc.getSocketFactory().createSocket(socket, socket.getInetAddress().getHostAddress(),
+                    socket.getPort(), true);
+            ((SSLSocket) socket).startHandshake();
+            connection.setSecured(true);
+            logger.debug("stream {} secured", connection.getStreamID());
+            connection.setInputStream(socket.getInputStream());
+            connection.setOutputStream(socket.getOutputStream());
+            connection.restartStream();
+            connection.sendOpenStream();
+        } catch (NoSuchElementException | XmlPullParserException | IOException sex) {
+            logger.error("s2s ssl error: {} {}, error {}", connection.to, connection.getStreamID(), sex);
+            connection.send("<failed xmlns\"" + Connection.NS_TLS + "\" />");
+            removeConnectionOut(connection);
+            connection.logoff();
+        }
+    }
+
+    @Override
+    public void verify(ConnectionOut connection, String from, String type, String sid) {
+        if (from != null && from.equals(connection.to.toEscapedString()) && sid != null && !sid.isEmpty() && type != null) {
+            getConnectionIn(sid).ifPresent(c -> c.sendDialbackResult(Jid.of(from), type));
+        }
+    }
+
+    @Override
+    public void dialbackError(ConnectionOut connection, StreamError error) {
+        logger.warn("Stream error from {}: {}", connection.getStreamID(), error.getCondition());
+        removeConnectionOut(connection);
+        connection.logoff();
+    }
+
+    @Override
+    public void finished(ConnectionOut connection, boolean dirty) {
+        logger.warn("stream to {} {} finished, dirty={}", connection.to, connection.getStreamID(), dirty);
+        removeConnectionOut(connection);
+        connection.logoff();
+    }
+
+    @Override
+    public void exception(ConnectionOut connection, Exception ex) {
+        logger.error("s2s out exception: {} {}, exception {}", connection.to, connection.getStreamID(), ex);
+        removeConnectionOut(connection);
+        connection.logoff();
+    }
+
+    @Override
+    public void ready(ConnectionOut connection) {
+        logger.debug("stream to {} {} ready", connection.to, connection.getStreamID());
+        String cache = getFromCache(connection.to);
+        if (cache != null) {
+            logger.debug("stream to {} {} sending cache", connection.to, connection.getStreamID());
+            connection.send(cache);
+        }
+    }
+
+    @Override
+    public boolean securing(ConnectionOut connection) {
+        return !Arrays.asList(brokenSSLhosts).contains(connection.to.toEscapedString());
+    }
+
+    public Stanza parse(String xml) {
+        try {
+            Unmarshaller unmarshaller = session.createUnmarshaller();
+            return (Stanza)unmarshaller.unmarshal(new StringReader(xml));
+        } catch (JAXBException e) {
+            logger.error("JAXB exception", e);
+        }
+        return null;
+    }
+
+    public Jid getJid() {
+        return jid;
+    }
+    @Scheduled(fixedDelay = 10000)
+    public void cleanUp() {
+        Instant now = Instant.now();
+        outConnections.keySet().stream().filter(c -> Duration.between(now, c.getUpdated()).toMinutes() > TIMEOUT_MINUTES)
+                .forEach(c -> {
+                    logger.info("closing idle outgoing connection to {}", c.to);
+                    c.logoff();
+                    outConnections.remove(c);
+                });
+
+        inConnections.stream().filter(c -> Duration.between(now, c.updated).toMinutes() > TIMEOUT_MINUTES)
+                .forEach(c -> {
+                    logger.info("closing idle incoming connection from {}", c.from);
+                    c.closeConnection();
+                    inConnections.remove(c);
+                });
+    }
+    @PreDestroy
+    public void preDestroy() {
+        closeFlag.set(true);
+    }
+}