From a201bab42facd56290195703f6847e1a29033fb4 Mon Sep 17 00:00:00 2001 From: Vitaly Takmazov Date: Sun, 30 Dec 2018 23:05:16 +0300 Subject: Name XMPP threads --- src/main/java/com/juick/server/xmpp/s2s/ConnectionIn.java | 1 + src/main/java/com/juick/server/xmpp/s2s/ConnectionOut.java | 1 + 2 files changed, 2 insertions(+) (limited to 'src/main/java/com/juick/server/xmpp/s2s') diff --git a/src/main/java/com/juick/server/xmpp/s2s/ConnectionIn.java b/src/main/java/com/juick/server/xmpp/s2s/ConnectionIn.java index 72c3ba8d..3929a69f 100644 --- a/src/main/java/com/juick/server/xmpp/s2s/ConnectionIn.java +++ b/src/main/java/com/juick/server/xmpp/s2s/ConnectionIn.java @@ -69,6 +69,7 @@ public class ConnectionIn extends Connection implements Runnable { } boolean xmppversionnew = parser.getAttributeValue(null, "version") != null; String from = parser.getAttributeValue(null, "from"); + Thread.currentThread().setName(String.format("XMPP S2S IN %s - %d", from, Thread.currentThread().getId())); if (Arrays.asList(xmpp.bannedHosts).contains(from)) { closeConnection(); diff --git a/src/main/java/com/juick/server/xmpp/s2s/ConnectionOut.java b/src/main/java/com/juick/server/xmpp/s2s/ConnectionOut.java index be485ab1..2b919da4 100644 --- a/src/main/java/com/juick/server/xmpp/s2s/ConnectionOut.java +++ b/src/main/java/com/juick/server/xmpp/s2s/ConnectionOut.java @@ -65,6 +65,7 @@ public class ConnectionOut extends Stream { this.dbKey = DialbackUtils.generateDialbackKey(generator.generate(15), to, from, streamID); } streamID = UUID.randomUUID().toString(); + Thread.currentThread().setName(String.format("XMPP S2S OUT %s - %d", to.toEscapedString(), Thread.currentThread().getId())); } public void sendOpenStream() throws IOException { -- cgit v1.2.3 From 3af1fa43a25ba844b999d0be6c1139e6a98046fb Mon Sep 17 00:00:00 2001 From: Vitaly Takmazov Date: Mon, 31 Dec 2018 00:02:02 +0300 Subject: Use external XMPP server --- src/main/java/com/juick/server/XMPPConnection.java | 19 +- src/main/java/com/juick/server/XMPPServer.java | 429 --------------------- .../com/juick/server/configuration/XMPPConfig.java | 11 - .../java/com/juick/server/xmpp/XMPPStatusPage.java | 32 -- .../com/juick/server/xmpp/helpers/XMPPStatus.java | 48 --- .../com/juick/server/xmpp/router/Handshake.java | 39 -- .../java/com/juick/server/xmpp/router/Stream.java | 202 ---------- .../server/xmpp/router/StreamComponentServer.java | 57 --- .../com/juick/server/xmpp/router/StreamError.java | 57 --- .../juick/server/xmpp/router/StreamFeatures.java | 95 ----- .../juick/server/xmpp/router/StreamHandler.java | 13 - .../juick/server/xmpp/router/StreamNamespaces.java | 10 - .../com/juick/server/xmpp/router/XMPPError.java | 73 ---- .../com/juick/server/xmpp/router/XMPPRouter.java | 220 ----------- .../com/juick/server/xmpp/router/XmlUtils.java | 88 ----- .../java/com/juick/server/xmpp/s2s/CacheEntry.java | 40 -- .../java/com/juick/server/xmpp/s2s/Connection.java | 158 -------- .../com/juick/server/xmpp/s2s/ConnectionIn.java | 232 ----------- .../juick/server/xmpp/s2s/ConnectionListener.java | 16 - .../com/juick/server/xmpp/s2s/ConnectionOut.java | 190 --------- .../java/com/juick/server/xmpp/s2s/DNSQueries.java | 65 ---- .../juick/server/xmpp/s2s/util/DialbackUtils.java | 37 -- .../java/com/juick/server/tests/ServerTests.java | 81 ---- 23 files changed, 6 insertions(+), 2206 deletions(-) delete mode 100644 src/main/java/com/juick/server/XMPPServer.java delete mode 100644 src/main/java/com/juick/server/xmpp/XMPPStatusPage.java delete mode 100644 src/main/java/com/juick/server/xmpp/helpers/XMPPStatus.java delete mode 100644 src/main/java/com/juick/server/xmpp/router/Handshake.java delete mode 100644 src/main/java/com/juick/server/xmpp/router/Stream.java delete mode 100644 src/main/java/com/juick/server/xmpp/router/StreamComponentServer.java delete mode 100644 src/main/java/com/juick/server/xmpp/router/StreamError.java delete mode 100644 src/main/java/com/juick/server/xmpp/router/StreamFeatures.java delete mode 100644 src/main/java/com/juick/server/xmpp/router/StreamHandler.java delete mode 100644 src/main/java/com/juick/server/xmpp/router/StreamNamespaces.java delete mode 100644 src/main/java/com/juick/server/xmpp/router/XMPPError.java delete mode 100644 src/main/java/com/juick/server/xmpp/router/XMPPRouter.java delete mode 100644 src/main/java/com/juick/server/xmpp/router/XmlUtils.java delete mode 100644 src/main/java/com/juick/server/xmpp/s2s/CacheEntry.java delete mode 100644 src/main/java/com/juick/server/xmpp/s2s/Connection.java delete mode 100644 src/main/java/com/juick/server/xmpp/s2s/ConnectionIn.java delete mode 100644 src/main/java/com/juick/server/xmpp/s2s/ConnectionListener.java delete mode 100644 src/main/java/com/juick/server/xmpp/s2s/ConnectionOut.java delete mode 100644 src/main/java/com/juick/server/xmpp/s2s/DNSQueries.java delete mode 100644 src/main/java/com/juick/server/xmpp/s2s/util/DialbackUtils.java (limited to 'src/main/java/com/juick/server/xmpp/s2s') diff --git a/src/main/java/com/juick/server/XMPPConnection.java b/src/main/java/com/juick/server/XMPPConnection.java index dfbe92d5..c4816478 100644 --- a/src/main/java/com/juick/server/XMPPConnection.java +++ b/src/main/java/com/juick/server/XMPPConnection.java @@ -24,7 +24,6 @@ import com.juick.service.component.*; import com.juick.model.CommandResult; import com.juick.server.xmpp.iq.MessageQuery; import com.juick.server.xmpp.s2s.BasicXmppSession; -import com.juick.server.xmpp.s2s.StanzaListener; import com.juick.service.*; import com.juick.util.MessageUtils; import org.apache.commons.codec.digest.DigestUtils; @@ -81,14 +80,12 @@ import java.util.concurrent.ExecutorService; /** * @author ugnich */ -public class XMPPConnection implements StanzaListener, NotificationListener { +public class XMPPConnection implements NotificationListener { private static final Logger logger = LoggerFactory.getLogger("com.juick.server.xmpp"); private ExternalComponent router; @Inject - private XMPPServer xmpp; - @Inject private CommandsManager commandsManager; @Value("${xmppbot_jid:juick@localhost}") private Jid jid; @@ -96,6 +93,8 @@ public class XMPPConnection implements StanzaListener, NotificationListener { private String componentName; @Value("${component_port:5347}") private int componentPort; + @Value("${component_host:localhost}") + private String componentHost; @Value("${xmpp_password:secret}") private String password; @Value("${upload_tmp_dir:#{systemEnvironment['TEMP'] ?: '/tmp'}}") @@ -110,7 +109,7 @@ public class XMPPConnection implements StanzaListener, NotificationListener { @Inject private BasicXmppSession session; @Inject - private ExecutorService service; + private ExecutorService executorService; @Value("${service_user:juick}") private String serviceUsername; @Inject @@ -121,8 +120,7 @@ public class XMPPConnection implements StanzaListener, NotificationListener { @PostConstruct public void init() { logger.info("stream router start connecting to {}", componentPort); - xmpp.addStanzaListener(this); - router = ExternalComponent.create(componentName, password, session.getConfiguration(), "localhost", componentPort); + router = ExternalComponent.create(componentName, password, session.getConfiguration(), componentHost, componentPort); ServiceDiscoveryManager serviceDiscoveryManager = router.getManager(ServiceDiscoveryManager.class); serviceDiscoveryManager.addIdentity(Identity.clientBot().withName("Juick")); EntityCapabilitiesManager entityCapabilitiesManager = router.getManager(EntityCapabilitiesManager.class); @@ -259,7 +257,7 @@ public class XMPPConnection implements StanzaListener, NotificationListener { router.addInboundPresenceListener(event -> { incomingPresence(event.getPresence()); }); - service.submit(() -> { + executorService.submit(() -> { try { router.connect(); } catch (XmppException e) { @@ -644,11 +642,6 @@ public class XMPPConnection implements StanzaListener, NotificationListener { return null; } - @Override - public void stanzaReceived(Stanza xmlValue) { - router.send(xmlValue); - } - private void broadcastPresence(Presence.Type type) { Presence presence = new Presence(); presence.setFrom(jid); diff --git a/src/main/java/com/juick/server/XMPPServer.java b/src/main/java/com/juick/server/XMPPServer.java deleted file mode 100644 index 86ab6a78..00000000 --- a/src/main/java/com/juick/server/XMPPServer.java +++ /dev/null @@ -1,429 +0,0 @@ -/* - * 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 . - */ - -package com.juick.server; - -import com.juick.server.xmpp.router.StreamError; -import com.juick.server.xmpp.s2s.*; -import com.juick.service.UserService; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.scheduling.annotation.Scheduled; -import org.xmlpull.v1.XmlPullParserException; -import rocks.xmpp.addr.Jid; -import rocks.xmpp.core.stanza.model.Stanza; - -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 java.io.IOException; -import java.io.StringReader; -import java.net.ServerSocket; -import java.net.Socket; -import java.net.SocketException; -import java.security.InvalidAlgorithmParameterException; -import java.security.KeyStore; -import java.security.KeyStoreException; -import java.security.SecureRandom; -import java.security.cert.*; -import java.time.Duration; -import java.time.Instant; -import java.util.*; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.atomic.AtomicBoolean; - -/** - * @author ugnich - */ -public class XMPPServer implements ConnectionListener { - private static final Logger logger = LoggerFactory.getLogger("com.juick.server.xmpp"); - - 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("${broken_ssl_hosts:}") - public String[] brokenSSLhosts; - @Value("${banned_hosts:}") - public String[] bannedHosts; - - private final List inConnections = new CopyOnWriteArrayList<>(); - private final Map> outConnections = new ConcurrentHashMap<>(); - private final List outCache = new CopyOnWriteArrayList<>(); - private final List stanzaListeners = new CopyOnWriteArrayList<>(); - private final AtomicBoolean closeFlag = new AtomicBoolean(false); - - SSLContext sc; - CertificateFactory cf; - CertPathValidator cpv; - PKIXParameters params; - 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 new X509Certificate[0]; - } - } - }; - private boolean tlsConfigured = false; - - - private ServerSocket listener; - - @Inject - private BasicXmppSession session; - @Inject - private UserService userService; - @Inject - private KeystoreManager keystoreManager; - - @PostConstruct - public void init() throws KeyStoreException { - closeFlag.set(false); - try { - sc = SSLContext.getInstance("TLSv1.2"); - sc.init(keystoreManager.getKeymanagerFactory().getKeyManagers(), trustAllCerts, new SecureRandom()); - TrustManagerFactory trustManagerFactory = - TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); - Set ca = new HashSet<>(); - trustManagerFactory.init((KeyStore)null); - Arrays.stream(trustManagerFactory.getTrustManagers()).forEach(t -> Arrays.stream(((X509TrustManager)t).getAcceptedIssuers()).forEach(cert -> ca.add(new TrustAnchor(cert, null)))); - params = new PKIXParameters(ca); - params.setRevocationEnabled(false); - cpv = CertPathValidator.getInstance("PKIX"); - cf = CertificateFactory.getInstance( "X.509" ); - 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); - } - }); - } - - public void addConnectionIn(ConnectionIn c) { - c.setListener(this); - inConnections.add(c); - } - - public void addConnectionOut(ConnectionOut c, Optional 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 getConnectionOut(Jid hostname, boolean needReady) { - return outConnections.keySet().stream().filter(c -> c.to != null && - c.to.equals(hostname) && (!needReady || c.streamReady)).findFirst(); - } - - public Optional getConnectionIn(String streamID) { - return inConnections.stream().filter(c -> c.streamID != null && c.streamID.equals(streamID)).findFirst(); - } - - 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.connect(); - } catch (IOException e) { - logger.info("dialback to " + to + " exception", e); - } - }); - } - - public void startDialback(Jid from, String streamId, String dbKey) throws Exception { - Optional 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) { - logger.info("S2S: {}", xmlValue); - Stanza stanza = parse(xmlValue); - stanzaListeners.forEach(l -> l.stanzaReceived(stanza)); - } - - public BasicXmppSession getSession() { - return session; - } - - public List getInConnections() { - return inConnections; - } - - public Map> getOutConnections() { - return outConnections; - } - - @Override - public boolean isTlsAvailable() { - return tlsConfigured; - } - - @Override - public void starttls(ConnectionIn connection) { - logger.debug("stream {} securing", connection.streamID); - connection.sendStanza(""); - try { - connection.setSocket(sc.getSocketFactory().createSocket(connection.getSocket(), connection.getSocket().getInetAddress().getHostAddress(), - connection.getSocket().getPort(), false)); - SSLSocket sslSocket = (SSLSocket) connection.getSocket(); - sslSocket.addHandshakeCompletedListener(handshakeCompletedEvent -> { - try { - CertPath certPath = cf.generateCertPath(Arrays.asList(handshakeCompletedEvent.getPeerCertificates())); - cpv.validate(certPath, params); - connection.setTrusted(true); - logger.info("connection from {} is trusted", connection.from); - } catch (SSLPeerUnverifiedException | CertificateException | CertPathValidatorException | InvalidAlgorithmParameterException e) { - logger.info("connection from {} is NOT trusted, falling back to dialback", connection.from); - } - }); - sslSocket.setUseClientMode(false); - sslSocket.setNeedClientAuth(true); - sslSocket.startHandshake(); - connection.setSecured(true); - logger.debug("stream from {} secured", connection.streamID); - connection.restartParser(); - } catch (XmlPullParserException | IOException sex) { - logger.warn("stream {} ssl error {}", connection.streamID, sex); - connection.sendStanza(""); - 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(), false); - SSLSocket sslSocket = (SSLSocket) socket; - sslSocket.addHandshakeCompletedListener(handshakeCompletedEvent -> { - try { - CertPath certPath = cf.generateCertPath(Arrays.asList(handshakeCompletedEvent.getPeerCertificates())); - cpv.validate(certPath, params); - connection.setTrusted(true); - logger.info("connection to {} is trusted", connection.to); - } catch (SSLPeerUnverifiedException | CertificateException | CertPathValidatorException | InvalidAlgorithmParameterException e) { - logger.info("connection to {} is NOT trusted, falling back to dialback", connection.to); - } - }); - sslSocket.setNeedClientAuth(true); - sslSocket.startHandshake(); - connection.setSecured(true); - logger.debug("stream to {} 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(""); - 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 tlsConfigured && !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(c.getUpdated(), now).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(c.updated, now).toMinutes() > TIMEOUT_MINUTES) - .forEach(c -> { - logger.info("closing idle incoming connection from {}", c.from); - c.closeConnection(); - inConnections.remove(c); - }); - } - @PreDestroy - public void preDestroy() throws IOException { - closeFlag.set(true); - if (listener != null && !listener.isClosed()) { - listener.close(); - } - service.shutdown(); - logger.info("XMPP server destroyed"); - } - - public int getServerPort() { - return s2sPort; - } -} diff --git a/src/main/java/com/juick/server/configuration/XMPPConfig.java b/src/main/java/com/juick/server/configuration/XMPPConfig.java index 2feef286..ffacf4ef 100644 --- a/src/main/java/com/juick/server/configuration/XMPPConfig.java +++ b/src/main/java/com/juick/server/configuration/XMPPConfig.java @@ -1,10 +1,8 @@ package com.juick.server.configuration; import com.juick.server.XMPPConnection; -import com.juick.server.XMPPServer; import com.juick.server.xmpp.JidConverter; import com.juick.server.xmpp.iq.MessageQuery; -import com.juick.server.xmpp.router.XMPPRouter; import com.juick.server.xmpp.s2s.BasicXmppSession; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; @@ -40,15 +38,6 @@ public class XMPPConfig { return cs; } @Bean - public XMPPServer xmppServer() { - return new XMPPServer(); - } - @Bean - public XMPPRouter xmppRouter() { - return new XMPPRouter(); - } - @Bean - @DependsOn("xmppRouter") public XMPPConnection xmppConnection() { return new XMPPConnection(); } diff --git a/src/main/java/com/juick/server/xmpp/XMPPStatusPage.java b/src/main/java/com/juick/server/xmpp/XMPPStatusPage.java deleted file mode 100644 index 231696ec..00000000 --- a/src/main/java/com/juick/server/xmpp/XMPPStatusPage.java +++ /dev/null @@ -1,32 +0,0 @@ -package com.juick.server.xmpp; - -import com.juick.server.XMPPServer; -import com.juick.server.xmpp.helpers.XMPPStatus; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.http.MediaType; -import org.springframework.web.bind.annotation.RequestMapping; -import org.springframework.web.bind.annotation.RequestMethod; -import org.springframework.web.bind.annotation.RestController; -import rocks.xmpp.addr.Jid; -import springfox.documentation.annotations.ApiIgnore; - -import javax.inject.Inject; -import java.util.stream.Collectors; - -@RestController -@ConditionalOnProperty("xmppbot_jid") -public class XMPPStatusPage { - @Inject - private XMPPServer xmpp; - @ApiIgnore - @RequestMapping(method = RequestMethod.GET, value = "/api/xmpp-status", produces = MediaType.APPLICATION_JSON_UTF8_VALUE) - public XMPPStatus xmppStatus() { - XMPPStatus status = new XMPPStatus(); - if (xmpp != null) { - status.setInbound(xmpp.getInConnections().stream().map(c -> c.from).flatMap(j -> j.stream().map(Jid::getDomain)).collect(Collectors.toList())); - status.setOutbound(xmpp.getOutConnections().keySet().stream() - .map(c -> c.to).map(Jid::getDomain).collect(Collectors.toList())); - } - return status; - } -} diff --git a/src/main/java/com/juick/server/xmpp/helpers/XMPPStatus.java b/src/main/java/com/juick/server/xmpp/helpers/XMPPStatus.java deleted file mode 100644 index 99d89866..00000000 --- a/src/main/java/com/juick/server/xmpp/helpers/XMPPStatus.java +++ /dev/null @@ -1,48 +0,0 @@ -/* - * 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 . - */ - -package com.juick.server.xmpp.helpers; - -import com.juick.server.xmpp.s2s.ConnectionIn; -import com.juick.server.xmpp.s2s.ConnectionOut; - -import java.util.List; -import java.util.Set; - -/** - * Created by vitalyster on 16.02.2017. - */ -public class XMPPStatus { - private List inbound; - private List outbound; - - public List getInbound() { - return inbound; - } - - public void setInbound(List inbound) { - this.inbound = inbound; - } - - public List getOutbound() { - return outbound; - } - - public void setOutbound(List outbound) { - this.outbound = outbound; - } -} diff --git a/src/main/java/com/juick/server/xmpp/router/Handshake.java b/src/main/java/com/juick/server/xmpp/router/Handshake.java deleted file mode 100644 index 0bc501dd..00000000 --- a/src/main/java/com/juick/server/xmpp/router/Handshake.java +++ /dev/null @@ -1,39 +0,0 @@ -package com.juick.server.xmpp.router; - -import org.xmlpull.v1.XmlPullParser; -import org.xmlpull.v1.XmlPullParserException; - -import java.io.IOException; - -/** - * Created by vitalyster on 30.01.2017. - */ -public class Handshake { - private String value; - - public static Handshake parse(XmlPullParser parser) throws IOException, XmlPullParserException { - parser.next(); - Handshake handshake = new Handshake(); - handshake.setValue(XmlUtils.getTagText(parser)); - return handshake; - } - - public String getValue() { - return value; - } - - public void setValue(String value) { - this.value = value; - } - - @Override - public String toString() { - StringBuilder str = new StringBuilder("").append(getValue()).append(""); - } else { - str.append("/>"); - } - return str.toString(); - } -} diff --git a/src/main/java/com/juick/server/xmpp/router/Stream.java b/src/main/java/com/juick/server/xmpp/router/Stream.java deleted file mode 100644 index 2154edf6..00000000 --- a/src/main/java/com/juick/server/xmpp/router/Stream.java +++ /dev/null @@ -1,202 +0,0 @@ -/* - * Juick - * Copyright (C) 2008-2011, Ugnich Anton - * - * 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 . - */ -package com.juick.server.xmpp.router; - -import org.xmlpull.v1.XmlPullParser; -import org.xmlpull.v1.XmlPullParserException; -import org.xmlpull.v1.XmlPullParserFactory; -import rocks.xmpp.addr.Jid; - -import java.io.IOException; -import java.io.InputStream; -import java.io.InputStreamReader; -import java.io.OutputStream; -import java.io.OutputStreamWriter; -import java.nio.charset.StandardCharsets; -import java.time.Instant; -import java.util.UUID; - -/** - * - * @author Ugnich Anton - */ -public abstract class Stream { - - public boolean isLoggedIn() { - return loggedIn; - } - - public void setLoggedIn(boolean loggedIn) { - this.loggedIn = loggedIn; - } - - public Jid from; - public Jid to; - private InputStream is; - private OutputStream os; - private XmlPullParserFactory factory; - protected XmlPullParser parser; - private OutputStreamWriter writer; - StreamHandler streamHandler; - private boolean loggedIn; - private Instant created; - private Instant updated; - String streamId; - private boolean secured; - - public Stream(final Jid from, final Jid to, final InputStream is, final OutputStream os) throws XmlPullParserException { - this.from = from; - this.to = to; - this.is = is; - this.os = os; - factory = XmlPullParserFactory.newInstance(); - created = updated = Instant.now(); - streamId = UUID.randomUUID().toString(); - } - - public void restartStream() throws XmlPullParserException { - parser = factory.newPullParser(); - parser.setInput(new InputStreamReader(is, StandardCharsets.UTF_8)); - parser.setFeature(XmlPullParser.FEATURE_PROCESS_NAMESPACES, true); - writer = new OutputStreamWriter(os, StandardCharsets.UTF_8); - } - - public void connect() { - try { - restartStream(); - handshake(); - parse(); - } catch (XmlPullParserException e) { - StreamError invalidXmlError = new StreamError("invalid-xml"); - send(invalidXmlError.toString()); - connectionFailed(new Exception(invalidXmlError.getCondition())); - } catch (IOException e) { - connectionFailed(e); - } - } - - public void setHandler(final StreamHandler streamHandler) { - this.streamHandler = streamHandler; - } - - public abstract void handshake() throws XmlPullParserException, IOException; - - public void logoff() { - setLoggedIn(false); - try { - writer.flush(); - writer.close(); - //TODO close parser - } catch (final Exception e) { - connectionFailed(e); - } - } - - public void send(final String str) { - try { - updated = Instant.now(); - writer.write(str); - writer.flush(); - } catch (final Exception e) { - connectionFailed(e); - } - } - - private void parse() throws IOException, XmlPullParserException { - while (parser.next() != XmlPullParser.END_DOCUMENT) { - if (parser.getEventType() == XmlPullParser.IGNORABLE_WHITESPACE) { - setUpdated(); - } - if (parser.getEventType() != XmlPullParser.START_TAG) { - continue; - } - setUpdated(); - final String tag = parser.getName(); - switch (tag) { - case "message": - case "presence": - case "iq": - streamHandler.stanzaReceived(XmlUtils.parseToString(parser, false)); - break; - case "error": - StreamError error = StreamError.parse(parser); - connectionFailed(new Exception(error.getCondition())); - return; - default: - XmlUtils.skip(parser); - break; - } - } - } - - /** - * This method is used to be called on a parser or a connection error. - * It tries to close the XML-Reader and XML-Writer one last time. - */ - private void connectionFailed(final Exception ex) { - if (isLoggedIn()) { - try { - writer.close(); - //TODO close parser - } catch (Exception e) { - } - } - if (streamHandler != null) { - streamHandler.fail(ex); - } - } - - public Instant getCreated() { - return created; - } - - public Instant getUpdated() { - return updated; - } - public String getStreamId() { - return streamId; - } - - public boolean isSecured() { - return secured; - } - - public void setSecured(boolean secured) { - this.secured = secured; - } - - public void setUpdated() { - this.updated = Instant.now(); - } - - public InputStream getInputStream() { - return is; - } - - public void setInputStream(InputStream is) { - this.is = is; - } - - public OutputStream getOutputStream() { - return os; - } - - public void setOutputStream(OutputStream os) { - this.os = os; - } -} diff --git a/src/main/java/com/juick/server/xmpp/router/StreamComponentServer.java b/src/main/java/com/juick/server/xmpp/router/StreamComponentServer.java deleted file mode 100644 index a58adfc5..00000000 --- a/src/main/java/com/juick/server/xmpp/router/StreamComponentServer.java +++ /dev/null @@ -1,57 +0,0 @@ -package com.juick.server.xmpp.router; - -import org.apache.commons.codec.digest.DigestUtils; -import org.xmlpull.v1.XmlPullParserException; -import rocks.xmpp.addr.Jid; - -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.util.UUID; - -/** - * Created by vitalyster on 30.01.2017. - */ -public class StreamComponentServer extends Stream { - - private String streamId, secret; - - public String getStreamId() { - return streamId; - } - - - public StreamComponentServer(InputStream is, OutputStream os, String password) throws XmlPullParserException { - super(null, null, is, os); - secret = password; - streamId = UUID.randomUUID().toString(); - } - @Override - public void handshake() throws XmlPullParserException, IOException { - parser.next(); - if (!parser.getName().equals("stream") - || !parser.getNamespace(null).equals(StreamNamespaces.NS_COMPONENT_ACCEPT) - || !parser.getNamespace("stream").equals(StreamNamespaces.NS_STREAM)) { - throw new IOException("invalid stream"); - } - Jid domain = Jid.of(parser.getAttributeValue(null, "to")); - if (streamHandler.filter(null, domain)) { - send(new XMPPError(XMPPError.Type.cancel, "forbidden").toString()); - throw new IOException("invalid domain"); - } - from = domain; - to = domain; - send(String.format("", StreamNamespaces.NS_STREAM, StreamNamespaces.NS_COMPONENT_ACCEPT, from.asBareJid().toEscapedString(), streamId)); - Handshake handshake = Handshake.parse(parser); - boolean authenticated = handshake.getValue().equals(DigestUtils.sha1Hex(streamId + secret)); - setLoggedIn(authenticated); - if (!authenticated) { - send(new XMPPError(XMPPError.Type.cancel, "not-authorized").toString()); - streamHandler.fail(new IOException("stream:stream, failed authentication")); - return; - } - send(new Handshake().toString()); - streamHandler.ready(this); - } -} diff --git a/src/main/java/com/juick/server/xmpp/router/StreamError.java b/src/main/java/com/juick/server/xmpp/router/StreamError.java deleted file mode 100644 index f731f039..00000000 --- a/src/main/java/com/juick/server/xmpp/router/StreamError.java +++ /dev/null @@ -1,57 +0,0 @@ -package com.juick.server.xmpp.router; - -import org.xmlpull.v1.XmlPullParser; -import org.xmlpull.v1.XmlPullParserException; - -import java.io.IOException; - - -/** - * Created by vitalyster on 03.02.2017. - */ -public class StreamError { - - private String condition; - private String text; - - public StreamError() {} - - public StreamError(String condition) { - this.condition = condition; - } - - public static StreamError parse(XmlPullParser parser) throws IOException, XmlPullParserException { - StreamError streamError = new StreamError(); - final int initial = parser.getDepth(); - while (true) { - int eventType = parser.next(); - if (eventType == XmlPullParser.START_TAG && parser.getDepth() == initial + 1) { - final String tag = parser.getName(); - final String xmlns = parser.getNamespace(); - if (tag.equals("text") && xmlns.equals(StreamNamespaces.NS_XMPP_STREAMS)) { - streamError.text = XmlUtils.getTagText(parser); - } else if (xmlns.equals(StreamNamespaces.NS_XMPP_STREAMS)) { - streamError.condition = tag; - } else { - XmlUtils.skip(parser); - } - } else if (eventType == XmlPullParser.END_TAG && parser.getDepth() == initial) { - break; - } - } - return streamError; - } - - public String getCondition() { - return condition; - } - - @Override - public String toString() { - return String.format("<%s xmlns='%s'/>", condition, StreamNamespaces.NS_XMPP_STREAMS); - } - - public String getText() { - return text; - } -} diff --git a/src/main/java/com/juick/server/xmpp/router/StreamFeatures.java b/src/main/java/com/juick/server/xmpp/router/StreamFeatures.java deleted file mode 100644 index e8fc324f..00000000 --- a/src/main/java/com/juick/server/xmpp/router/StreamFeatures.java +++ /dev/null @@ -1,95 +0,0 @@ -/* - * Juick - * Copyright (C) 2008-2013, Ugnich Anton - * - * 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 . - */ -package com.juick.server.xmpp.router; - -import java.io.IOException; -import org.xmlpull.v1.XmlPullParser; -import org.xmlpull.v1.XmlPullParserException; - -/** - * - * @author Ugnich Anton - */ -public class StreamFeatures { - - public static final int NOTAVAILABLE = -1; - public static final int AVAILABLE = 0; - public static final int REQUIRED = 1; - public int STARTTLS = NOTAVAILABLE; - public int ZLIB = NOTAVAILABLE; - public int PLAIN = NOTAVAILABLE; - public int DIGEST_MD5 = NOTAVAILABLE; - public int REGISTER = NOTAVAILABLE; - public int EXTERNAL = NOTAVAILABLE; - - public static StreamFeatures parse(final XmlPullParser parser) throws XmlPullParserException, IOException { - StreamFeatures features = new StreamFeatures(); - final int initial = parser.getDepth(); - while (true) { - int eventType = parser.next(); - if (eventType == XmlPullParser.START_TAG && parser.getDepth() == initial + 1) { - final String tag = parser.getName(); - final String xmlns = parser.getNamespace(); - if (tag.equals("starttls") && xmlns != null && xmlns.equals("urn:ietf:params:xml:ns:xmpp-tls")) { - features.STARTTLS = AVAILABLE; - while (parser.next() == XmlPullParser.START_TAG) { - if (parser.getName().equals("required")) { - features.STARTTLS = REQUIRED; - } else { - XmlUtils.skip(parser); - } - } - } else if (tag.equals("compression") && xmlns != null && xmlns.equals("http://jabber.org/features/compress")) { - while (parser.next() == XmlPullParser.START_TAG) { - if (parser.getName().equals("method")) { - final String method = XmlUtils.getTagText(parser).toUpperCase(); - if (method.equals("ZLIB")) { - features.ZLIB = AVAILABLE; - } - } else { - XmlUtils.skip(parser); - } - } - } else if (tag.equals("mechanisms") && xmlns != null && xmlns.equals("urn:ietf:params:xml:ns:xmpp-sasl")) { - while (parser.next() == XmlPullParser.START_TAG) { - if (parser.getName().equals("mechanism")) { - final String mechanism = XmlUtils.getTagText(parser).toUpperCase(); - if (mechanism.equals("PLAIN")) { - features.PLAIN = AVAILABLE; - } else if (mechanism.equals("DIGEST-MD5")) { - features.DIGEST_MD5 = AVAILABLE; - } else if (mechanism.equals("EXTERNAL")) { - features.EXTERNAL = AVAILABLE; - } - } else { - XmlUtils.skip(parser); - } - } - } else if (tag.equals("register") && xmlns != null && xmlns.equals("http://jabber.org/features/iq-register")) { - features.REGISTER = AVAILABLE; - XmlUtils.skip(parser); - } else { - XmlUtils.skip(parser); - } - } else if (eventType == XmlPullParser.END_TAG && parser.getDepth() == initial) { - break; - } - } - return features; - } -} diff --git a/src/main/java/com/juick/server/xmpp/router/StreamHandler.java b/src/main/java/com/juick/server/xmpp/router/StreamHandler.java deleted file mode 100644 index 048c61ec..00000000 --- a/src/main/java/com/juick/server/xmpp/router/StreamHandler.java +++ /dev/null @@ -1,13 +0,0 @@ -package com.juick.server.xmpp.router; - -import rocks.xmpp.addr.Jid; - -/** - * Created by vitalyster on 01.02.2017. - */ -public interface StreamHandler { - void ready(StreamComponentServer componentServer); - void fail(final Exception ex); - boolean filter(Jid from, Jid to); - void stanzaReceived(String stanza); -} diff --git a/src/main/java/com/juick/server/xmpp/router/StreamNamespaces.java b/src/main/java/com/juick/server/xmpp/router/StreamNamespaces.java deleted file mode 100644 index 1b9b1965..00000000 --- a/src/main/java/com/juick/server/xmpp/router/StreamNamespaces.java +++ /dev/null @@ -1,10 +0,0 @@ -package com.juick.server.xmpp.router; - -public class StreamNamespaces { - public static final String NS_STREAM = "http://etherx.jabber.org/streams"; - public static final String NS_TLS = "urn:ietf:params:xml:ns:xmpp-tls"; - public static final String NS_DB = "jabber:server:dialback"; - public static final String NS_SERVER = "jabber:server"; - public static final String NS_COMPONENT_ACCEPT = "jabber:component:accept"; - public static final String NS_XMPP_STREAMS = "urn:ietf:params:xml:ns:xmpp-streams"; -} diff --git a/src/main/java/com/juick/server/xmpp/router/XMPPError.java b/src/main/java/com/juick/server/xmpp/router/XMPPError.java deleted file mode 100644 index 0cf9a3bc..00000000 --- a/src/main/java/com/juick/server/xmpp/router/XMPPError.java +++ /dev/null @@ -1,73 +0,0 @@ -/* - * Juick - * Copyright (C) 2008-2013, ugnich - * - * 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 . - */ -package com.juick.server.xmpp.router; - -import org.apache.commons.text.StringEscapeUtils; - -/** - * - * @author ugnich - */ -public class XMPPError { - - public static final class Type { - - public static final String auth = "auth"; - public static final String cancel = "cancel"; - public static final String continue_ = "continue"; - public static final String modify = "modify"; - public static final String wait = "wait"; - } - private final static String TagName = "error"; - public String by = null; - private String type; - private String condition; - private String text = null; - - public XMPPError(String type, String condition) { - this.type = type; - this.condition = condition; - } - - @Override - public String toString() { - StringBuilder str = new StringBuilder("<").append(TagName).append(""); - if (by != null) { - str.append(" by=\"").append(StringEscapeUtils.escapeXml10(by)).append("\""); - } - if (type != null) { - str.append(" type=\"").append(StringEscapeUtils.escapeXml10(type)).append("\""); - } - - if (condition != null) { - str.append(">"); - str.append("<").append(StringEscapeUtils.escapeXml10(condition)).append(" xmlns=\"urn:ietf:params:xml:ns:xmpp-stanzas\""); - if (text != null) { - str.append(">").append(StringEscapeUtils.escapeXml10(text)).append(""); - } else { - str.append("/>"); - } - str.append(""); - } else { - str.append("/>"); - } - - return str.toString(); - } -} diff --git a/src/main/java/com/juick/server/xmpp/router/XMPPRouter.java b/src/main/java/com/juick/server/xmpp/router/XMPPRouter.java deleted file mode 100644 index 6d67fa9c..00000000 --- a/src/main/java/com/juick/server/xmpp/router/XMPPRouter.java +++ /dev/null @@ -1,220 +0,0 @@ -package com.juick.server.xmpp.router; - -import com.juick.server.XMPPServer; -import com.juick.server.xmpp.s2s.BasicXmppSession; -import com.juick.server.xmpp.s2s.CacheEntry; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.stereotype.Component; -import org.xmlpull.v1.XmlPullParserException; -import rocks.xmpp.addr.Jid; -import rocks.xmpp.core.stanza.model.IQ; -import rocks.xmpp.core.stanza.model.Message; -import rocks.xmpp.core.stanza.model.Presence; -import rocks.xmpp.core.stanza.model.Stanza; -import rocks.xmpp.core.stanza.model.server.ServerIQ; -import rocks.xmpp.core.stanza.model.server.ServerMessage; -import rocks.xmpp.core.stanza.model.server.ServerPresence; -import rocks.xmpp.util.XmppUtils; - -import javax.annotation.PostConstruct; -import javax.annotation.PreDestroy; -import javax.inject.Inject; -import javax.xml.bind.JAXBException; -import javax.xml.bind.Unmarshaller; -import javax.xml.stream.XMLStreamException; -import javax.xml.stream.XMLStreamWriter; -import java.io.IOException; -import java.io.StringReader; -import java.io.StringWriter; -import java.net.ServerSocket; -import java.net.Socket; -import java.net.SocketException; -import java.time.Instant; -import java.util.ArrayList; -import java.util.Collections; -import java.util.Iterator; -import java.util.List; -import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.ExecutorService; - -public class XMPPRouter implements StreamHandler { - private static final Logger logger = LoggerFactory.getLogger("com.juick.server.xmpp"); - - @Inject - private ExecutorService service; - - private final List connections = Collections.synchronizedList(new ArrayList<>()); - private final List outCache = new CopyOnWriteArrayList<>(); - - private ServerSocket listener; - - @Inject - private BasicXmppSession session; - - @Value("${router_port:5347}") - private int routerPort; - - @Inject - private XMPPServer xmppServer; - - @PostConstruct - public void init() { - logger.info("component router initialized"); - service.submit(() -> { - try { - listener = new ServerSocket(routerPort); - logger.info("component router listening on {}", routerPort); - while (!listener.isClosed()) { - if (Thread.currentThread().isInterrupted()) break; - Socket socket = listener.accept(); - service.submit(() -> { - try { - StreamComponentServer client = new StreamComponentServer(socket.getInputStream(), socket.getOutputStream(), "secret"); - addConnectionIn(client); - client.setHandler(this); - client.connect(); - } catch (IOException e) { - logger.error("component error", e); - } catch (XmlPullParserException e) { - e.printStackTrace(); - } - }); - } - } catch (SocketException e) { - // shutdown - } catch (IOException e) { - logger.warn("io exception", e); - } - }); - } - - @PreDestroy - public void close() throws Exception { - if (!listener.isClosed()) { - listener.close(); - } - synchronized (getConnections()) { - for (Iterator i = getConnections().iterator(); i.hasNext(); ) { - StreamComponentServer c = i.next(); - c.logoff(); - i.remove(); - } - } - service.shutdown(); - logger.info("XMPP router destroyed"); - } - - private void addConnectionIn(StreamComponentServer c) { - synchronized (getConnections()) { - getConnections().add(c); - } - } - - private 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.info("XMPPRouter (out): {}", xml); - sendOut(s.getTo().getDomain(), xml); - } catch (XMLStreamException | JAXBException e1) { - logger.info("jaxb exception", e1); - } - } - - private void sendOut(String hostname, String xml) { - boolean haveAnyConn = false; - - StreamComponentServer connOut = null; - synchronized (getConnections()) { - for (StreamComponentServer c : getConnections()) { - if (c.to != null && c.to.getDomain().equals(hostname)) { - if (c.isLoggedIn()) { - connOut = c; - break; - } else { - logger.info("bouncing stanza to {} component until it will be ready", hostname); - boolean haveCache = false; - for (CacheEntry entry : outCache) { - if (entry.hostname != null && entry.hostname.equals(hostname)) { - entry.xml += xml; - entry.updated = Instant.now(); - haveCache = true; - break; - } - } - if (!haveCache) { - outCache.add(new CacheEntry(Jid.of(hostname), xml)); - } - } - } - } - } - if (connOut != null) { - connOut.send(xml); - return; - } - xmppServer.sendOut(Jid.of(hostname), xml); - - } - - public List getConnections() { - return connections; - } - - private 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; - } - @Override - public void stanzaReceived(String stanza) { - Stanza input = parse(stanza); - if (input instanceof Message) { - sendOut(ServerMessage.from((Message)input)); - } else if (input instanceof IQ) { - sendOut(ServerIQ.from((IQ)input)); - } else { - sendOut(ServerPresence.from((Presence) input)); - } - } - - 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]; - } - - @Override - public void ready(StreamComponentServer componentServer) { - logger.info("component {} ready", componentServer.to); - String cache = getFromCache(componentServer.to); - if (cache != null) { - logger.debug("sending cache to {}", componentServer.to); - componentServer.send(cache); - } - } - - @Override - public void fail(Exception e) { - - } - - @Override - public boolean filter(Jid jid, Jid jid1) { - return false; - } -} \ No newline at end of file diff --git a/src/main/java/com/juick/server/xmpp/router/XmlUtils.java b/src/main/java/com/juick/server/xmpp/router/XmlUtils.java deleted file mode 100644 index 7579489f..00000000 --- a/src/main/java/com/juick/server/xmpp/router/XmlUtils.java +++ /dev/null @@ -1,88 +0,0 @@ -/* - * Juick - * Copyright (C) 2008-2011, Ugnich Anton - * - * 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 . - */ -package com.juick.server.xmpp.router; - -import java.io.IOException; - -import org.apache.commons.text.StringEscapeUtils; -import org.xmlpull.v1.XmlPullParser; -import org.xmlpull.v1.XmlPullParserException; - -/** - * - * @author Ugnich Anton - */ -public class XmlUtils { - - public static void skip(XmlPullParser parser) throws XmlPullParserException, IOException { - String tag = parser.getName(); - while (parser.getName() != null && !(parser.next() == XmlPullParser.END_TAG && parser.getName().equals(tag))) { - } - } - - public static String getTagText(XmlPullParser parser) throws XmlPullParserException, IOException { - String ret = ""; - String tag = parser.getName(); - - if (parser.next() == XmlPullParser.TEXT) { - ret = parser.getText(); - } - - while (!(parser.getEventType() == XmlPullParser.END_TAG && parser.getName().equals(tag))) { - parser.next(); - } - - return ret; - } - - public static String parseToString(XmlPullParser parser, boolean skipXMLNS) throws XmlPullParserException, IOException { - String tag = parser.getName(); - StringBuilder ret = new StringBuilder("<").append(tag); - - // skipXMLNS for xmlns="jabber:client" - - String ns = parser.getNamespace(); - if (!skipXMLNS && ns != null && !ns.isEmpty()) { - ret.append(" xmlns=\"").append(ns).append("\""); - } - - for (int i = 0; i < parser.getAttributeCount(); i++) { - String attr = parser.getAttributeName(i); - if ((!skipXMLNS || !attr.equals("xmlns")) && !attr.contains(":")) { - ret.append(" ").append(attr).append("=\"").append(StringEscapeUtils.escapeXml10(parser.getAttributeValue(i))).append("\""); - } - } - ret.append(">"); - - while (!(parser.next() == XmlPullParser.END_TAG && parser.getName().equals(tag))) { - int event = parser.getEventType(); - if (event == XmlPullParser.START_TAG) { - if (!parser.getName().contains(":")) { - ret.append(parseToString(parser, false)); - } else { - skip(parser); - } - } else if (event == XmlPullParser.TEXT) { - ret.append(StringEscapeUtils.escapeXml10(parser.getText())); - } - } - - ret.append(""); - return ret.toString(); - } -} diff --git a/src/main/java/com/juick/server/xmpp/s2s/CacheEntry.java b/src/main/java/com/juick/server/xmpp/s2s/CacheEntry.java deleted file mode 100644 index 33e875bd..00000000 --- a/src/main/java/com/juick/server/xmpp/s2s/CacheEntry.java +++ /dev/null @@ -1,40 +0,0 @@ -/* - * 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 . - */ - -package com.juick.server.xmpp.s2s; - -import rocks.xmpp.addr.Jid; - -import java.time.Instant; - -/** - * - * @author ugnich - */ -public class CacheEntry { - - public Jid hostname; - public Instant created; - public Instant updated; - public String xml; - - public CacheEntry(Jid hostname, String xml) { - this.hostname = hostname; - this.created = this.updated =Instant.now(); - this.xml = xml; - } -} diff --git a/src/main/java/com/juick/server/xmpp/s2s/Connection.java b/src/main/java/com/juick/server/xmpp/s2s/Connection.java deleted file mode 100644 index 4fa8e741..00000000 --- a/src/main/java/com/juick/server/xmpp/s2s/Connection.java +++ /dev/null @@ -1,158 +0,0 @@ -/* - * 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 . - */ - -package com.juick.server.xmpp.s2s; - -import com.fasterxml.jackson.annotation.JsonIgnore; -import com.juick.server.XMPPServer; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.xmlpull.v1.XmlPullParser; -import org.xmlpull.v1.XmlPullParserException; -import org.xmlpull.v1.XmlPullParserFactory; - -import java.io.IOException; -import java.io.InputStreamReader; -import java.io.OutputStreamWriter; -import java.net.Socket; -import java.nio.charset.StandardCharsets; -import java.time.Instant; -import java.util.UUID; - -/** - * - * @author ugnich - */ -public class Connection { - - protected static final Logger logger = LoggerFactory.getLogger(Connection.class); - - public String streamID; - public Instant created; - public Instant updated; - public long bytesLocal = 0; - public long packetsLocal = 0; - XMPPServer xmpp; - private Socket socket; - public static final String NS_DB = "jabber:server:dialback"; - public static final String NS_TLS = "urn:ietf:params:xml:ns:xmpp-tls"; - public static final String NS_SASL = "urn:ietf:params:xml:ns:xmpp-sasl"; - public static final String NS_STREAM = "http://etherx.jabber.org/streams"; - XmlPullParserFactory factory = XmlPullParserFactory.newInstance(); - XmlPullParser parser = factory.newPullParser(); - OutputStreamWriter writer; - private boolean secured = false; - private boolean authenticated = false; - private boolean trusted = false; - - - - public Connection(XMPPServer xmpp) throws XmlPullParserException { - this.xmpp = xmpp; - created = updated = Instant.now(); - } - - public void logParser() { - if (streamID == null) { - return; - } - String tag = "IN: <" + parser.getName(); - for (int i = 0; i < parser.getAttributeCount(); i++) { - tag += " " + parser.getAttributeName(i) + "=\"" + parser.getAttributeValue(i) + "\""; - } - tag += ">...\n"; - logger.trace(tag); - } - - public void sendStanza(String xml) { - if (streamID != null) { - logger.trace("OUT: {}\n", xml); - } - try { - writer.write(xml); - writer.flush(); - } catch (IOException e) { - logger.error("send stanza failed", e); - } - - updated = Instant.now(); - bytesLocal += xml.length(); - packetsLocal++; - } - - public void closeConnection() { - if (streamID != null) { - logger.debug("closing stream {}", streamID); - } - - try { - writer.write(""); - } catch (Exception e) { - } - - try { - writer.close(); - } catch (Exception e) { - } - - try { - socket.close(); - } catch (Exception e) { - } - } - - public boolean isSecured() { - return secured; - } - - public void setSecured(boolean secured) { - this.secured = secured; - } - - public void restartParser() throws XmlPullParserException, IOException { - streamID = UUID.randomUUID().toString(); - parser = factory.newPullParser(); - parser.setFeature(XmlPullParser.FEATURE_PROCESS_NAMESPACES, true); - parser.setInput(new InputStreamReader(socket.getInputStream())); - writer = new OutputStreamWriter(socket.getOutputStream(), StandardCharsets.UTF_8); - } - - @JsonIgnore - public Socket getSocket() { - return socket; - } - - public void setSocket(Socket socket) { - this.socket = socket; - } - - public boolean isAuthenticated() { - return authenticated; - } - - public void setAuthenticated(boolean authenticated) { - this.authenticated = authenticated; - } - - public boolean isTrusted() { - return trusted; - } - - public void setTrusted(boolean trusted) { - this.trusted = trusted; - } -} diff --git a/src/main/java/com/juick/server/xmpp/s2s/ConnectionIn.java b/src/main/java/com/juick/server/xmpp/s2s/ConnectionIn.java deleted file mode 100644 index 3929a69f..00000000 --- a/src/main/java/com/juick/server/xmpp/s2s/ConnectionIn.java +++ /dev/null @@ -1,232 +0,0 @@ -/* - * 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 . - */ - -package com.juick.server.xmpp.s2s; - -import com.juick.server.XMPPServer; -import com.juick.server.xmpp.router.StreamError; -import com.juick.server.xmpp.router.XmlUtils; -import org.apache.commons.lang3.StringUtils; -import org.xmlpull.v1.XmlPullParser; -import org.xmlpull.v1.XmlPullParserException; -import rocks.xmpp.addr.Jid; - -import java.io.EOFException; -import java.io.IOException; -import java.net.Socket; -import java.net.SocketException; -import java.time.Instant; -import java.util.Arrays; -import java.util.List; -import java.util.UUID; -import java.util.concurrent.CopyOnWriteArrayList; -import java.util.stream.Collectors; - -/** - * @author ugnich - */ -public class ConnectionIn extends Connection implements Runnable { - - final public List from = new CopyOnWriteArrayList<>(); - public Instant received; - public long packetsRemote = 0; - ConnectionListener listener; - - public ConnectionIn(XMPPServer xmpp, Socket socket) throws XmlPullParserException, IOException { - super(xmpp); - this.setSocket(socket); - restartParser(); - } - - @Override - public void run() { - try { - parser.next(); // stream:stream - updateTsRemoteData(); - if (!parser.getName().equals("stream") - || !parser.getNamespace("stream").equals(NS_STREAM)) { -// || !parser.getAttributeValue(null, "version").equals("1.0") -// || !parser.getAttributeValue(null, "to").equals(Main.HOSTNAME)) { - throw new Exception(String.format("stream from %s invalid", getSocket().getRemoteSocketAddress())); - } - streamID = parser.getAttributeValue(null, "id"); - if (streamID == null) { - streamID = UUID.randomUUID().toString(); - } - boolean xmppversionnew = parser.getAttributeValue(null, "version") != null; - String from = parser.getAttributeValue(null, "from"); - Thread.currentThread().setName(String.format("XMPP S2S IN %s - %d", from, Thread.currentThread().getId())); - - if (Arrays.asList(xmpp.bannedHosts).contains(from)) { - closeConnection(); - return; - } - sendOpenStream(from, xmppversionnew); - - while (parser.next() != XmlPullParser.END_DOCUMENT) { - updateTsRemoteData(); - if (parser.getEventType() != XmlPullParser.START_TAG) { - continue; - } - logParser(); - - packetsRemote++; - - String tag = parser.getName(); - if (tag.equals("result") && parser.getNamespace().equals(NS_DB)) { - String dfrom = parser.getAttributeValue(null, "from"); - String to = parser.getAttributeValue(null, "to"); - logger.debug("stream from {} to {} {} asking for dialback", dfrom, to, streamID); - if (dfrom.endsWith(xmpp.getJid().toEscapedString()) && (dfrom.equals(xmpp.getJid().toEscapedString()) - || dfrom.endsWith("." + xmpp.getJid()))) { - logger.warn("stream from {} is invalid", dfrom); - break; - } - if (to != null && to.equals(xmpp.getJid().toEscapedString())) { - String dbKey = XmlUtils.getTagText(parser); - updateTsRemoteData(); - xmpp.startDialback(Jid.of(dfrom), streamID, dbKey); - } else { - logger.warn("stream from " + dfrom + " " + streamID + " invalid to " + to); - break; - } - } else if (tag.equals("verify") && parser.getNamespace().equals(NS_DB)) { - String vfrom = parser.getAttributeValue(null, "from"); - String vto = parser.getAttributeValue(null, "to"); - String vid = parser.getAttributeValue(null, "id"); - String vkey = XmlUtils.getTagText(parser); - updateTsRemoteData(); - final boolean[] valid = {false}; - if (vfrom != null && vto != null && vid != null && vkey != null) { - xmpp.getConnectionOut(Jid.of(vfrom), false).ifPresent(c -> { - String dialbackKey = c.dbKey; - valid[0] = vkey.equals(dialbackKey); - }); - } - if (valid[0]) { - sendStanza(""); - logger.debug("stream from {} {} dialback verify valid", vfrom, streamID); - setAuthenticated(true); - } else { - sendStanza(""); - logger.warn("stream from {} {} dialback verify invalid", vfrom, streamID); - } - } else if (tag.equals("presence") && checkFromTo(parser) && isAuthenticated()) { - String xml = XmlUtils.parseToString(parser, false); - logger.debug("stream {} presence: {}", streamID, xml); - xmpp.onStanzaReceived(xml); - } else if (tag.equals("message") && checkFromTo(parser)) { - updateTsRemoteData(); - String xml = XmlUtils.parseToString(parser, false); - logger.debug("stream {} message: {}", streamID, xml); - xmpp.onStanzaReceived(xml); - - } else if (tag.equals("iq") && checkFromTo(parser) && isAuthenticated()) { - updateTsRemoteData(); - String type = parser.getAttributeValue(null, "type"); - String xml = XmlUtils.parseToString(parser, false); - if (type == null || !type.equals("error")) { - logger.debug("stream {} iq: {}", streamID, xml); - xmpp.onStanzaReceived(xml); - } - } else if (!isSecured() && tag.equals("starttls") && !isAuthenticated()) { - listener.starttls(this); - } else if (isSecured() && tag.equals("stream") && parser.getNamespace().equals(NS_STREAM)) { - sendOpenStream(null, true); - } else if (isSecured() && tag.equals("auth") && parser.getNamespace().equals(NS_SASL) - && parser.getAttributeValue(null, "mechanism").equals("EXTERNAL") - && !isAuthenticated() && isTrusted()) { - sendStanza(""); - logger.info("stream {} authenticated externally", streamID); - this.from.add(Jid.of(from)); - setAuthenticated(true); - restartParser(); - } else if (tag.equals("error")) { - StreamError streamError = StreamError.parse(parser); - logger.debug("Stream error {} from {}: {}", streamError.getCondition(), streamID, streamError.getText()); - xmpp.removeConnectionIn(this); - closeConnection(); - } else { - String unhandledStanza = XmlUtils.parseToString(parser, true); - logger.warn("Unhandled stanza from {}: {}", streamID, unhandledStanza); - } - } - logger.warn("stream {} finished", streamID); - xmpp.removeConnectionIn(this); - closeConnection(); - } catch (EOFException | SocketException ex) { - logger.debug("stream {} closed (dirty)", streamID); - xmpp.removeConnectionIn(this); - closeConnection(); - } catch (Exception e) { - logger.debug("stream {} error {}", streamID, e); - xmpp.removeConnectionIn(this); - closeConnection(); - } - } - - void updateTsRemoteData() { - received = Instant.now(); - } - - void sendOpenStream(String from, boolean xmppversionnew) throws IOException { - String openStream = ""; - if (xmppversionnew) { - openStream += ""; - if (listener != null && listener.isTlsAvailable() && !Arrays.asList(xmpp.brokenSSLhosts).contains(from)) { - if (!isSecured()) { - openStream += ""; - } else if (!isAuthenticated() && isTrusted()) { - openStream += "" + - "EXTERNAL" + - ""; - } - } - openStream += ""; - } - sendStanza(openStream); - } - - public void sendDialbackResult(Jid sfrom, String type) { - sendStanza(""); - if (type.equals("valid")) { - from.add(sfrom); - logger.debug("stream from {} {} ready", sfrom, streamID); - setAuthenticated(true); - } - } - - boolean checkFromTo(XmlPullParser parser) throws Exception { - String cfrom = parser.getAttributeValue(null, "from"); - String cto = parser.getAttributeValue(null, "to"); - if (StringUtils.isNotEmpty(cfrom) && StringUtils.isNotEmpty(cto)) { - Jid jidfrom = Jid.of(cfrom); - for (Jid aFrom : from) { - if (aFrom.equals(Jid.of(jidfrom.getDomain()))) { - return true; - } - } - } - logger.warn("rejected from {}, to {}, stream {}", cfrom, cto, from.stream().collect(Collectors.joining(","))); - return false; - } - public void setListener(ConnectionListener listener) { - this.listener = listener; - } -} diff --git a/src/main/java/com/juick/server/xmpp/s2s/ConnectionListener.java b/src/main/java/com/juick/server/xmpp/s2s/ConnectionListener.java deleted file mode 100644 index 4c32b9ae..00000000 --- a/src/main/java/com/juick/server/xmpp/s2s/ConnectionListener.java +++ /dev/null @@ -1,16 +0,0 @@ -package com.juick.server.xmpp.s2s; - - -import com.juick.server.xmpp.router.StreamError; - -public interface ConnectionListener { - boolean isTlsAvailable(); - void starttls(ConnectionIn connection); - void proceed(ConnectionOut connection); - void verify(ConnectionOut connection, String from, String type, String sid); - void dialbackError(ConnectionOut connection, StreamError error); - void finished(ConnectionOut connection, boolean dirty); - void exception(ConnectionOut connection, Exception ex); - void ready(ConnectionOut connection); - boolean securing(ConnectionOut connection); -} diff --git a/src/main/java/com/juick/server/xmpp/s2s/ConnectionOut.java b/src/main/java/com/juick/server/xmpp/s2s/ConnectionOut.java deleted file mode 100644 index 2b919da4..00000000 --- a/src/main/java/com/juick/server/xmpp/s2s/ConnectionOut.java +++ /dev/null @@ -1,190 +0,0 @@ -/* - * 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 . - */ - -package com.juick.server.xmpp.s2s; - -import com.juick.server.xmpp.router.Stream; -import com.juick.server.xmpp.router.StreamError; -import com.juick.server.xmpp.router.StreamFeatures; -import com.juick.server.xmpp.router.XmlUtils; -import com.juick.server.xmpp.s2s.util.DialbackUtils; -import org.apache.commons.codec.Charsets; -import org.apache.commons.codec.binary.Base64; -import org.apache.commons.text.RandomStringGenerator; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.xmlpull.v1.XmlPullParser; -import rocks.xmpp.addr.Jid; - -import java.io.EOFException; -import java.io.IOException; -import java.io.InputStream; -import java.io.OutputStream; -import java.net.SocketException; -import java.util.UUID; - -import static com.juick.server.xmpp.router.StreamNamespaces.NS_STREAM; -import static com.juick.server.xmpp.s2s.Connection.NS_SASL; - -/** - * @author ugnich - */ -public class ConnectionOut extends Stream { - protected static final Logger logger = LoggerFactory.getLogger(ConnectionOut.class); - public static final String NS_TLS = "urn:ietf:params:xml:ns:xmpp-tls"; - public static final String NS_DB = "jabber:server:dialback"; - private boolean secured = false; - private boolean trusted = false; - public boolean streamReady = false; - String checkSID = null; - String dbKey = null; - private String streamID; - ConnectionListener listener; - RandomStringGenerator generator = new RandomStringGenerator.Builder().withinRange('a', 'z').build(); - - public ConnectionOut(Jid from, Jid to, InputStream is, OutputStream os, String checkSID, String dbKey) throws Exception { - super(from, to, is, os); - this.to = to; - this.checkSID = checkSID; - this.dbKey = dbKey; - if (dbKey == null) { - this.dbKey = DialbackUtils.generateDialbackKey(generator.generate(15), to, from, streamID); - } - streamID = UUID.randomUUID().toString(); - Thread.currentThread().setName(String.format("XMPP S2S OUT %s - %d", to.toEscapedString(), Thread.currentThread().getId())); - } - - public void sendOpenStream() throws IOException { - send(""); - } - - void processDialback() throws Exception { - if (checkSID != null) { - sendDialbackVerify(checkSID, dbKey); - } - send("" + - dbKey + ""); - } - - @Override - public void handshake() { - try { - restartStream(); - - sendOpenStream(); - - parser.next(); // stream:stream - streamID = parser.getAttributeValue(null, "id"); - if (streamID == null || streamID.isEmpty()) { - throw new Exception("stream to " + to + " invalid first packet"); - } - - logger.debug("stream to {} {} open", to, streamID); - boolean xmppversionnew = parser.getAttributeValue(null, "version") != null; - if (!xmppversionnew) { - processDialback(); - } - - while (parser.next() != XmlPullParser.END_DOCUMENT) { - if (parser.getEventType() != XmlPullParser.START_TAG) { - continue; - } - - String tag = parser.getName(); - if (tag.equals("result") && parser.getNamespace().equals(NS_DB)) { - String type = parser.getAttributeValue(null, "type"); - if (type != null && type.equals("valid")) { - streamReady = true; - listener.ready(this); - } else { - logger.warn("stream to {} {} dialback fail", to, streamID); - } - XmlUtils.skip(parser); - } else if (tag.equals("verify") && parser.getNamespace().equals(NS_DB)) { - String from = parser.getAttributeValue(null, "from"); - String type = parser.getAttributeValue(null, "type"); - String sid = parser.getAttributeValue(null, "id"); - listener.verify(this, from, type, sid); - XmlUtils.skip(parser); - } else if (tag.equals("features") && parser.getNamespace().equals(NS_STREAM)) { - StreamFeatures features = StreamFeatures.parse(parser); - if (listener != null && !secured && features.STARTTLS >= 0 - && listener.securing(this)) { - logger.debug("stream to {} {} securing", to.toEscapedString(), streamID); - send(""); - } else if (secured && features.EXTERNAL >=0) { - String authid = Base64.encodeBase64String(from.toEscapedString().getBytes(Charsets.UTF_8)); - send(String.format("%s", authid)); - } else if (secured && streamReady) { - listener.ready(this); - } else { - processDialback(); - } - } else if (tag.equals("proceed") && parser.getNamespace().equals(NS_TLS)) { - listener.proceed(this); - } else if (tag.equals("success") && parser.getNamespace().equals(NS_SASL)) { - streamReady = true; - restartStream(); - sendOpenStream(); - } else if (secured && tag.equals("stream") && parser.getNamespace().equals(NS_STREAM)) { - streamID = parser.getAttributeValue(null, "id"); - } else if (tag.equals("error")) { - StreamError streamError = StreamError.parse(parser); - listener.dialbackError(this, streamError); - } else { - String unhandledStanza = XmlUtils.parseToString(parser, false); - logger.warn("Unhandled stanza from {} {} : {}", to, streamID, unhandledStanza); - } - } - listener.finished(this, false); - } catch (EOFException | SocketException eofex) { - listener.finished(this, true); - } catch (Exception e) { - listener.exception(this, e); - } - } - - public void sendDialbackVerify(String sid, String key) { - send("" + - key + ""); - } - public void setListener(ConnectionListener listener) { - this.listener = listener; - } - - public String getStreamID() { - return streamID; - } - - public boolean isSecured() { - return secured; - } - - public void setSecured(boolean secured) { - this.secured = secured; - } - - public boolean isTrusted() { - return trusted; - } - - public void setTrusted(boolean trusted) { - this.trusted = trusted; - } -} diff --git a/src/main/java/com/juick/server/xmpp/s2s/DNSQueries.java b/src/main/java/com/juick/server/xmpp/s2s/DNSQueries.java deleted file mode 100644 index 1367d333..00000000 --- a/src/main/java/com/juick/server/xmpp/s2s/DNSQueries.java +++ /dev/null @@ -1,65 +0,0 @@ -/* - * 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 . - */ - -package com.juick.server.xmpp.s2s; - -import org.apache.commons.lang3.math.NumberUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.net.InetSocketAddress; -import java.util.Hashtable; -import java.util.Random; -import javax.naming.NamingException; -import javax.naming.directory.Attribute; -import javax.naming.directory.DirContext; -import javax.naming.directory.InitialDirContext; - -/** - * - * @author ugnich - */ -public class DNSQueries { - - private static final Logger logger = LoggerFactory.getLogger(DNSQueries.class); - - private static Random rand = new Random(); - - public static InetSocketAddress getServerAddress(String hostname) { - - String host = hostname; - int port = 5269; - - Hashtable env = new Hashtable<>(5); - env.put("java.naming.factory.initial", "com.sun.jndi.dns.DnsContextFactory"); - try { - DirContext ctx = new InitialDirContext(env); - Attribute att = ctx.getAttributes("_xmpp-server._tcp." + hostname, new String[]{"SRV"}).get("SRV"); - - if (att != null && att.size() > 0) { - int i = rand.nextInt(att.size()); - String srv[] = att.get(i).toString().split(" "); - port = NumberUtils.toInt(srv[2], 5269); - host = srv[3]; - } - ctx.close(); - } catch (NamingException e) { - logger.debug("SRV record for {} is not resolved, falling back to A record", hostname); - } - return new InetSocketAddress(host, port); - } -} diff --git a/src/main/java/com/juick/server/xmpp/s2s/util/DialbackUtils.java b/src/main/java/com/juick/server/xmpp/s2s/util/DialbackUtils.java deleted file mode 100644 index d25dbad8..00000000 --- a/src/main/java/com/juick/server/xmpp/s2s/util/DialbackUtils.java +++ /dev/null @@ -1,37 +0,0 @@ -/* - * 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 . - */ - -package com.juick.server.xmpp.s2s.util; - -import org.apache.commons.codec.digest.DigestUtils; -import org.apache.commons.codec.digest.HmacAlgorithms; -import org.apache.commons.codec.digest.HmacUtils; -import rocks.xmpp.addr.Jid; - -/** - * Created by vitalyster on 05.12.2016. - */ -public class DialbackUtils { - private DialbackUtils() { - throw new IllegalStateException(); - } - - public static String generateDialbackKey(String secret, Jid to, Jid from, String id) { - return new HmacUtils(HmacAlgorithms.HMAC_SHA_256, DigestUtils.sha256(secret)) - .hmacHex(to.toEscapedString() + " " + from.toEscapedString() + " " + id); - } -} diff --git a/src/test/java/com/juick/server/tests/ServerTests.java b/src/test/java/com/juick/server/tests/ServerTests.java index 32afd4f9..620e210c 100644 --- a/src/test/java/com/juick/server/tests/ServerTests.java +++ b/src/test/java/com/juick/server/tests/ServerTests.java @@ -42,8 +42,6 @@ import com.juick.server.api.xnodeinfo2.model.NodeInfo; import com.juick.server.util.HttpUtils; import com.juick.server.util.ImageUtils; import com.juick.server.www.WebApp; -import com.juick.server.xmpp.helpers.XMPPStatus; -import com.juick.server.xmpp.s2s.ConnectionIn; import com.juick.service.*; import com.juick.service.component.MessageEvent; import com.juick.util.DateFormattersHolder; @@ -51,7 +49,6 @@ import com.juick.util.MessageUtils; import com.mitchellbosecke.pebble.PebbleEngine; import com.mitchellbosecke.pebble.error.PebbleException; import com.mitchellbosecke.pebble.template.PebbleTemplate; -import org.apache.commons.codec.CharEncoding; import org.apache.commons.collections4.CollectionUtils; import org.apache.commons.collections4.IteratorUtils; import org.apache.commons.io.IOUtils; @@ -162,8 +159,6 @@ public class ServerTests { @Inject private ObjectMapper jsonMapper; @Inject - private XMPPServer server; - @Inject private CommandsManager commandsManager; @Inject private XMPPConnection router; @@ -687,63 +682,6 @@ public class ServerTests { .andExpect(xpath("/rss/channel/description").string("The latest messages at Juick")); } @Test - public void statusPageIsUp() throws Exception { - mockMvc.perform(get("/api/status").with(httpBasic(ugnichName, ugnichPassword))).andExpect(status().isOk()); - assertThat(server.getJid(), equalTo(jid)); - } - @Test - public void botIsUpAndProcessingResourceConstraints() throws Exception { - jdbcTemplate.execute("DELETE FROM users WHERE nick='renha'"); - int renhaId = userService.createUser("renha", "umnnbt"); - Jid from = Jid.of("renha@serverstorageisfull.tld"); - jdbcTemplate.update("INSERT INTO jids(user_id,jid,active) VALUES(?,?,?)", - renhaId, from.toEscapedString(), 1); - rocks.xmpp.core.stanza.model.Message xmppMessage = new rocks.xmpp.core.stanza.model.Message(); - xmppMessage.setType(rocks.xmpp.core.stanza.model.Message.Type.ERROR); - xmppMessage.setFrom(from); - xmppMessage.setTo(botJid); - StanzaError err = new StanzaError(StanzaError.Type.CANCEL, Condition.RESOURCE_CONSTRAINT); - xmppMessage.setError(err); - Function isActive = f -> jdbcTemplate.queryForObject("SELECT active FROM jids WHERE user_id=?", Integer.class, f) == 1; - assertThat(isActive.apply(renhaId), equalTo(true)); - ClientMessage result = router.incomingMessage(xmppMessage); - assertNull(result); - assertThat(isActive.apply(renhaId), equalTo(false)); - xmppMessage.setError(null); - xmppMessage.setType(rocks.xmpp.core.stanza.model.Message.Type.CHAT); - xmppMessage.setBody("On"); - result = router.incomingMessage(xmppMessage); - assertThat(result.getBody(), equalTo("XMPP notifications are activated")); - assertTrue(isActive.apply(renhaId)); - xmppMessage.setBody("*test test"); - result = router.incomingMessage(xmppMessage); - assertThat(result.getBody(), startsWith("New message posted")); - xmppMessage.setFrom(from); - xmppMessage.setBody("PING"); - result = router.incomingMessage(xmppMessage); - assertThat(result.getBody(), equalTo("PONG")); - int secretlySadId = userService.createUser("secretlysad", "bbk"); - xmppMessage.setTo(botJid.withLocal("secretlysad")); - xmppMessage.setBody("What's up?"); - result = router.incomingMessage(xmppMessage); - assertThat(result, is(nullValue())); - xmppMessage.setTo(botJid); - xmppMessage.setBody("@secretlysad Hey!"); - result = router.incomingMessage(xmppMessage); - assertThat(result.getBody(), is("Private message sent")); - assertThat(pmQueriesService.getPMMessages(renhaId, secretlySadId).size(), is(2)); - String xml = "@yo:\n" + - "343432434\n" + - "\n" + - "#2 http://juick.com/2juick-2343432434@yo"; - result = router.incomingMessage((rocks.xmpp.core.stanza.model.Message)server.parse(xml)); - String active = ""; - result = router.incomingMessage((rocks.xmpp.core.stanza.model.Message)server.parse(active)); - xmppMessage.setFrom(botJid); - // TODO: assert events - } - @Test public void botCommandsTests() throws Exception { assertThat(commandsManager.processCommand(AnonymousUser.INSTANCE, "PING", emptyUri).getText(), is("PONG")); // subscription commands have two lines, others have 1 @@ -1328,25 +1266,6 @@ public class ServerTests { assertThat(subscriptionService.getUsersSubscribedToComments(msg, reply).size(), is(0)); } @Test - public void xmppStatusApi() throws Exception { - Supplier getStatus = () -> { - try { - MvcResult result = mockMvc.perform(get("/api/xmpp-status").with(httpBasic(ugnichName, ugnichPassword))) - .andExpect(status().isOk()).andReturn(); - return jsonMapper.readValue(result.getResponse().getContentAsString(), XMPPStatus.class); - } catch (Exception e) { - e.printStackTrace(); - return null; - } - }; - assertThat(getStatus.get().getInbound(), is(nullValue())); - ConnectionIn test = new ConnectionIn(server, new Socket("localhost", server.getServerPort())); - test.from.add(Jid.of("test")); - server.getInConnections().clear(); - server.addConnectionIn(test); - assertThat(getStatus.get().getInbound().size(), is(1)); - } - @Test public void credentialsShouldNeverBeSerialized() throws Exception { int uid = userService.createUser("yyy", "xxxx"); User yyy = userService.getUserByUID(uid).get(); -- cgit v1.2.3 From dcf1b9e5e4312cfd685c1f3cb0f819e7ac64be75 Mon Sep 17 00:00:00 2001 From: Vitaly Takmazov Date: Mon, 31 Dec 2018 01:35:11 +0300 Subject: XMPP cleanup --- src/main/java/com/juick/server/XMPPConnection.java | 672 -------------------- src/main/java/com/juick/server/XMPPManager.java | 678 +++++++++++++++++++++ .../com/juick/server/configuration/XMPPConfig.java | 23 +- .../juick/server/xmpp/s2s/BasicXmppSession.java | 68 --- .../com/juick/server/xmpp/s2s/StanzaListener.java | 28 - .../java/com/juick/server/tests/ServerTests.java | 8 - 6 files changed, 681 insertions(+), 796 deletions(-) delete mode 100644 src/main/java/com/juick/server/XMPPConnection.java create mode 100644 src/main/java/com/juick/server/XMPPManager.java delete mode 100644 src/main/java/com/juick/server/xmpp/s2s/BasicXmppSession.java delete mode 100644 src/main/java/com/juick/server/xmpp/s2s/StanzaListener.java (limited to 'src/main/java/com/juick/server/xmpp/s2s') diff --git a/src/main/java/com/juick/server/XMPPConnection.java b/src/main/java/com/juick/server/XMPPConnection.java deleted file mode 100644 index c4816478..00000000 --- a/src/main/java/com/juick/server/XMPPConnection.java +++ /dev/null @@ -1,672 +0,0 @@ -/* - * 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 . - */ - -package com.juick.server; - -import com.juick.User; -import com.juick.formatters.PlainTextFormatter; -import com.juick.server.www.WebApp; -import com.juick.service.component.*; -import com.juick.model.CommandResult; -import com.juick.server.xmpp.iq.MessageQuery; -import com.juick.server.xmpp.s2s.BasicXmppSession; -import com.juick.service.*; -import com.juick.util.MessageUtils; -import org.apache.commons.codec.digest.DigestUtils; -import org.apache.commons.io.FilenameUtils; -import org.apache.commons.io.IOUtils; -import org.apache.commons.lang3.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.annotation.Value; -import rocks.xmpp.addr.Jid; -import rocks.xmpp.core.XmppException; -import rocks.xmpp.core.session.XmppSession; -import rocks.xmpp.core.stanza.AbstractIQHandler; -import rocks.xmpp.core.stanza.model.*; -import rocks.xmpp.core.stanza.model.client.ClientMessage; -import rocks.xmpp.core.stanza.model.client.ClientPresence; -import rocks.xmpp.core.stanza.model.errors.Condition; -import rocks.xmpp.extensions.caps.EntityCapabilitiesManager; -import rocks.xmpp.extensions.component.accept.ExternalComponent; -import rocks.xmpp.extensions.disco.ServiceDiscoveryManager; -import rocks.xmpp.extensions.disco.model.info.Identity; -import rocks.xmpp.extensions.filetransfer.FileTransfer; -import rocks.xmpp.extensions.filetransfer.FileTransferManager; -import rocks.xmpp.extensions.nick.model.Nickname; -import rocks.xmpp.extensions.oob.model.x.OobX; -import rocks.xmpp.extensions.ping.PingManager; -import rocks.xmpp.extensions.receipts.MessageDeliveryReceiptsManager; -import rocks.xmpp.extensions.vcard.temp.model.VCard; -import rocks.xmpp.extensions.version.SoftwareVersionManager; -import rocks.xmpp.extensions.version.model.SoftwareVersion; -import rocks.xmpp.util.XmppUtils; - -import javax.annotation.Nonnull; -import javax.annotation.PostConstruct; -import javax.annotation.PreDestroy; -import javax.inject.Inject; -import javax.xml.bind.JAXBException; -import javax.xml.stream.XMLStreamException; -import javax.xml.stream.XMLStreamWriter; -import java.io.IOException; -import java.io.StringWriter; -import java.net.MalformedURLException; -import java.net.URI; -import java.net.URISyntaxException; -import java.net.URL; -import java.nio.file.Path; -import java.nio.file.Paths; -import java.time.LocalDate; -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; - -/** - * @author ugnich - */ -public class XMPPConnection implements NotificationListener { - - private static final Logger logger = LoggerFactory.getLogger("com.juick.server.xmpp"); - - private ExternalComponent router; - @Inject - private CommandsManager commandsManager; - @Value("${xmppbot_jid:juick@localhost}") - private Jid jid; - @Value("${hostname:localhost}") - private String componentName; - @Value("${component_port:5347}") - private int componentPort; - @Value("${component_host:localhost}") - private String componentHost; - @Value("${xmpp_password:secret}") - private String password; - @Value("${upload_tmp_dir:#{systemEnvironment['TEMP'] ?: '/tmp'}}") - private String tmpDir; - - @Inject - private MessagesService messagesService; - @Inject - private UserService userService; - @Inject - private PMQueriesService pmQueriesService; - @Inject - private BasicXmppSession session; - @Inject - private ExecutorService executorService; - @Value("${service_user:juick}") - private String serviceUsername; - @Inject - private WebApp webApp; - - private User serviceUser; - - @PostConstruct - public void init() { - logger.info("stream router start connecting to {}", componentPort); - router = ExternalComponent.create(componentName, password, session.getConfiguration(), componentHost, componentPort); - ServiceDiscoveryManager serviceDiscoveryManager = router.getManager(ServiceDiscoveryManager.class); - serviceDiscoveryManager.addIdentity(Identity.clientBot().withName("Juick")); - EntityCapabilitiesManager entityCapabilitiesManager = router.getManager(EntityCapabilitiesManager.class); - entityCapabilitiesManager.setNode("https://juick.com/caps"); - MessageDeliveryReceiptsManager messageDeliveryReceiptsManager = router.getManager(MessageDeliveryReceiptsManager.class); - messageDeliveryReceiptsManager.setEnabled(true); - PingManager pingManager = router.getManager(PingManager.class); - pingManager.setEnabled(true); - SoftwareVersionManager softwareVersionManager = router.getManager(SoftwareVersionManager.class); - softwareVersionManager.setSoftwareVersion(new SoftwareVersion("Juick", "2.x", - System.getProperty("os.name", "generic"))); - VCard vCard = new VCard(); - vCard.setFormattedName("Juick"); - vCard.setBirthday(LocalDate.of(2008, 10, 22)); - try { - vCard.setUrl(new URL("http://juick.com/")); - vCard.setPhoto(new VCard.Image("image/png", IOUtils.toByteArray( - getClass().getClassLoader().getResource("juick.png")))); - } catch (MalformedURLException e) { - logger.error("invalid url", e); - } catch (IOException e) { - logger.warn("invalid resource", e); - } - router.addIQHandler(MessageQuery.class, iq -> { - Message warningMessage = new Message(iq.getFrom(), Message.Type.CHAT); - warningMessage.setFrom(jid); - warningMessage.setBody("Your XMPP client constantly polls us with XMPP query which is unsupported for years, please find http://juick.com/query#messages in your client code and remove that"); - router.send(warningMessage); - return iq.createError(new StanzaError(Condition.BAD_REQUEST, "Please stop this spam")); - }); - router.addIQHandler(VCard.class, new AbstractIQHandler(IQ.Type.GET) { - @Override - protected IQ processRequest(IQ iq) { - if (iq.getTo().equals(jid) || iq.getTo().asBareJid().equals(jid.asBareJid()) - || iq.getTo().asBareJid().toEscapedString().equals(jid.getDomain())) { - return iq.createResult(vCard); - } - User user = userService.getUserByName(iq.getTo().getLocal()); - if (!user.isAnonymous()) { - User info = userService.getUserInfo(user); - VCard userVCard = new VCard(); - userVCard.setFormattedName(info.getFullName()); - userVCard.setNickname(user.getName()); - try { - userVCard.setPhoto(new VCard.Image(URI.create(webApp.getAvatarUrl(user)))); - if (info.getUrl() != null) { - userVCard.setUrl(new URL(info.getUrl())); - } - } catch (MalformedURLException e) { - logger.warn("url exception", e); - } - return iq.createResult(userVCard); - } - return iq.createError(Condition.BAD_REQUEST); - } - }); - router.addInboundMessageListener(e -> { - ClientMessage result = incomingMessage(e.getMessage()); - if (result != null) { - router.send(result); - } - }); - router.addInboundIQListener(e -> { - IQ iq = e.getIQ(); - Jid jid = iq.getTo(); - if (!jid.getDomain().equals(this.jid.getDomain())) { - router.send(iq); - } - }); - FileTransferManager fileTransferManager = router.getManager(FileTransferManager.class); - fileTransferManager.addFileTransferOfferListener(e -> { - try { - List allowedTypes = new ArrayList() {{ - add("png"); - add("jpg"); - }}; - String attachmentExtension = FilenameUtils.getExtension(e.getName()).toLowerCase(); - String targetFilename = String.format("%s.%s", - DigestUtils.md5Hex(String.format("%s-%s", - e.getInitiator().toString(), e.getSessionId()).getBytes()), attachmentExtension); - if (allowedTypes.contains(attachmentExtension)) { - Path filePath = Paths.get(tmpDir, targetFilename); - FileTransfer ft = e.accept(filePath).get(); - ft.addFileTransferStatusListener(st -> { - logger.debug("{}: received {} of {}", e.getName(), st.getBytesTransferred(), e.getSize()); - if (st.getStatus().equals(FileTransfer.Status.COMPLETED)) { - logger.info("transfer completed"); - try { - Jid initiator = e.getInitiator(); - ClientMessage result = incomingMessageJuick( - userService.getUserByJID(initiator.asBareJid().toEscapedString()), initiator, - jid.getLocal(), StringUtils.defaultString(e.getDescription()).trim(), URI.create(String.format("juick://%s", targetFilename))); - if (result != null) { - router.send(result); - } - } catch (Exception e1) { - logger.error("ft error", e1); - } - - } else if (st.getStatus().equals(FileTransfer.Status.FAILED)) { - logger.info("transfer failed", ft.getException()); - Message msg = new Message(); - msg.setType(Message.Type.CHAT); - msg.setFrom(jid); - msg.setTo(e.getInitiator()); - msg.setBody("File transfer failed, please report to us"); - router.sendMessage(msg); - } else if (st.getStatus().equals(FileTransfer.Status.CANCELED)) { - logger.info("transfer cancelled"); - } - }); - ft.transfer(); - logger.info("transfer started"); - } else { - e.reject(); - logger.info("transfer rejected"); - } - } catch (IOException | InterruptedException | ExecutionException e1) { - logger.error("ft error", e1); - } - }); - router.addConnectionListener(event -> { - if (event.getType().equals(rocks.xmpp.core.session.ConnectionEvent.Type.RECONNECTION_SUCCEEDED)) { - logger.info("component connected"); - } - }); - router.addSessionStatusListener(event -> { - logger.info("event: " + event.getStatus(), event.getThrowable()); - if (event.getStatus().equals(XmppSession.Status.AUTHENTICATED)) { - logger.info("Authenticated, broadcasting..."); - broadcastPresence(null); - } - }); - router.addInboundPresenceListener(event -> { - incomingPresence(event.getPresence()); - }); - executorService.submit(() -> { - try { - router.connect(); - } catch (XmppException e) { - logger.warn("xmpp exception", e); - } - }); - serviceUser = userService.getUserByName(serviceUsername); - } - - private String stanzaToString(Stanza stanza) throws XMLStreamException, JAXBException { - StringWriter stanzaWriter = new StringWriter(); - XMLStreamWriter xmppStreamWriter = XmppUtils.createXmppStreamWriter( - router.getConfiguration().getXmlOutputFactory().createXMLStreamWriter(stanzaWriter)); - router.createMarshaller().marshal(stanza, xmppStreamWriter); - xmppStreamWriter.flush(); - xmppStreamWriter.close(); - return stanzaWriter.toString(); - } - - private void sendJuickMessage(com.juick.Message jmsg, List users) { - List jids = new ArrayList<>(); - - for (User user : users) { - jids.addAll(userService.getJIDsbyUID(user.getUid())); - } - com.juick.Message fullMsg = messagesService.getMessage(jmsg.getMid()).orElseThrow(IllegalStateException::new); - String txt = "@" + jmsg.getUser().getName() + ":" + MessageUtils.getTagsString(fullMsg) + "\n"; - String attachmentUrl = MessageUtils.attachmentUrl(fullMsg); - if (StringUtils.isNotEmpty(attachmentUrl)) { - txt += attachmentUrl + "\n"; - } - txt += StringUtils.defaultString(jmsg.getText()) + "\n\n"; - txt += "#" + jmsg.getMid() + " http://juick.com/m/" + jmsg.getMid(); - - Nickname nick = new Nickname("@" + jmsg.getUser().getName()); - - Message msg = new Message(); - msg.setFrom(jid); - msg.setBody(txt); - msg.setType(Message.Type.CHAT); - msg.setThread("juick-" + jmsg.getMid()); - msg.addExtension(jmsg); - msg.addExtension(nick); - if (StringUtils.isNotEmpty(attachmentUrl)) { - try { - OobX oob = new OobX(new URI(attachmentUrl)); - msg.addExtension(oob); - } catch (URISyntaxException e) { - logger.warn("uri exception", e); - } - } - for (String jid : jids) { - msg.setTo(Jid.of(jid)); - router.send(ClientMessage.from(msg)); - } - } - - public void sendJuickComment(com.juick.Message jmsg, List users) { - String replyQuote; - String replyTo; - - com.juick.Message replyMessage = jmsg.getReplyto() > 0 ? messagesService.getReply(jmsg.getMid(), jmsg.getReplyto()) - : messagesService.getMessage(jmsg.getMid()).orElseThrow(IllegalStateException::new); - replyTo = replyMessage.getUser().getName(); - com.juick.Message fullReply = messagesService.getReply(jmsg.getMid(), jmsg.getRid()); - replyQuote = StringUtils.defaultString(fullReply.getReplyQuote()); - - String txt = "Reply by @" + jmsg.getUser().getName() + ":\n" + replyQuote + "\n@" + replyTo + " "; - String attachmentUrl = MessageUtils.attachmentUrl(fullReply); - if (StringUtils.isNotEmpty(attachmentUrl)) { - txt += attachmentUrl + "\n"; - } - txt += StringUtils.defaultString(jmsg.getText()) + "\n\n" + "#" + jmsg.getMid() + "/" + jmsg.getRid() + " http://juick.com/m/" + jmsg.getMid() + "#" + jmsg.getRid(); - - Message msg = new Message(); - msg.setFrom(jid); - msg.setBody(txt); - msg.setType(Message.Type.CHAT); - msg.addExtension(jmsg); - for (User user : users) { - for (String jid : userService.getJIDsbyUID(user.getUid())) { - msg.setTo(Jid.of(jid)); - router.send(ClientMessage.from(msg)); - } - } - } - - @Override - public void processMessageEvent(MessageEvent event) { - com.juick.Message msg = event.getMessage(); - List subscribers = event.getUsers(); - if (msg.isService()) { - return; - } - if (MessageUtils.isPM(msg)) { - userService.getJIDsbyUID(msg.getTo().getUid()) - .forEach(userJid -> { - Message mm = new Message(); - mm.setTo(Jid.of(userJid)); - mm.setType(Message.Type.CHAT); - boolean inroster = pmQueriesService.havePMinRoster(msg.getUser().getUid(), userJid); - if (inroster) { - mm.setFrom(Jid.of(msg.getUser().getName(), "juick.com", "Juick")); - mm.setBody(msg.getText()); - } else { - mm.setFrom(jid); - mm.setBody("Private message from @" + msg.getUser().getName() + ":\n" + msg.getText()); - } - router.send(ClientMessage.from(mm)); - }); - } else if (MessageUtils.isReply(msg)) { - sendJuickComment(msg, subscribers); - } - else { - sendJuickMessage(msg, subscribers); - } - } - - private ClientMessage makeReply(Jid jidTo, String txt) { - Message reply = new Message(); - reply.setFrom(jid); - reply.setTo(jidTo); - reply.setType(Message.Type.CHAT); - reply.setBody(txt); - return ClientMessage.from(reply); - } - - @Override - public void processSubscribeEvent(SubscribeEvent subscribeEvent) { - - } - - @Override - public void processLikeEvent(LikeEvent likeEvent) { - List users = likeEvent.getSubscribers(); - com.juick.Message jmsg = likeEvent.getMessage(); - User liker = likeEvent.getUser(); - - if (!userService.isInBLAny(jmsg.getUser().getUid(), liker.getUid())) { - userService.getJIDsbyUID(jmsg.getUser().getUid()).forEach(authorJid -> { - Message xmppMessage = new Message(); - xmppMessage.setFrom(jid); - xmppMessage.setTo(Jid.of(authorJid)); - xmppMessage.setType(Message.Type.CHAT); - xmppMessage.addExtension(jmsg); - xmppMessage.setBody(String.format("%s recommended your post #%d. %s", - liker.getName(), jmsg.getMid(), PlainTextFormatter.formatUrl(jmsg))); - router.send(ClientMessage.from(xmppMessage)); - }); - } - - String txt = "Recommended by @" + liker.getName() + ":\n"; - txt += "@" + jmsg.getUser().getName() + ":" + MessageUtils.getTagsString(jmsg) + "\n"; - String attachmentUrl = MessageUtils.attachmentUrl(jmsg); - if (StringUtils.isNotEmpty(attachmentUrl)) { - txt += attachmentUrl + "\n"; - } - txt += StringUtils.defaultString(jmsg.getText()) + "\n\n"; - txt += "#" + jmsg.getMid(); - if (jmsg.getReplies() > 0) { - if (jmsg.getReplies() % 10 == 1 && jmsg.getReplies() % 100 != 11) { - txt += " (" + jmsg.getReplies() + " reply)"; - } else { - txt += " (" + jmsg.getReplies() + " replies)"; - } - } - txt += " http://juick.com/m/" + jmsg.getMid(); - - Nickname nick = new Nickname("@" + jmsg.getUser().getName()); - - Message msg = new Message(); - msg.setFrom(jid); - msg.setBody(txt); - msg.setType(Message.Type.CHAT); - msg.setThread("juick-" + jmsg.getMid()); - msg.addExtension(jmsg); - msg.addExtension(nick); - if (StringUtils.isNotEmpty(attachmentUrl)) { - try { - OobX oob = new OobX(new URI(attachmentUrl)); - msg.addExtension(oob); - } catch (URISyntaxException e) { - logger.warn("uri exception", e); - } - } - - for (User user : users) { - for (String jid : userService.getJIDsbyUID(user.getUid())) { - msg.setTo(Jid.of(jid)); - router.send(ClientMessage.from(msg)); - } - } - } - - @Override - public void processPingEvent(PingEvent pingEvent) { - userService.getJIDsbyUID(pingEvent.getPinger().getUid()) - .forEach(userJid -> { - Presence p = new Presence(Jid.of(userJid)); - p.setFrom(jid); - p.setPriority((byte) 10); - router.send(ClientPresence.from(p)); - }); - } - - @Override - public void processMessageReadEvent(MessageReadEvent messageReadEvent) { - - } - - @Override - public void processTopEvent(TopEvent topEvent) { - com.juick.Message message = topEvent.getMessage(); - try { - commandsManager.processCommand(serviceUser, String.format("! #%d", message.getMid()), URI.create(StringUtils.EMPTY)); - } catch (Exception e) { - logger.warn("XMPP error", e); - } - } - - private void incomingPresence(Presence p) { - final String username = p.getTo().getLocal(); - final boolean toJuick = username.equals(jid.getLocal()); - - if (p.getType() == null) { - Presence reply = new Presence(); - reply.setFrom(p.getTo().asBareJid()); - reply.setTo(p.getFrom().asBareJid()); - reply.setType(Presence.Type.UNSUBSCRIBE); - router.send(ClientPresence.from(reply)); - } else if (p.getType().equals(Presence.Type.PROBE)) { - int uid_to = 0; - if (!toJuick) { - uid_to = userService.getUIDbyName(username); - } else { - User visitor = userService.getUserByJID(p.getFrom().asBareJid().toEscapedString()); - if (visitor != null) { - userService.updateLastSeen(visitor); - } - } - - if (toJuick || uid_to > 0) { - Presence reply = new Presence(); - reply.setFrom(p.getTo().withResource(jid.getResource())); - reply.setTo(p.getFrom()); - reply.setPriority((byte)10); - if (!userService.getActiveJIDs().contains(p.getFrom().asBareJid().toEscapedString())) { - reply.setStatus("Send ON to enable notifications"); - } - router.send(ClientPresence.from(reply)); - } else { - Presence reply = new Presence(); - reply.setFrom(p.getTo()); - reply.setTo(p.getFrom()); - reply.setType(Presence.Type.ERROR); - reply.setId(p.getId()); - reply.setError(new StanzaError(StanzaError.Type.CANCEL, Condition.ITEM_NOT_FOUND)); - router.send(ClientPresence.from(reply)); - } - } else if (p.getType().equals(Presence.Type.SUBSCRIBE)) { - boolean canSubscribe = false; - if (toJuick) { - canSubscribe = true; - } else { - int uid_to = userService.getUIDbyName(username); - if (uid_to > 0) { - pmQueriesService.addPMinRoster(uid_to, p.getFrom().asBareJid().toEscapedString()); - canSubscribe = true; - } - } - if (canSubscribe) { - Presence reply = new Presence(); - reply.setFrom(p.getTo()); - reply.setTo(p.getFrom()); - reply.setType(Presence.Type.SUBSCRIBED); - router.send(ClientPresence.from(reply)); - - reply.setFrom(reply.getFrom().withResource(jid.getResource())); - reply.setPriority((byte) 10); - reply.setType(null); - router.send(ClientPresence.from(reply)); - } else { - Presence reply = new Presence(); - reply.setFrom(p.getTo()); - reply.setTo(p.getFrom()); - reply.setType(Presence.Type.ERROR); - reply.setId(p.getId()); - reply.setError(new StanzaError(StanzaError.Type.CANCEL, Condition.ITEM_NOT_FOUND)); - router.send(ClientPresence.from(reply)); - } - } else if (p.getType().equals(Presence.Type.UNSUBSCRIBE)) { - if (!toJuick) { - int uid_to = userService.getUIDbyName(username); - if (uid_to > 0) { - pmQueriesService.removePMinRoster(uid_to, p.getFrom().asBareJid().toEscapedString()); - } - } - - Presence reply = new Presence(); - reply.setFrom(p.getTo()); - reply.setTo(p.getFrom()); - reply.setType(Presence.Type.UNSUBSCRIBED); - router.send(ClientPresence.from(reply)); - } - } - - public ClientMessage incomingMessage(Message msg) { - ClientMessage result = null; - if (msg.getType() != null && msg.getType().equals(Message.Type.ERROR)) { - StanzaError error = msg.getError(); - if (error != null && error.getCondition().equals(Condition.RESOURCE_CONSTRAINT)) { - // offline query is full, deactivating this jid - if (userService.setActiveStatusForJID(msg.getFrom().toEscapedString(), UserService.ActiveStatus.Inactive)) { - logger.info("{} is inactive now", msg.getFrom()); - return null; - } - } - return null; - } - Jid to = msg.getTo(); - if (to.getDomain().equals(router.getDomain().toEscapedString()) || to.equals(this.jid)) { - User user_from = userService.getUserByJID(msg.getFrom().asBareJid().toEscapedString()); - if ((user_from == null || user_from.isAnonymous()) && !msg.getFrom().equals(jid)) { - String signuphash = userService.getSignUpHashByJID(msg.getFrom().asBareJid().toEscapedString()); - return makeReply(msg.getFrom(), "Для того, чтобы начать пользоваться сервисом, пожалуйста пройдите быструю регистрацию: http://juick.com/signup?type=xmpp&hash=" + signuphash + "\nЕсли у вас уже есть учетная запись на Juick, вы сможете присоединить этот JabberID к ней.\n\nTo start using Juick, please sign up: http://juick.com/signup?type=xmpp&hash=" + signuphash + "\nIf you already have an account on Juick, you will be proposed to attach this JabberID to your existing account."); - } - URI attachment = URI.create(StringUtils.EMPTY); - OobX oobX = msg.getExtension(OobX.class); - if (oobX != null) { - attachment = oobX.getUri(); - } - try { - return incomingMessageJuick(user_from, msg.getFrom(), msg.getTo().getLocal(), StringUtils.defaultString(msg.getBody()).trim(), attachment); - } catch (Exception e1) { - logger.warn("message exception", e1); - } - } else if (to.getDomain().endsWith(jid.getDomain()) && (to.getDomain().equals(jid.getDomain()) - || to.getDomain().endsWith("." + jid.getDomain()))) { - if (logger.isInfoEnabled()) { - try { - logger.info("unhandled message: {}", stanzaToString(msg)); - } catch (JAXBException | XMLStreamException ex) { - logger.error("JAXB exception", ex); - } - } - } else { - return ClientMessage.from(msg); - } - return result; - } - private ClientMessage incomingMessageJuick(User user_from, Jid from, String to, String command, @Nonnull URI attachment) { - if (StringUtils.isBlank(command) && attachment.toString().isEmpty()) { - return null; - } - - messagesService.getUnread(user_from).forEach(mid -> messagesService.setRead(user_from, mid)); - - int commandlen = command.length(); - - // COMPATIBILITY - if (commandlen > 7 && command.substring(0, 3).equalsIgnoreCase("PM ")) { - command = command.substring(3); - } - - if (!jid.getLocal().equals(to)) { - // PM - if (!StringUtils.isEmpty(command)) { - commandsManager.commandPM(user_from, null, to, command); - return null; - } - } - - try { - CommandResult result = commandsManager.processCommand(user_from, command, attachment); - if (StringUtils.isNotBlank(result.getText())) { - return makeReply(from, result.getText()); - } - } catch (Exception e) { - logger.warn("xmpp command exception", e); - return makeReply(from, "Error processing command"); - } - return null; - } - - private void broadcastPresence(Presence.Type type) { - Presence presence = new Presence(); - presence.setFrom(jid); - if (type != null) { - presence.setType(type); - } - userService.getActiveJIDs().forEach(j -> { - try { - presence.setTo(Jid.of(j)); - router.send(ClientPresence.from(presence)); - } catch (IllegalArgumentException ex) { - logger.warn("Invalid jid: {}", j, ex); - } - }); - } - - @PreDestroy - public void close() throws Exception { - broadcastPresence(Presence.Type.UNAVAILABLE); - if (router != null) { - router.close(); - } - } - - public ExternalComponent getRouter() { - return router; - } -} diff --git a/src/main/java/com/juick/server/XMPPManager.java b/src/main/java/com/juick/server/XMPPManager.java new file mode 100644 index 00000000..fb8f9711 --- /dev/null +++ b/src/main/java/com/juick/server/XMPPManager.java @@ -0,0 +1,678 @@ +/* + * 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 . + */ + +package com.juick.server; + +import com.juick.User; +import com.juick.formatters.PlainTextFormatter; +import com.juick.server.www.WebApp; +import com.juick.service.component.*; +import com.juick.model.CommandResult; +import com.juick.server.xmpp.iq.MessageQuery; +import com.juick.service.*; +import com.juick.util.MessageUtils; +import org.apache.commons.codec.digest.DigestUtils; +import org.apache.commons.io.FilenameUtils; +import org.apache.commons.io.IOUtils; +import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Value; +import rocks.xmpp.addr.Jid; +import rocks.xmpp.core.XmppException; +import rocks.xmpp.core.session.Extension; +import rocks.xmpp.core.session.XmppSession; +import rocks.xmpp.core.session.XmppSessionConfiguration; +import rocks.xmpp.core.session.debug.LogbackDebugger; +import rocks.xmpp.core.stanza.AbstractIQHandler; +import rocks.xmpp.core.stanza.model.*; +import rocks.xmpp.core.stanza.model.client.ClientMessage; +import rocks.xmpp.core.stanza.model.client.ClientPresence; +import rocks.xmpp.core.stanza.model.errors.Condition; +import rocks.xmpp.extensions.caps.EntityCapabilitiesManager; +import rocks.xmpp.extensions.component.accept.ExternalComponent; +import rocks.xmpp.extensions.disco.ServiceDiscoveryManager; +import rocks.xmpp.extensions.disco.model.info.Identity; +import rocks.xmpp.extensions.filetransfer.FileTransfer; +import rocks.xmpp.extensions.filetransfer.FileTransferManager; +import rocks.xmpp.extensions.nick.model.Nickname; +import rocks.xmpp.extensions.oob.model.x.OobX; +import rocks.xmpp.extensions.ping.PingManager; +import rocks.xmpp.extensions.receipts.MessageDeliveryReceiptsManager; +import rocks.xmpp.extensions.vcard.temp.model.VCard; +import rocks.xmpp.extensions.version.SoftwareVersionManager; +import rocks.xmpp.extensions.version.model.SoftwareVersion; +import rocks.xmpp.util.XmppUtils; + +import javax.annotation.Nonnull; +import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; +import javax.inject.Inject; +import javax.xml.bind.JAXBException; +import javax.xml.stream.XMLStreamException; +import javax.xml.stream.XMLStreamWriter; +import java.io.IOException; +import java.io.StringWriter; +import java.net.MalformedURLException; +import java.net.URI; +import java.net.URISyntaxException; +import java.net.URL; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.time.Duration; +import java.time.LocalDate; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; + +/** + * @author ugnich + */ +public class XMPPManager implements NotificationListener { + + private static final Logger logger = LoggerFactory.getLogger("com.juick.server.xmpp"); + + private ExternalComponent router; + @Inject + private CommandsManager commandsManager; + @Value("${xmppbot_jid:juick@localhost}") + private Jid jid; + @Value("${hostname:localhost}") + private String componentName; + @Value("${component_port:5347}") + private int componentPort; + @Value("${component_host:localhost}") + private String componentHost; + @Value("${xmpp_password:secret}") + private String password; + @Value("${upload_tmp_dir:#{systemEnvironment['TEMP'] ?: '/tmp'}}") + private String tmpDir; + + @Inject + private MessagesService messagesService; + @Inject + private UserService userService; + @Inject + private PMQueriesService pmQueriesService; + @Inject + private ExecutorService executorService; + @Value("${service_user:juick}") + private String serviceUsername; + @Inject + private WebApp webApp; + + private User serviceUser; + + @PostConstruct + public void init() { + logger.info("stream router start connecting to {}", componentPort); + XmppSessionConfiguration configuration = XmppSessionConfiguration.builder() + .extensions(Extension.of(com.juick.Message.class), Extension.of(MessageQuery.class)) + .debugger(LogbackDebugger.class) + .defaultResponseTimeout(Duration.ofMillis(120000)) + .build(); + router = ExternalComponent.create(componentName, password, configuration, componentHost, componentPort); + ServiceDiscoveryManager serviceDiscoveryManager = router.getManager(ServiceDiscoveryManager.class); + serviceDiscoveryManager.addIdentity(Identity.clientBot().withName("Juick")); + EntityCapabilitiesManager entityCapabilitiesManager = router.getManager(EntityCapabilitiesManager.class); + entityCapabilitiesManager.setNode("https://juick.com/caps"); + MessageDeliveryReceiptsManager messageDeliveryReceiptsManager = router.getManager(MessageDeliveryReceiptsManager.class); + messageDeliveryReceiptsManager.setEnabled(true); + PingManager pingManager = router.getManager(PingManager.class); + pingManager.setEnabled(true); + SoftwareVersionManager softwareVersionManager = router.getManager(SoftwareVersionManager.class); + softwareVersionManager.setSoftwareVersion(new SoftwareVersion("Juick", "2.x", + System.getProperty("os.name", "generic"))); + VCard vCard = new VCard(); + vCard.setFormattedName("Juick"); + vCard.setBirthday(LocalDate.of(2008, 10, 22)); + try { + vCard.setUrl(new URL("http://juick.com/")); + vCard.setPhoto(new VCard.Image("image/png", IOUtils.toByteArray( + getClass().getClassLoader().getResource("juick.png")))); + } catch (MalformedURLException e) { + logger.error("invalid url", e); + } catch (IOException e) { + logger.warn("invalid resource", e); + } + router.addIQHandler(MessageQuery.class, iq -> { + Message warningMessage = new Message(iq.getFrom(), Message.Type.CHAT); + warningMessage.setFrom(jid); + warningMessage.setBody("Your XMPP client constantly polls us with XMPP query which is unsupported for years, please find http://juick.com/query#messages in your client code and remove that"); + router.send(warningMessage); + return iq.createError(new StanzaError(Condition.BAD_REQUEST, "Please stop this spam")); + }); + router.addIQHandler(VCard.class, new AbstractIQHandler(IQ.Type.GET) { + @Override + protected IQ processRequest(IQ iq) { + if (iq.getTo().equals(jid) || iq.getTo().asBareJid().equals(jid.asBareJid()) + || iq.getTo().asBareJid().toEscapedString().equals(jid.getDomain())) { + return iq.createResult(vCard); + } + User user = userService.getUserByName(iq.getTo().getLocal()); + if (!user.isAnonymous()) { + User info = userService.getUserInfo(user); + VCard userVCard = new VCard(); + userVCard.setFormattedName(info.getFullName()); + userVCard.setNickname(user.getName()); + try { + userVCard.setPhoto(new VCard.Image(URI.create(webApp.getAvatarUrl(user)))); + if (info.getUrl() != null) { + userVCard.setUrl(new URL(info.getUrl())); + } + } catch (MalformedURLException e) { + logger.warn("url exception", e); + } + return iq.createResult(userVCard); + } + return iq.createError(Condition.BAD_REQUEST); + } + }); + router.addInboundMessageListener(e -> { + ClientMessage result = incomingMessage(e.getMessage()); + if (result != null) { + router.send(result); + } + }); + router.addInboundIQListener(e -> { + IQ iq = e.getIQ(); + Jid jid = iq.getTo(); + if (!jid.getDomain().equals(this.jid.getDomain())) { + router.send(iq); + } + }); + FileTransferManager fileTransferManager = router.getManager(FileTransferManager.class); + fileTransferManager.addFileTransferOfferListener(e -> { + try { + List allowedTypes = new ArrayList() {{ + add("png"); + add("jpg"); + }}; + String attachmentExtension = FilenameUtils.getExtension(e.getName()).toLowerCase(); + String targetFilename = String.format("%s.%s", + DigestUtils.md5Hex(String.format("%s-%s", + e.getInitiator().toString(), e.getSessionId()).getBytes()), attachmentExtension); + if (allowedTypes.contains(attachmentExtension)) { + Path filePath = Paths.get(tmpDir, targetFilename); + FileTransfer ft = e.accept(filePath).get(); + ft.addFileTransferStatusListener(st -> { + logger.debug("{}: received {} of {}", e.getName(), st.getBytesTransferred(), e.getSize()); + if (st.getStatus().equals(FileTransfer.Status.COMPLETED)) { + logger.info("transfer completed"); + try { + Jid initiator = e.getInitiator(); + ClientMessage result = incomingMessageJuick( + userService.getUserByJID(initiator.asBareJid().toEscapedString()), initiator, + jid.getLocal(), StringUtils.defaultString(e.getDescription()).trim(), URI.create(String.format("juick://%s", targetFilename))); + if (result != null) { + router.send(result); + } + } catch (Exception e1) { + logger.error("ft error", e1); + } + + } else if (st.getStatus().equals(FileTransfer.Status.FAILED)) { + logger.info("transfer failed", ft.getException()); + Message msg = new Message(); + msg.setType(Message.Type.CHAT); + msg.setFrom(jid); + msg.setTo(e.getInitiator()); + msg.setBody("File transfer failed, please report to us"); + router.sendMessage(msg); + } else if (st.getStatus().equals(FileTransfer.Status.CANCELED)) { + logger.info("transfer cancelled"); + } + }); + ft.transfer(); + logger.info("transfer started"); + } else { + e.reject(); + logger.info("transfer rejected"); + } + } catch (IOException | InterruptedException | ExecutionException e1) { + logger.error("ft error", e1); + } + }); + router.addConnectionListener(event -> { + if (event.getType().equals(rocks.xmpp.core.session.ConnectionEvent.Type.RECONNECTION_SUCCEEDED)) { + logger.info("component connected"); + } + }); + router.addSessionStatusListener(event -> { + logger.info("event: " + event.getStatus(), event.getThrowable()); + if (event.getStatus().equals(XmppSession.Status.AUTHENTICATED)) { + logger.info("Authenticated, broadcasting..."); + broadcastPresence(null); + } + }); + router.addInboundPresenceListener(event -> { + incomingPresence(event.getPresence()); + }); + executorService.submit(() -> { + try { + router.connect(); + } catch (XmppException e) { + logger.warn("xmpp exception", e); + } + }); + serviceUser = userService.getUserByName(serviceUsername); + } + + private String stanzaToString(Stanza stanza) throws XMLStreamException, JAXBException { + StringWriter stanzaWriter = new StringWriter(); + XMLStreamWriter xmppStreamWriter = XmppUtils.createXmppStreamWriter( + router.getConfiguration().getXmlOutputFactory().createXMLStreamWriter(stanzaWriter)); + router.createMarshaller().marshal(stanza, xmppStreamWriter); + xmppStreamWriter.flush(); + xmppStreamWriter.close(); + return stanzaWriter.toString(); + } + + private void sendJuickMessage(com.juick.Message jmsg, List users) { + List jids = new ArrayList<>(); + + for (User user : users) { + jids.addAll(userService.getJIDsbyUID(user.getUid())); + } + com.juick.Message fullMsg = messagesService.getMessage(jmsg.getMid()).orElseThrow(IllegalStateException::new); + String txt = "@" + jmsg.getUser().getName() + ":" + MessageUtils.getTagsString(fullMsg) + "\n"; + String attachmentUrl = MessageUtils.attachmentUrl(fullMsg); + if (StringUtils.isNotEmpty(attachmentUrl)) { + txt += attachmentUrl + "\n"; + } + txt += StringUtils.defaultString(jmsg.getText()) + "\n\n"; + txt += "#" + jmsg.getMid() + " http://juick.com/m/" + jmsg.getMid(); + + Nickname nick = new Nickname("@" + jmsg.getUser().getName()); + + Message msg = new Message(); + msg.setFrom(jid); + msg.setBody(txt); + msg.setType(Message.Type.CHAT); + msg.setThread("juick-" + jmsg.getMid()); + msg.addExtension(jmsg); + msg.addExtension(nick); + if (StringUtils.isNotEmpty(attachmentUrl)) { + try { + OobX oob = new OobX(new URI(attachmentUrl)); + msg.addExtension(oob); + } catch (URISyntaxException e) { + logger.warn("uri exception", e); + } + } + for (String jid : jids) { + msg.setTo(Jid.of(jid)); + router.send(ClientMessage.from(msg)); + } + } + + public void sendJuickComment(com.juick.Message jmsg, List users) { + String replyQuote; + String replyTo; + + com.juick.Message replyMessage = jmsg.getReplyto() > 0 ? messagesService.getReply(jmsg.getMid(), jmsg.getReplyto()) + : messagesService.getMessage(jmsg.getMid()).orElseThrow(IllegalStateException::new); + replyTo = replyMessage.getUser().getName(); + com.juick.Message fullReply = messagesService.getReply(jmsg.getMid(), jmsg.getRid()); + replyQuote = StringUtils.defaultString(fullReply.getReplyQuote()); + + String txt = "Reply by @" + jmsg.getUser().getName() + ":\n" + replyQuote + "\n@" + replyTo + " "; + String attachmentUrl = MessageUtils.attachmentUrl(fullReply); + if (StringUtils.isNotEmpty(attachmentUrl)) { + txt += attachmentUrl + "\n"; + } + txt += StringUtils.defaultString(jmsg.getText()) + "\n\n" + "#" + jmsg.getMid() + "/" + jmsg.getRid() + " http://juick.com/m/" + jmsg.getMid() + "#" + jmsg.getRid(); + + Message msg = new Message(); + msg.setFrom(jid); + msg.setBody(txt); + msg.setType(Message.Type.CHAT); + msg.addExtension(jmsg); + for (User user : users) { + for (String jid : userService.getJIDsbyUID(user.getUid())) { + msg.setTo(Jid.of(jid)); + router.send(ClientMessage.from(msg)); + } + } + } + + @Override + public void processMessageEvent(MessageEvent event) { + com.juick.Message msg = event.getMessage(); + List subscribers = event.getUsers(); + if (msg.isService()) { + return; + } + if (MessageUtils.isPM(msg)) { + userService.getJIDsbyUID(msg.getTo().getUid()) + .forEach(userJid -> { + Message mm = new Message(); + mm.setTo(Jid.of(userJid)); + mm.setType(Message.Type.CHAT); + boolean inroster = pmQueriesService.havePMinRoster(msg.getUser().getUid(), userJid); + if (inroster) { + mm.setFrom(Jid.of(msg.getUser().getName(), "juick.com", "Juick")); + mm.setBody(msg.getText()); + } else { + mm.setFrom(jid); + mm.setBody("Private message from @" + msg.getUser().getName() + ":\n" + msg.getText()); + } + router.send(ClientMessage.from(mm)); + }); + } else if (MessageUtils.isReply(msg)) { + sendJuickComment(msg, subscribers); + } + else { + sendJuickMessage(msg, subscribers); + } + } + + private ClientMessage makeReply(Jid jidTo, String txt) { + Message reply = new Message(); + reply.setFrom(jid); + reply.setTo(jidTo); + reply.setType(Message.Type.CHAT); + reply.setBody(txt); + return ClientMessage.from(reply); + } + + @Override + public void processSubscribeEvent(SubscribeEvent subscribeEvent) { + + } + + @Override + public void processLikeEvent(LikeEvent likeEvent) { + List users = likeEvent.getSubscribers(); + com.juick.Message jmsg = likeEvent.getMessage(); + User liker = likeEvent.getUser(); + + if (!userService.isInBLAny(jmsg.getUser().getUid(), liker.getUid())) { + userService.getJIDsbyUID(jmsg.getUser().getUid()).forEach(authorJid -> { + Message xmppMessage = new Message(); + xmppMessage.setFrom(jid); + xmppMessage.setTo(Jid.of(authorJid)); + xmppMessage.setType(Message.Type.CHAT); + xmppMessage.addExtension(jmsg); + xmppMessage.setBody(String.format("%s recommended your post #%d. %s", + liker.getName(), jmsg.getMid(), PlainTextFormatter.formatUrl(jmsg))); + router.send(ClientMessage.from(xmppMessage)); + }); + } + + String txt = "Recommended by @" + liker.getName() + ":\n"; + txt += "@" + jmsg.getUser().getName() + ":" + MessageUtils.getTagsString(jmsg) + "\n"; + String attachmentUrl = MessageUtils.attachmentUrl(jmsg); + if (StringUtils.isNotEmpty(attachmentUrl)) { + txt += attachmentUrl + "\n"; + } + txt += StringUtils.defaultString(jmsg.getText()) + "\n\n"; + txt += "#" + jmsg.getMid(); + if (jmsg.getReplies() > 0) { + if (jmsg.getReplies() % 10 == 1 && jmsg.getReplies() % 100 != 11) { + txt += " (" + jmsg.getReplies() + " reply)"; + } else { + txt += " (" + jmsg.getReplies() + " replies)"; + } + } + txt += " http://juick.com/m/" + jmsg.getMid(); + + Nickname nick = new Nickname("@" + jmsg.getUser().getName()); + + Message msg = new Message(); + msg.setFrom(jid); + msg.setBody(txt); + msg.setType(Message.Type.CHAT); + msg.setThread("juick-" + jmsg.getMid()); + msg.addExtension(jmsg); + msg.addExtension(nick); + if (StringUtils.isNotEmpty(attachmentUrl)) { + try { + OobX oob = new OobX(new URI(attachmentUrl)); + msg.addExtension(oob); + } catch (URISyntaxException e) { + logger.warn("uri exception", e); + } + } + + for (User user : users) { + for (String jid : userService.getJIDsbyUID(user.getUid())) { + msg.setTo(Jid.of(jid)); + router.send(ClientMessage.from(msg)); + } + } + } + + @Override + public void processPingEvent(PingEvent pingEvent) { + userService.getJIDsbyUID(pingEvent.getPinger().getUid()) + .forEach(userJid -> { + Presence p = new Presence(Jid.of(userJid)); + p.setFrom(jid); + p.setPriority((byte) 10); + router.send(ClientPresence.from(p)); + }); + } + + @Override + public void processMessageReadEvent(MessageReadEvent messageReadEvent) { + + } + + @Override + public void processTopEvent(TopEvent topEvent) { + com.juick.Message message = topEvent.getMessage(); + try { + commandsManager.processCommand(serviceUser, String.format("! #%d", message.getMid()), URI.create(StringUtils.EMPTY)); + } catch (Exception e) { + logger.warn("XMPP error", e); + } + } + + private void incomingPresence(Presence p) { + final String username = p.getTo().getLocal(); + final boolean toJuick = username.equals(jid.getLocal()); + + if (p.getType() == null) { + Presence reply = new Presence(); + reply.setFrom(p.getTo().asBareJid()); + reply.setTo(p.getFrom().asBareJid()); + reply.setType(Presence.Type.UNSUBSCRIBE); + router.send(ClientPresence.from(reply)); + } else if (p.getType().equals(Presence.Type.PROBE)) { + int uid_to = 0; + if (!toJuick) { + uid_to = userService.getUIDbyName(username); + } else { + User visitor = userService.getUserByJID(p.getFrom().asBareJid().toEscapedString()); + if (visitor != null) { + userService.updateLastSeen(visitor); + } + } + + if (toJuick || uid_to > 0) { + Presence reply = new Presence(); + reply.setFrom(p.getTo().withResource(jid.getResource())); + reply.setTo(p.getFrom()); + reply.setPriority((byte)10); + if (!userService.getActiveJIDs().contains(p.getFrom().asBareJid().toEscapedString())) { + reply.setStatus("Send ON to enable notifications"); + } + router.send(ClientPresence.from(reply)); + } else { + Presence reply = new Presence(); + reply.setFrom(p.getTo()); + reply.setTo(p.getFrom()); + reply.setType(Presence.Type.ERROR); + reply.setId(p.getId()); + reply.setError(new StanzaError(StanzaError.Type.CANCEL, Condition.ITEM_NOT_FOUND)); + router.send(ClientPresence.from(reply)); + } + } else if (p.getType().equals(Presence.Type.SUBSCRIBE)) { + boolean canSubscribe = false; + if (toJuick) { + canSubscribe = true; + } else { + int uid_to = userService.getUIDbyName(username); + if (uid_to > 0) { + pmQueriesService.addPMinRoster(uid_to, p.getFrom().asBareJid().toEscapedString()); + canSubscribe = true; + } + } + if (canSubscribe) { + Presence reply = new Presence(); + reply.setFrom(p.getTo()); + reply.setTo(p.getFrom()); + reply.setType(Presence.Type.SUBSCRIBED); + router.send(ClientPresence.from(reply)); + + reply.setFrom(reply.getFrom().withResource(jid.getResource())); + reply.setPriority((byte) 10); + reply.setType(null); + router.send(ClientPresence.from(reply)); + } else { + Presence reply = new Presence(); + reply.setFrom(p.getTo()); + reply.setTo(p.getFrom()); + reply.setType(Presence.Type.ERROR); + reply.setId(p.getId()); + reply.setError(new StanzaError(StanzaError.Type.CANCEL, Condition.ITEM_NOT_FOUND)); + router.send(ClientPresence.from(reply)); + } + } else if (p.getType().equals(Presence.Type.UNSUBSCRIBE)) { + if (!toJuick) { + int uid_to = userService.getUIDbyName(username); + if (uid_to > 0) { + pmQueriesService.removePMinRoster(uid_to, p.getFrom().asBareJid().toEscapedString()); + } + } + + Presence reply = new Presence(); + reply.setFrom(p.getTo()); + reply.setTo(p.getFrom()); + reply.setType(Presence.Type.UNSUBSCRIBED); + router.send(ClientPresence.from(reply)); + } + } + + public ClientMessage incomingMessage(Message msg) { + ClientMessage result = null; + if (msg.getType() != null && msg.getType().equals(Message.Type.ERROR)) { + StanzaError error = msg.getError(); + if (error != null && error.getCondition().equals(Condition.RESOURCE_CONSTRAINT)) { + // offline query is full, deactivating this jid + if (userService.setActiveStatusForJID(msg.getFrom().toEscapedString(), UserService.ActiveStatus.Inactive)) { + logger.info("{} is inactive now", msg.getFrom()); + return null; + } + } + return null; + } + Jid to = msg.getTo(); + if (to.getDomain().equals(router.getDomain().toEscapedString()) || to.equals(this.jid)) { + User user_from = userService.getUserByJID(msg.getFrom().asBareJid().toEscapedString()); + if ((user_from == null || user_from.isAnonymous()) && !msg.getFrom().equals(jid)) { + String signuphash = userService.getSignUpHashByJID(msg.getFrom().asBareJid().toEscapedString()); + return makeReply(msg.getFrom(), "Для того, чтобы начать пользоваться сервисом, пожалуйста пройдите быструю регистрацию: http://juick.com/signup?type=xmpp&hash=" + signuphash + "\nЕсли у вас уже есть учетная запись на Juick, вы сможете присоединить этот JabberID к ней.\n\nTo start using Juick, please sign up: http://juick.com/signup?type=xmpp&hash=" + signuphash + "\nIf you already have an account on Juick, you will be proposed to attach this JabberID to your existing account."); + } + URI attachment = URI.create(StringUtils.EMPTY); + OobX oobX = msg.getExtension(OobX.class); + if (oobX != null) { + attachment = oobX.getUri(); + } + try { + return incomingMessageJuick(user_from, msg.getFrom(), msg.getTo().getLocal(), StringUtils.defaultString(msg.getBody()).trim(), attachment); + } catch (Exception e1) { + logger.warn("message exception", e1); + } + } else if (to.getDomain().endsWith(jid.getDomain()) && (to.getDomain().equals(jid.getDomain()) + || to.getDomain().endsWith("." + jid.getDomain()))) { + if (logger.isInfoEnabled()) { + try { + logger.info("unhandled message: {}", stanzaToString(msg)); + } catch (JAXBException | XMLStreamException ex) { + logger.error("JAXB exception", ex); + } + } + } else { + return ClientMessage.from(msg); + } + return result; + } + private ClientMessage incomingMessageJuick(User user_from, Jid from, String to, String command, @Nonnull URI attachment) { + if (StringUtils.isBlank(command) && attachment.toString().isEmpty()) { + return null; + } + + messagesService.getUnread(user_from).forEach(mid -> messagesService.setRead(user_from, mid)); + + int commandlen = command.length(); + + // COMPATIBILITY + if (commandlen > 7 && command.substring(0, 3).equalsIgnoreCase("PM ")) { + command = command.substring(3); + } + + if (!jid.getLocal().equals(to)) { + // PM + if (!StringUtils.isEmpty(command)) { + commandsManager.commandPM(user_from, null, to, command); + return null; + } + } + + try { + CommandResult result = commandsManager.processCommand(user_from, command, attachment); + if (StringUtils.isNotBlank(result.getText())) { + return makeReply(from, result.getText()); + } + } catch (Exception e) { + logger.warn("xmpp command exception", e); + return makeReply(from, "Error processing command"); + } + return null; + } + + private void broadcastPresence(Presence.Type type) { + Presence presence = new Presence(); + presence.setFrom(jid); + if (type != null) { + presence.setType(type); + } + userService.getActiveJIDs().forEach(j -> { + try { + presence.setTo(Jid.of(j)); + router.send(ClientPresence.from(presence)); + } catch (IllegalArgumentException ex) { + logger.warn("Invalid jid: {}", j, ex); + } + }); + } + + @PreDestroy + public void close() throws Exception { + broadcastPresence(Presence.Type.UNAVAILABLE); + if (router != null) { + router.close(); + } + } + + public ExternalComponent getRouter() { + return router; + } +} diff --git a/src/main/java/com/juick/server/configuration/XMPPConfig.java b/src/main/java/com/juick/server/configuration/XMPPConfig.java index ffacf4ef..f9b6f092 100644 --- a/src/main/java/com/juick/server/configuration/XMPPConfig.java +++ b/src/main/java/com/juick/server/configuration/XMPPConfig.java @@ -1,21 +1,13 @@ package com.juick.server.configuration; -import com.juick.server.XMPPConnection; +import com.juick.server.XMPPManager; import com.juick.server.xmpp.JidConverter; -import com.juick.server.xmpp.iq.MessageQuery; -import com.juick.server.xmpp.s2s.BasicXmppSession; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.context.annotation.DependsOn; import org.springframework.core.convert.ConversionService; import org.springframework.format.support.DefaultFormattingConversionService; -import rocks.xmpp.core.session.Extension; -import rocks.xmpp.core.session.XmppSessionConfiguration; -import rocks.xmpp.core.session.debug.LogbackDebugger; - -import java.time.Duration; @Configuration @ConditionalOnProperty("xmppbot_jid") @@ -23,22 +15,13 @@ public class XMPPConfig { @Value("${hostname:localhost}") private String hostname; @Bean - public BasicXmppSession session() { - XmppSessionConfiguration configuration = XmppSessionConfiguration.builder() - .extensions(Extension.of(com.juick.Message.class), Extension.of(MessageQuery.class)) - .debugger(LogbackDebugger.class) - .defaultResponseTimeout(Duration.ofMillis(120000)) - .build(); - return BasicXmppSession.create(hostname, configuration); - } - @Bean public static ConversionService conversionService() { DefaultFormattingConversionService cs = new DefaultFormattingConversionService(); cs.addConverter(new JidConverter()); return cs; } @Bean - public XMPPConnection xmppConnection() { - return new XMPPConnection(); + public XMPPManager xmppConnection() { + return new XMPPManager(); } } diff --git a/src/main/java/com/juick/server/xmpp/s2s/BasicXmppSession.java b/src/main/java/com/juick/server/xmpp/s2s/BasicXmppSession.java deleted file mode 100644 index ae28f827..00000000 --- a/src/main/java/com/juick/server/xmpp/s2s/BasicXmppSession.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * 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 . - */ - -package com.juick.server.xmpp.s2s; - -import rocks.xmpp.addr.Jid; -import rocks.xmpp.core.XmppException; -import rocks.xmpp.core.session.XmppSession; -import rocks.xmpp.core.session.XmppSessionConfiguration; -import rocks.xmpp.core.stanza.model.IQ; -import rocks.xmpp.core.stanza.model.Message; -import rocks.xmpp.core.stanza.model.Presence; -import rocks.xmpp.core.stanza.model.server.ServerIQ; -import rocks.xmpp.core.stanza.model.server.ServerMessage; -import rocks.xmpp.core.stanza.model.server.ServerPresence; -import rocks.xmpp.core.stream.model.StreamElement; - -/** - * Created by vitalyster on 06.02.2017. - */ -public class BasicXmppSession extends XmppSession { - protected BasicXmppSession(String xmppServiceDomain, XmppSessionConfiguration configuration) { - super(xmppServiceDomain, configuration); - } - - public static BasicXmppSession create(String xmppServiceDomain, XmppSessionConfiguration configuration) { - BasicXmppSession session = new BasicXmppSession(xmppServiceDomain, configuration); - notifyCreationListeners(session); - return session; - } - - @Override - public void connect(Jid from) throws XmppException { - - } - - @Override - public Jid getConnectedResource() { - return null; - } - - @Override - protected StreamElement prepareElement(StreamElement element) { - if (element instanceof Message) { - element = ServerMessage.from((Message) element); - } else if (element instanceof Presence) { - element = ServerPresence.from((Presence) element); - } else if (element instanceof IQ) { - element = ServerIQ.from((IQ) element); - } - - return element; - } -} diff --git a/src/main/java/com/juick/server/xmpp/s2s/StanzaListener.java b/src/main/java/com/juick/server/xmpp/s2s/StanzaListener.java deleted file mode 100644 index 6932298f..00000000 --- a/src/main/java/com/juick/server/xmpp/s2s/StanzaListener.java +++ /dev/null @@ -1,28 +0,0 @@ -/* - * 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 . - */ - -package com.juick.server.xmpp.s2s; - - -import rocks.xmpp.core.stanza.model.Stanza; - -/** - * Created by vitalyster on 07.12.2016. - */ -public interface StanzaListener { - void stanzaReceived(Stanza xmlValue); -} diff --git a/src/test/java/com/juick/server/tests/ServerTests.java b/src/test/java/com/juick/server/tests/ServerTests.java index 620e210c..fa2e2ce9 100644 --- a/src/test/java/com/juick/server/tests/ServerTests.java +++ b/src/test/java/com/juick/server/tests/ServerTests.java @@ -87,9 +87,6 @@ import rocks.xmpp.addr.Jid; import rocks.xmpp.core.session.Extension; import rocks.xmpp.core.session.XmppSession; import rocks.xmpp.core.session.XmppSessionConfiguration; -import rocks.xmpp.core.stanza.model.StanzaError; -import rocks.xmpp.core.stanza.model.client.ClientMessage; -import rocks.xmpp.core.stanza.model.errors.Condition; import javax.inject.Inject; import javax.servlet.http.Cookie; @@ -101,7 +98,6 @@ import javax.xml.parsers.DocumentBuilder; import javax.xml.parsers.DocumentBuilderFactory; import javax.xml.parsers.ParserConfigurationException; import java.io.*; -import java.net.Socket; import java.net.URI; import java.net.URISyntaxException; import java.nio.charset.StandardCharsets; @@ -113,8 +109,6 @@ import java.sql.Timestamp; import java.time.Instant; import java.util.*; import java.util.function.BiFunction; -import java.util.function.Function; -import java.util.function.Supplier; import java.util.stream.Collectors; import java.util.stream.IntStream; import java.util.stream.StreamSupport; @@ -161,8 +155,6 @@ public class ServerTests { @Inject private CommandsManager commandsManager; @Inject - private XMPPConnection router; - @Inject private SubscriptionService subscriptionService; @Inject private PrivacyQueriesService privacyQueriesService; -- cgit v1.2.3