i2p/i2p.i2p
1
0
Fork 0
mirror of https://github.com/i2p/i2p.i2p.git synced 2024-12-06 19:27:00 +01:00
Code Issues Actions 317 Packages Projects Releases Wiki Activity

SSU2: Token improvements and fixes part 1

Browse Source 
- Set cache size based on connection limit
- Track average inbound cache eviction time
- Set inbound expiration based on cache time
- Reduce max inbound expiration
- Fix saving inbound token sent after relay response or hole punch
- Dont send or save tokens if we are symmetric natted
- Sort persisted tokens by expiration so they are expired in correct order on reload
- Periodically expire tokens from cache
- Add getters to Token class
- Add missing case IPV4_SNAT_IPV6_UNKNOWN to EnumSets
This commit is contained in:
zzz
2022-12-04 10:04:18 -05:00
parent 974ffabd18
commit 706cd5a129
10 changed files with 222 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
router/java/src/net/i2p/router/RouterVersion.java
+1 -1
View File
@@ -18,7 +18,7 @@ public class RouterVersion {
/** deprecated */
public final static String ID = "Git";
public final static String VERSION = CoreVersion.VERSION;
public final static long BUILD = 3;
public final static long BUILD = 4;
/** for example "-test" */
public final static String EXTRA = "";
router/java/src/net/i2p/router/transport/udp/EstablishmentManager.java
+171 -44
View File
@@ -14,6 +14,8 @@ import java.net.UnknownHostException;
import java.security.GeneralSecurityException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -45,6 +47,8 @@ import static net.i2p.router.transport.udp.OutboundEstablishState2.IntroState.*;
import static net.i2p.router.transport.udp.SSU2Util.*;
import net.i2p.router.util.DecayingHashSet;
import net.i2p.router.util.DecayingBloomFilter;
import net.i2p.stat.Rate;
import net.i2p.stat.RateStat;
import net.i2p.util.Addresses;
import net.i2p.util.HexDump;
import net.i2p.util.I2PThread;
@@ -176,8 +180,9 @@ class EstablishmentManager {
private static final String PROP_DISABLE_EXT_OPTS = "i2np.udp.disableExtendedOptions";
// SSU 2
private static final int MAX_TOKENS = 512;
public static final long IB_TOKEN_EXPIRATION = 2*60*60*1000L;
private static final int MIN_TOKENS = 128;
private static final int MAX_TOKENS = 2048;
public static final long IB_TOKEN_EXPIRATION = 60*60*1000L;
private static final long MAX_SKEW = 2*60*1000;
private static final String TOKEN_FILE = "ssu2tokens.txt";
@@ -198,8 +203,10 @@ class EstablishmentManager {
_outboundByHash = new ConcurrentHashMap<Hash, OutboundEstablishState>();
_inboundBans = new LHMCache<RemoteHostId, Long>(32);
if (_enableSSU2) {
_inboundTokens = new LHMCache<RemoteHostId, Token>(MAX_TOKENS);
_outboundTokens = new LHMCache<RemoteHostId, Token>(MAX_TOKENS);
// roughly scale based on expected traffic
int tokenCacheSize = Math.max(MIN_TOKENS, Math.min(MAX_TOKENS, 3 * _transport.getMaxConnections() / 4));
_inboundTokens = new InboundTokens(tokenCacheSize);
_outboundTokens = new LHMCache<RemoteHostId, Token>(tokenCacheSize);
} else {
_inboundTokens = null;
_outboundTokens = null;
@@ -233,6 +240,8 @@ class EstablishmentManager {
//_context.statManager().createRateStat("udp.queueDropSize", "How many messages were queued up when it was considered full, causing a tail drop?", "udp", UDPTransport.RATES);
//_context.statManager().createRateStat("udp.queueAllowTotalLifetime", "When a peer is retransmitting and we probabalistically allow a new message, what is the sum of the pending message lifetimes? (period is the new message's lifetime)?", "udp", UDPTransport.RATES);
_context.statManager().createRateStat("udp.dupDHX", "Session request replay", "udp", new long[] { 24*60*60*1000L } );
if (_enableSSU2)
_context.statManager().createRequiredRateStat("udp.inboundTokenLifetime", "SSU2 token lifetime (ms)", "udp", new long[] { 5*60*1000L } );
}
public synchronized void startup() {
@@ -2538,14 +2547,20 @@ class EstablishmentManager {
* Remember a token that can be used later to connect to the peer
*
* @param token nonzero
* @param expires absolute time
* @since 0.9.54
*/
public void addOutboundToken(RemoteHostId peer, long token, long expires) {
// so we don't use a token about to expire
expires -= 2*60*1000;
if (expires < _context.clock().now())
long now = _context.clock().now();
if (expires < now)
return;
Token tok = new Token(token, expires);
if (expires > now + 2*60*1000) {
// don't save if symmetric natted
byte[] ip = peer.getIP();
if (ip != null && ip.length == 4 && _transport.isSnatted())
return;
}
Token tok = new Token(token, expires, now);
synchronized(_outboundTokens) {
_outboundTokens.put(peer, tok);
}
@@ -2564,9 +2579,9 @@ class EstablishmentManager {
}
if (tok == null)
return 0;
if (tok.expires < _context.clock().now())
if (tok.getExpiration() < _context.clock().now())
return 0;
return tok.token;
return tok.getToken();
}
/**
@@ -2624,9 +2639,10 @@ class EstablishmentManager {
}
/**
* Get a token that can be used later for the peer to connect to us
* Get a token that can be used later for the peer to connect to us.
*
* @param expiration time from now
* @param expiration time from now, will be reduced if necessary based on cache eviction time.
* @return non-null
* @since 0.9.55
*/
public Token getInboundToken(RemoteHostId peer, long expiration) {
@@ -2635,15 +2651,27 @@ class EstablishmentManager {
token = _context.random().nextLong();
} while (token == 0);
long now = _context.clock().now();
Token tok;
// shorten expiration based on average eviction time
RateStat rs = _context.statManager().getRate("udp.inboundTokenLifetime");
if (rs != null) {
Rate r = rs.getRate(5*60*1000);
if (r != null) {
long lifetime = (long) (r.getAverageValue() * 0.9d); // margin
if (lifetime > 0) {
if (lifetime < 2*60*1000)
lifetime = 2*60*1000;
if (lifetime < expiration)
expiration = lifetime;
}
}
}
long expires = now + expiration;
Token tok = new Token(token, expires, now);
synchronized(_inboundTokens) {
// shorten expiration based on _inboundTokens size
if (expiration > 2*60*1000 && _inboundTokens.size() > MAX_TOKENS / 2)
expiration /= 2;
long expires = now + expiration;
tok = new Token(token, expires);
_inboundTokens.put(peer, tok);
}
if (_log.shouldDebug())
_log.debug("Add inbound " + tok + " for " + peer);
return tok;
}
@@ -2661,21 +2689,45 @@ class EstablishmentManager {
tok = _inboundTokens.get(peer);
if (tok == null)
return false;
if (tok.token != token)
if (tok.getToken() != token)
return false;
_inboundTokens.remove(peer);
}
return tok.expires >= _context.clock().now();
boolean rv = tok.getExpiration() >= _context.clock().now();
if (rv && _log.shouldDebug())
_log.debug("Used inbound " + tok + " for " + peer);
return rv;
}
public static class Token {
public final long token, expires;
public Token(long tok, long exp) {
token = tok; expires = exp;
private final long token;
// save space until 2106
private final int expires;
private final int added;
/**
* @param exp absolute time, not relative to now
*/
public Token(long tok, long exp, long now) {
token = tok;
expires = (int) (exp >> 10);
added = (int) (now >> 10);
}
/** @since 0.9.57 */
public long getToken() { return token; }
/** @since 0.9.57 */
public long getExpiration() { return (expires & 0xFFFFFFFFL) << 10; }
/** @since 0.9.57 */
public long getWhenAdded() { return (added & 0xFFFFFFFFL) << 10; }
/** @since 0.9.57 */
public String toString() {
return "Token " + token + " added " + DataHelper.formatTime(getWhenAdded()) + " expires " + DataHelper.formatTime(getExpiration());
}
}
/**
* Not threaded, because we're holding the token cache locks anyway.
*
* Format:
*
*<pre>
@@ -2741,7 +2793,7 @@ class EstablishmentManager {
int port = Integer.parseInt(s[2]);
long tok = Long.parseLong(s[3]);
RemoteHostId id = new RemoteHostId(ip, port);
Token token = new Token(tok, exp);
Token token = new Token(tok, exp, now);
if (s[0].equals("I"))
_inboundTokens.put(id, token);
else
@@ -2790,27 +2842,39 @@ class EstablishmentManager {
}
long now = _context.clock().now();
int count = 0;
// Roughly speaking, the LHMCache will iterate newest-first,
// so when we add them back in loadTokens(), the oldest would be at
// the head of the map and the newest would be purged first.
// Sort them by expiration oldest-first so loadTokens() will
// put them in the LHMCache in the right order.
TokenComparator comp = new TokenComparator();
List<Map.Entry<RemoteHostId, Token>> tmp;
synchronized(_inboundTokens) {
for (Map.Entry<RemoteHostId, Token> e : _inboundTokens.entrySet()) {
Token token = e.getValue();
long exp = token.expires;
if (exp <= now)
continue;
RemoteHostId id = e.getKey();
out.println("I " + Addresses.toString(id.getIP()) + ' ' + id.getPort() + ' ' + token.token + ' ' + exp);
count++;
}
tmp = new ArrayList<Map.Entry<RemoteHostId, Token>>(_inboundTokens.entrySet());
}
Collections.sort(tmp, comp);
for (Map.Entry<RemoteHostId, Token> e : tmp) {
Token token = e.getValue();
long exp = token.getExpiration();
if (exp <= now)
continue;
RemoteHostId id = e.getKey();
out.println("I " + Addresses.toString(id.getIP()) + ' ' + id.getPort() + ' ' + token.getToken() + ' ' + exp);
count++;
}
tmp.clear();
synchronized(_outboundTokens) {
for (Map.Entry<RemoteHostId, Token> e : _outboundTokens.entrySet()) {
Token token = e.getValue();
long exp = token.expires;
if (exp <= now)
continue;
RemoteHostId id = e.getKey();
out.println("O " + Addresses.toString(id.getIP()) + ' ' + id.getPort() + ' ' + token.token + ' ' + exp);
count++;
}
tmp.addAll(_outboundTokens.entrySet());
}
Collections.sort(tmp, comp);
for (Map.Entry<RemoteHostId, Token> e : tmp) {
Token token = e.getValue();
long exp = token.getExpiration();
if (exp <= now)
continue;
RemoteHostId id = e.getKey();
out.println("O " + Addresses.toString(id.getIP()) + ' ' + id.getPort() + ' ' + token.getToken() + ' ' + exp);
count++;
}
if (out.checkError())
throw new IOException("Failed write to " + f);
@@ -2825,6 +2889,45 @@ class EstablishmentManager {
}
/**
* Soonest expiration first
* @since 0.9.57
*/
private static class TokenComparator implements Comparator<Map.Entry<RemoteHostId, Token>> {
public int compare(Map.Entry<RemoteHostId, Token> l, Map.Entry<RemoteHostId, Token> r) {
long le = l.getValue().expires;
long re = r.getValue().expires;
if (le < re) return -1;
if (le > re) return 1;
return 0;
}
}
/**
* For inbound tokens only, to record eviction time in a stat,
* for use in setting expiration times.
*
* @since 0.9.57
*/
private class InboundTokens extends LHMCache<RemoteHostId, Token> {
public InboundTokens(int max) {
super(max);
}
@Override
protected boolean removeEldestEntry(Map.Entry<RemoteHostId, Token> eldest) {
boolean rv = super.removeEldestEntry(eldest);
if (rv) {
long lifetime = _context.clock().now() - eldest.getValue().getWhenAdded();
_context.statManager().addRateData("udp.inboundTokenLifetime", lifetime);
if (_log.shouldDebug())
_log.debug("Remove oldest inbound " + eldest.getValue() + " for " + eldest.getKey());
}
return rv;
}
}
/**
* Process SSU2 hole punch payload
*
@@ -2968,7 +3071,7 @@ class EstablishmentManager {
_activity = 0;
if (_lastFailsafe + FAILSAFE_INTERVAL < now) {
_lastFailsafe = now;
doFailsafe();
doFailsafe(now);
}
long nextSendTime = Math.min(handleInbound(), handleOutbound());
@@ -2991,7 +3094,7 @@ class EstablishmentManager {
}
/** @since 0.9.2 */
private void doFailsafe() {
private void doFailsafe(long now) {
for (Iterator<OutboundEstablishState> iter = _liveIntroductions.values().iterator(); iter.hasNext(); ) {
OutboundEstablishState state = iter.next();
if (state.getLifetime() > 3*MAX_OB_ESTABLISH_TIME) {
@@ -3016,6 +3119,30 @@ class EstablishmentManager {
_log.warn("Failsafe remove OBBH " + state);
}
}
int count = 0;
synchronized(_inboundTokens) {
for (Iterator<Token> iter = _inboundTokens.values().iterator(); iter.hasNext(); ) {
Token tok = iter.next();
if (tok.getExpiration() < now) {
iter.remove();
count++;
}
}
}
if (count > 0 && _log.shouldDebug())
_log.debug("Expired " + count + " inbound tokens");
count = 0;
synchronized(_outboundTokens) {
for (Iterator<Token> iter = _outboundTokens.values().iterator(); iter.hasNext(); ) {
Token tok = iter.next();
if (tok.getExpiration() < now) {
iter.remove();
count++;
}
}
}
if (count > 0 && _log.shouldDebug())
_log.debug("Expired " + count + " outbound tokens");
}
}
}
router/java/src/net/i2p/router/transport/udp/InboundEstablishState2.java
+5
View File
@@ -469,7 +469,12 @@ class InboundEstablishState2 extends InboundEstablishState implements SSU2Payloa
public long getSendConnID() { return _sendConnID; }
public long getRcvConnID() { return _rcvConnID; }
public long getToken() { return _token; }
/**
* @return may be null
*/
public EstablishmentManager.Token getNextToken() {
if (_aliceIP.length == 4 && _transport.isSnatted())
return null;
return _transport.getEstablisher().getInboundToken(_remoteHostId);
}
public HandshakeState getHandshakeState() { return _handshakeState; }
router/java/src/net/i2p/router/transport/udp/IntroductionManager.java
+1 -1
View File
@@ -1059,7 +1059,7 @@ class IntroductionManager {
if (rcode == SSU2Util.RELAY_ACCEPT) {
RemoteHostId aliceID = new RemoteHostId(testIP, testPort);
EstablishmentManager.Token tok = _transport.getEstablisher().getInboundToken(aliceID, 60*1000);
token = tok.token;
token = tok.getToken();
} else {
token = 0;
}
router/java/src/net/i2p/router/transport/udp/OutboundEstablishState2.java
+5
View File
@@ -407,7 +407,12 @@ class OutboundEstablishState2 extends OutboundEstablishState implements SSU2Payl
public long getSendConnID() { return _sendConnID; }
public long getRcvConnID() { return _rcvConnID; }
public long getToken() { return _token; }
/**
* @return may be null
*/
public EstablishmentManager.Token getNextToken() {
if (_bobIP != null && _bobIP.length == 4 && _transport.isSnatted())
return null;
return _transport.getEstablisher().getInboundToken(_remoteHostId);
}
public HandshakeState getHandshakeState() { return _handshakeState; }
router/java/src/net/i2p/router/transport/udp/PacketBuilder2.java
+7 -4
View File
@@ -361,10 +361,11 @@ class PacketBuilder2 {
_log.debug("Sending termination " + reason + " to : " + peer);
List<Block> blocks = new ArrayList<Block>(2);
if (peer.getKeyEstablishedTime() - _context.clock().now() > EstablishmentManager.IB_TOKEN_EXPIRATION / 2 &&
!_context.router().gracefulShutdownInProgress()) {
!_context.router().gracefulShutdownInProgress() &&
(peer.isIPv6() || !_transport.isSnatted())) {
// update token
EstablishmentManager.Token token = _transport.getEstablisher().getInboundToken(peer.getRemoteHostId());
Block block = new SSU2Payload.NewTokenBlock(token.token, token.expires);
Block block = new SSU2Payload.NewTokenBlock(token);
blocks.add(block);
}
Block block = new SSU2Payload.TerminationBlock(reason, peer.getReceivedMessages().getHighestSet());
@@ -904,6 +905,7 @@ class PacketBuilder2 {
/**
* @param packet containing only 32 byte header
* @param token may be null
*/
private void encryptSessionCreated(UDPPacket packet, HandshakeState state,
byte[] hdrKey1, byte[] hdrKey2, long relayTag,
@@ -925,7 +927,7 @@ class PacketBuilder2 {
blocks.add(block);
}
if (token != null) {
block = new SSU2Payload.NewTokenBlock(token.token, token.expires);
block = new SSU2Payload.NewTokenBlock(token);
len += block.getTotalLength();
blocks.add(block);
}
@@ -1067,6 +1069,7 @@ class PacketBuilder2 {
*
* @param packet containing only 16 byte header
* @param addPadding force-add exactly this size a padding block, for jumbo only
* @param token may be null
*/
private void encryptSessionConfirmed(UDPPacket packet, HandshakeState state, int mtu, int numFragments, int addPadding,
boolean isIPv6, byte[] hdrKey1, byte[] hdrKey2,
@@ -1083,7 +1086,7 @@ class PacketBuilder2 {
blocks.add(riblock);
// only if room
if (token != null && mtu - (SHORT_HEADER_SIZE + KEY_LEN + MAC_LEN + len + MAC_LEN) >= 15) {
Block block = new SSU2Payload.NewTokenBlock(token.token, token.expires);
Block block = new SSU2Payload.NewTokenBlock(token);
len += block.getTotalLength();
blocks.add(block);
}
router/java/src/net/i2p/router/transport/udp/PeerState2.java
+10 -6
View File
@@ -797,12 +797,16 @@ public class PeerState2 extends PeerState implements SSU2Payload.PayloadCallback
_log.warn("Migration successful, changed address from " + _remoteHostId + " to " + from + " for " + this);
_transport.changePeerAddress(this, from);
_mtu = MIN_MTU;
EstablishmentManager.Token token = _transport.getEstablisher().getInboundToken(from);
SSU2Payload.Block block = new SSU2Payload.NewTokenBlock(token.token, token.expires);
UDPPacket pkt = _transport.getBuilder2().buildPacket(Collections.<Fragment>emptyList(),
Collections.singletonList(block),
this);
_transport.send(pkt);
if (isIPv6() || !_transport.isSnatted()) {
EstablishmentManager.Token token = _transport.getEstablisher().getInboundToken(from);
SSU2Payload.Block block = new SSU2Payload.NewTokenBlock(token);
UDPPacket pkt = _transport.getBuilder2().buildPacket(Collections.<Fragment>emptyList(),
Collections.singletonList(block),
this);
_transport.send(pkt);
} else {
messagePartiallyReceived();
}
} else {
// caller will handle
// ACK-eliciting
router/java/src/net/i2p/router/transport/udp/SSU2Payload.java
+5 -6
View File
@@ -846,12 +846,11 @@ class SSU2Payload {
}
public static class NewTokenBlock extends Block {
private final long t, e;
private final EstablishmentManager.Token tok;
public NewTokenBlock(long token, long expires) {
public NewTokenBlock(EstablishmentManager.Token token) {
super(BLOCK_NEWTOKEN);
t = token;
e = expires / 1000;
tok = token;
}
public int getDataLength() {
@@ -859,9 +858,9 @@ class SSU2Payload {
}
public int writeData(byte[] tgt, int off) {
DataHelper.toLong(tgt, off, 4, e);
DataHelper.toLong(tgt, off, 4, tok.getExpiration() / 1000);
off += 4;
DataHelper.toLong8(tgt, off, t);
DataHelper.toLong8(tgt, off, tok.getToken());
return off + 8;
}
}
router/java/src/net/i2p/router/transport/udp/UDPTransport.java
+14
View File
@@ -287,6 +287,7 @@ public class UDPTransport extends TransportImpl implements TimedWeightedPriority
Status.REJECT_UNSOLICITED,
Status.IPV4_FIREWALLED_IPV6_OK,
Status.IPV4_SNAT_IPV6_OK,
Status.IPV4_SNAT_IPV6_UNKNOWN,
Status.IPV4_FIREWALLED_IPV6_UNKNOWN);
private static final Set<Status> STATUS_IPV6_FW = EnumSet.of(Status.IPV4_OK_IPV6_FIREWALLED,
@@ -297,6 +298,7 @@ public class UDPTransport extends TransportImpl implements TimedWeightedPriority
Status.REJECT_UNSOLICITED,
Status.IPV4_FIREWALLED_IPV6_OK,
Status.IPV4_SNAT_IPV6_OK,
Status.IPV4_SNAT_IPV6_UNKNOWN,
Status.IPV4_FIREWALLED_IPV6_UNKNOWN,
Status.IPV4_OK_IPV6_FIREWALLED,
Status.IPV4_UNKNOWN_IPV6_FIREWALLED,
@@ -329,6 +331,10 @@ public class UDPTransport extends TransportImpl implements TimedWeightedPriority
private static final Set<Status> STATUS_OK = EnumSet.of(Status.OK,
Status.IPV4_DISABLED_IPV6_OK);
private static final Set<Status> STATUS_IPV4_SNAT = EnumSet.of(Status.DIFFERENT,
Status.IPV4_SNAT_IPV6_OK,
Status.IPV4_SNAT_IPV6_UNKNOWN);
/**
* @param dh non-null to enable SSU1
@@ -3906,6 +3912,14 @@ public class UDPTransport extends TransportImpl implements TimedWeightedPriority
return _reachabilityStatus;
}
/**
* Is IPv4 Symmetric NATted?
* @since 0.9.57
*/
boolean isSnatted() {
return STATUS_IPV4_SNAT.contains(getReachabilityStatus());
}
/**
* @deprecated unused
*/