From 362250147190e4ba4587a1eea10f064c93238009 Mon Sep 17 00:00:00 2001
From: zzz <zzz@mail.i2p>
Date: Sun, 2 Feb 2014 16:47:29 +0000
Subject: [PATCH] * Streaming: - Set ports on many packets that were
missing them - Use connection throttling methods on pings too (ticket
#1142) - Add methods to set ports on pings - Argument checking on ping
methods
---
.../client/streaming/I2PSocketManager.java | 21 ++++-
.../i2p/client/streaming/impl/Connection.java | 2 +
.../impl/ConnectionDataReceiver.java | 4 +-
.../streaming/impl/ConnectionManager.java | 81 +++++++++++++++----
.../streaming/impl/I2PSocketManagerFull.java | 32 +++++++-
.../client/streaming/impl/PacketHandler.java | 16 ++--
6 files changed, 130 insertions(+), 26 deletions(-)
diff --git a/apps/ministreaming/java/src/net/i2p/client/streaming/I2PSocketManager.java b/apps/ministreaming/java/src/net/i2p/client/streaming/I2PSocketManager.java
index 8a8aa4e4db..6ba93bf3b7 100644
--- a/apps/ministreaming/java/src/net/i2p/client/streaming/I2PSocketManager.java
+++ b/apps/ministreaming/java/src/net/i2p/client/streaming/I2PSocketManager.java
@@ -126,12 +126,31 @@ public interface I2PSocketManager {
* Ping the specified peer, returning true if they replied to the ping within
* the timeout specified, false otherwise. This call blocks.
*
+ * Uses the ports from the default options.
+ *
* @param peer Destination to ping
- * @param timeoutMs timeout in ms
+ * @param timeoutMs timeout in ms, greater than zero
+ * @throws IllegalArgumentException
* @return success or failure
*/
public boolean ping(Destination peer, long timeoutMs);
+ /**
+ * Ping the specified peer, returning true if they replied to the ping within
+ * the timeout specified, false otherwise. This call blocks.
+ *
+ * Uses the ports specified.
+ *
+ * @param peer Destination to ping
+ * @param localPort 0 - 65535
+ * @param remotePort 0 - 65535
+ * @param timeoutMs timeout in ms, greater than zero
+ * @return success or failure
+ * @throws IllegalArgumentException
+ * @since 0.9.12
+ */
+ public boolean ping(Destination peer, int localPort, int remotePort, long timeoutMs);
+
public String getName();
public void setName(String name);
diff --git a/apps/streaming/java/src/net/i2p/client/streaming/impl/Connection.java b/apps/streaming/java/src/net/i2p/client/streaming/impl/Connection.java
index 47cfd6a53d..8cec3cad8c 100644
--- a/apps/streaming/java/src/net/i2p/client/streaming/impl/Connection.java
+++ b/apps/streaming/java/src/net/i2p/client/streaming/impl/Connection.java
@@ -307,6 +307,8 @@ class Connection {
reply.setSendStreamId(_sendStreamId);
reply.setReceiveStreamId(_receiveStreamId);
reply.setOptionalFrom(_connectionManager.getSession().getMyDestination());
+ reply.setLocalPort(_localPort);
+ reply.setRemotePort(_remotePort);
// this just sends the packet - no retries or whatnot
if (_outboundQueue.enqueue(reply)) {
_unackedPacketsReceived = 0;
diff --git a/apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionDataReceiver.java b/apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionDataReceiver.java
index dbf546ad88..8568e7c69b 100644
--- a/apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionDataReceiver.java
+++ b/apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionDataReceiver.java
@@ -206,9 +206,9 @@ class ConnectionDataReceiver implements MessageOutputStream.DataReceiver {
packet.setFlag(Packet.FLAG_SYNCHRONIZE);
packet.setOptionalFrom(con.getSession().getMyDestination());
packet.setOptionalMaxSize(con.getOptions().getMaxMessageSize());
- packet.setLocalPort(con.getLocalPort());
- packet.setRemotePort(con.getPort());
}
+ packet.setLocalPort(con.getLocalPort());
+ packet.setRemotePort(con.getPort());
if (con.getSendStreamId() == Packet.STREAM_ID_UNKNOWN) {
packet.setFlag(Packet.FLAG_NO_ACK);
}
diff --git a/apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionManager.java b/apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionManager.java
index 435a09ce42..3e09c79978 100644
--- a/apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionManager.java
+++ b/apps/streaming/java/src/net/i2p/client/streaming/impl/ConnectionManager.java
@@ -262,6 +262,8 @@ class ConnectionManager {
reply.setSendStreamId(synPacket.getReceiveStreamId());
reply.setReceiveStreamId(0);
reply.setOptionalFrom(_session.getMyDestination());
+ reply.setLocalPort(synPacket.getLocalPort());
+ reply.setRemotePort(synPacket.getRemotePort());
// this just sends the packet - no retries or whatnot
_outboundQueue.enqueue(reply);
return null;
@@ -284,6 +286,44 @@ class ConnectionManager {
return con;
}
+ /**
+ * Process a ping by checking for throttling, etc., then sending a pong.
+ *
+ * @param con null if unknown
+ * @param ping Ping packet to process, must have From and Sig fields,
+ * with signature already verified, only if answerPings() returned true
+ * @return true if we sent a pong
+ * @since 0.9.12 from PacketHandler.receivePing()
+ */
+ public boolean receivePing(Connection con, Packet ping) {
+ Destination dest = ping.getOptionalFrom();
+ if (dest == null)
+ return false;
+ if (con == null) {
+ // Use the same throttling as for connections
+ String why = shouldRejectConnection(ping);
+ if (why != null) {
+ if ((!_defaultOptions.getDisableRejectLogging()) || _log.shouldLog(Log.WARN))
+ _log.logAlways(Log.WARN, "Dropping ping since peer is " + why + ": " + dest.calculateHash());
+ return false;
+ }
+ } else {
+ // in-connection ping to a 3rd party ???
+ if (!dest.equals(con.getRemotePeer())) {
+ _log.logAlways(Log.WARN, "Dropping ping from " + con.getRemotePeer().calculateHash() +
+ " to " + dest.calculateHash());
+ return false;
+ }
+ }
+ PacketLocal pong = new PacketLocal(_context, dest);
+ pong.setFlag(Packet.FLAG_ECHO | Packet.FLAG_NO_ACK);
+ pong.setReceiveStreamId(ping.getSendStreamId());
+ pong.setLocalPort(ping.getLocalPort());
+ pong.setRemotePort(ping.getRemotePort());
+ _outboundQueue.enqueue(pong);
+ return true;
+ }
+
private static final long DEFAULT_STREAM_DELAY_MAX = 10*1000;
/**
@@ -572,24 +612,34 @@ class ConnectionManager {
return new HashSet<Connection>(_connectionByInboundId.values());
}
- /** blocking */
- public boolean ping(Destination peer, long timeoutMs) {
- return ping(peer, timeoutMs, true, null);
- }
- public boolean ping(Destination peer, long timeoutMs, boolean blocking) {
- return ping(peer, timeoutMs, blocking, null);
+ /**
+ * blocking
+ *
+ * @param timeoutMs greater than zero
+ * @return true if pong received
+ * @since 0.9.12 added port args
+ */
+ public boolean ping(Destination peer, int fromPort, int toPort, long timeoutMs) {
+ return ping(peer, fromPort, toPort, timeoutMs, true, null);
}
/**
- * @deprecated I2PSession ignores tags, use non-tag variant
- * @param keyToUse ignored
- * @param tagsToSend ignored
+ * @param timeoutMs greater than zero
+ * @return true if blocking and pong received
+ * @since 0.9.12 added port args
*/
- public boolean ping(Destination peer, long timeoutMs, boolean blocking, SessionKey keyToUse, Set<?> tagsToSend, PingNotifier notifier) {
- return ping(peer, timeoutMs, blocking, notifier);
+ public boolean ping(Destination peer, int fromPort, int toPort, long timeoutMs, boolean blocking) {
+ return ping(peer, fromPort, toPort, timeoutMs, blocking, null);
}
- public boolean ping(Destination peer, long timeoutMs, boolean blocking, PingNotifier notifier) {
+ /**
+ * @param timeoutMs greater than zero
+ * @param notifier may be null
+ * @return true if blocking and pong received
+ * @since 0.9.12 added port args
+ */
+ public boolean ping(Destination peer, int fromPort, int toPort, long timeoutMs,
+ boolean blocking, PingNotifier notifier) {
Long id = Long.valueOf(_context.random().nextLong(Packet.MAX_STREAM_ID-1)+1);
PacketLocal packet = new PacketLocal(_context, peer);
packet.setSendStreamId(id.longValue());
@@ -597,13 +647,15 @@ class ConnectionManager {
Packet.FLAG_NO_ACK |
Packet.FLAG_SIGNATURE_INCLUDED);
packet.setOptionalFrom(_session.getMyDestination());
+ packet.setLocalPort(fromPort);
+ packet.setRemotePort(toPort);
//if ( (keyToUse != null) && (tagsToSend != null) ) {
// packet.setKeyUsed(keyToUse);
// packet.setTagsSent(tagsToSend);
//}
if (_log.shouldLog(Log.INFO)) {
- _log.info(String.format("about to ping %s timeout=%d blocking=%b",
- peer,timeoutMs,blocking));
+ _log.info(String.format("about to ping %s port %d from port %d timeout=%d blocking=%b",
+ peer.calculateHash().toString(), toPort, fromPort, timeoutMs, blocking));
}
@@ -658,6 +710,7 @@ class ConnectionManager {
private boolean _ponged;
private final PingNotifier _notifier;
+ /** @param notifier may be null */
public PingRequest(PingNotifier notifier) {
_notifier = notifier;
}
diff --git a/apps/streaming/java/src/net/i2p/client/streaming/impl/I2PSocketManagerFull.java b/apps/streaming/java/src/net/i2p/client/streaming/impl/I2PSocketManagerFull.java
index 757eb37f42..435207b6c3 100644
--- a/apps/streaming/java/src/net/i2p/client/streaming/impl/I2PSocketManagerFull.java
+++ b/apps/streaming/java/src/net/i2p/client/streaming/impl/I2PSocketManagerFull.java
@@ -148,13 +148,41 @@ public class I2PSocketManagerFull implements I2PSocketManager {
* Ping the specified peer, returning true if they replied to the ping within
* the timeout specified, false otherwise. This call blocks.
*
+ * Uses the ports from the default options.
*
* @param peer
- * @param timeoutMs
+ * @param timeoutMs timeout in ms, greater than zero
* @return true on success, false on failure
+ * @throws IllegalArgumentException
*/
public boolean ping(Destination peer, long timeoutMs) {
- return _connectionManager.ping(peer, timeoutMs);
+ if (timeoutMs <= 0)
+ throw new IllegalArgumentException("bad timeout");
+ return _connectionManager.ping(peer, _defaultOptions.getLocalPort(),
+ _defaultOptions.getPort(), timeoutMs);
+ }
+
+ /**
+ * Ping the specified peer, returning true if they replied to the ping within
+ * the timeout specified, false otherwise. This call blocks.
+ *
+ * Uses the ports specified.
+ *
+ * @param peer Destination to ping
+ * @param localPort 0 - 65535
+ * @param remotePort 0 - 65535
+ * @param timeoutMs timeout in ms, greater than zero
+ * @return success or failure
+ * @throws IllegalArgumentException
+ * @since 0.9.12
+ */
+ public boolean ping(Destination peer, int localPort, int remotePort, long timeoutMs) {
+ if (localPort < 0 || localPort > 65535 ||
+ remotePort < 0 || remotePort > 65535)
+ throw new IllegalArgumentException("bad port");
+ if (timeoutMs <= 0)
+ throw new IllegalArgumentException("bad timeout");
+ return _connectionManager.ping(peer, localPort, remotePort, timeoutMs);
}
/**
diff --git a/apps/streaming/java/src/net/i2p/client/streaming/impl/PacketHandler.java b/apps/streaming/java/src/net/i2p/client/streaming/impl/PacketHandler.java
index c40afe453a..1deef1f226 100644
--- a/apps/streaming/java/src/net/i2p/client/streaming/impl/PacketHandler.java
+++ b/apps/streaming/java/src/net/i2p/client/streaming/impl/PacketHandler.java
@@ -136,7 +136,7 @@ class PacketHandler {
if (packet.isFlagSet(Packet.FLAG_ECHO)) {
if (packet.getSendStreamId() > 0) {
if (con.getOptions().getAnswerPings())
- receivePing(packet);
+ receivePing(con, packet);
else if (_log.shouldLog(Log.WARN))
_log.warn("Dropping Echo packet on existing con: " + packet);
} else if (packet.getReceiveStreamId() > 0) {
@@ -247,6 +247,8 @@ class PacketHandler {
reply.setSendStreamId(packet.getReceiveStreamId());
reply.setReceiveStreamId(packet.getSendStreamId());
reply.setOptionalFrom(_manager.getSession().getMyDestination());
+ reply.setLocalPort(packet.getLocalPort());
+ reply.setRemotePort(packet.getRemotePort());
// this just sends the packet - no retries or whatnot
_manager.getPacketQueue().enqueue(reply);
}
@@ -255,7 +257,7 @@ class PacketHandler {
if (packet.isFlagSet(Packet.FLAG_ECHO)) {
if (packet.getSendStreamId() > 0) {
if (_manager.answerPings())
- receivePing(packet);
+ receivePing(null, packet);
else if (_log.shouldLog(Log.WARN))
_log.warn("Dropping Echo packet on unknown con: " + packet);
} else if (packet.getReceiveStreamId() > 0) {
@@ -335,7 +337,10 @@ class PacketHandler {
}
}
- private void receivePing(Packet packet) {
+ /**
+ * @param con null if unknown
+ */
+ private void receivePing(Connection con, Packet packet) {
boolean ok = packet.verifySignature(_context, packet.getOptionalFrom(), null);
if (!ok) {
if (_log.shouldLog(Log.WARN)) {
@@ -348,10 +353,7 @@ class PacketHandler {
+ " sig=" + packet.getOptionalSignature().toBase64() + ")");
}
} else {
- PacketLocal pong = new PacketLocal(_context, packet.getOptionalFrom());
- pong.setFlag(Packet.FLAG_ECHO | Packet.FLAG_NO_ACK);
- pong.setReceiveStreamId(packet.getSendStreamId());
- _manager.getPacketQueue().enqueue(pong);
+ _manager.receivePing(con, packet);
}
}
--
GitLab