* Streaming, I2PSession:
Prep for SessionKeyManager work in the router - Comment out, deprecate, and javadoc for unused keys and tags, they are vestiges of end-to-end crypto
This commit is contained in:
@ -16,9 +16,9 @@ net.i2p.client.streaming.I2PServerSocket#accept} method, which will provide an
|
||||
application wants to create a new stream to a peer, it should do so with the
|
||||
appropriate {@link net.i2p.client.streaming.I2PSocketManager#connect} call.</p>
|
||||
|
||||
<p>There is a simple pair of demo applications available as well - {@link
|
||||
net.i2p.client.streaming.StreamSinkServer} listens to a destination and dumps
|
||||
the data from all sockets it accepts to individual files, while {@link
|
||||
net.i2p.client.streaming.StreamSinkClient} connects to a particular destination
|
||||
<p>There is a simple pair of demo applications available as well -
|
||||
net.i2p.client.streaming.StreamSinkServer listens to a destination and dumps
|
||||
the data from all sockets it accepts to individual files, while
|
||||
net.i2p.client.streaming.StreamSinkClient connects to a particular destination
|
||||
and sends a specific amount of random data then disconnects.</p>
|
||||
</body></html>
|
||||
|
@ -354,6 +354,7 @@ public class Connection {
|
||||
*/
|
||||
}
|
||||
|
||||
/*********
|
||||
private class PingNotifier implements ConnectionManager.PingNotifier {
|
||||
private long _startedPingOn;
|
||||
public PingNotifier() {
|
||||
@ -367,6 +368,7 @@ public class Connection {
|
||||
_options.updateRTT((int)time*2);
|
||||
}
|
||||
}
|
||||
*********/
|
||||
|
||||
List ackPackets(long ackThrough, long nacks[]) {
|
||||
if (ackThrough < _highestAckedThrough) {
|
||||
@ -548,20 +550,21 @@ public class Connection {
|
||||
killOutstandingPackets();
|
||||
}
|
||||
|
||||
/** ignore tag issues */
|
||||
private void killOutstandingPackets() {
|
||||
boolean tagsCancelled = false;
|
||||
//boolean tagsCancelled = false;
|
||||
synchronized (_outboundPackets) {
|
||||
for (Iterator iter = _outboundPackets.values().iterator(); iter.hasNext(); ) {
|
||||
PacketLocal pl = (PacketLocal)iter.next();
|
||||
if ( (pl.getTagsSent() != null) && (pl.getTagsSent().size() > 0) )
|
||||
tagsCancelled = true;
|
||||
//if ( (pl.getTagsSent() != null) && (pl.getTagsSent().size() > 0) )
|
||||
// tagsCancelled = true;
|
||||
pl.cancelled();
|
||||
}
|
||||
_outboundPackets.clear();
|
||||
_outboundPackets.notifyAll();
|
||||
}
|
||||
if (tagsCancelled)
|
||||
_context.sessionKeyManager().failTags(_remotePeer.getPublicKey());
|
||||
//if (tagsCancelled)
|
||||
// _context.sessionKeyManager().failTags(_remotePeer.getPublicKey());
|
||||
}
|
||||
|
||||
private class DisconnectEvent implements SimpleTimer.TimedEvent {
|
||||
@ -1140,12 +1143,12 @@ public class Connection {
|
||||
|
||||
// in case things really suck, the other side may have lost thier
|
||||
// session tags (e.g. they restarted), so jump back to ElGamal.
|
||||
int failTagsAt = _options.getMaxResends() - 2;
|
||||
if ( (newWindowSize == 1) && (numSends == failTagsAt) ) {
|
||||
if (_log.shouldLog(Log.WARN))
|
||||
_log.warn("Optimistically failing tags at resend " + numSends);
|
||||
_context.sessionKeyManager().failTags(_remotePeer.getPublicKey());
|
||||
}
|
||||
//int failTagsAt = _options.getMaxResends() - 2;
|
||||
//if ( (newWindowSize == 1) && (numSends == failTagsAt) ) {
|
||||
// if (_log.shouldLog(Log.WARN))
|
||||
// _log.warn("Optimistically failing tags at resend " + numSends);
|
||||
// _context.sessionKeyManager().failTags(_remotePeer.getPublicKey());
|
||||
//}
|
||||
|
||||
if (numSends - 1 > _options.getMaxResends()) {
|
||||
if (_log.shouldLog(Log.DEBUG))
|
||||
|
@ -349,24 +349,35 @@ public class ConnectionManager {
|
||||
return new HashSet(_connectionByInboundId.values());
|
||||
}
|
||||
}
|
||||
|
||||
/** blocking */
|
||||
public boolean ping(Destination peer, long timeoutMs) {
|
||||
return ping(peer, timeoutMs, true);
|
||||
return ping(peer, timeoutMs, true, null);
|
||||
}
|
||||
public boolean ping(Destination peer, long timeoutMs, boolean blocking) {
|
||||
return ping(peer, timeoutMs, blocking, null, null, null);
|
||||
return ping(peer, timeoutMs, blocking, null);
|
||||
}
|
||||
|
||||
/**
|
||||
* @deprecated I2PSession ignores tags, use non-tag variant
|
||||
* @param keyToUse ignored
|
||||
* @param tagsToSend ignored
|
||||
*/
|
||||
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, long timeoutMs, boolean blocking, PingNotifier notifier) {
|
||||
Long id = new Long(_context.random().nextLong(Packet.MAX_STREAM_ID-1)+1);
|
||||
PacketLocal packet = new PacketLocal(_context, peer);
|
||||
packet.setSendStreamId(id.longValue());
|
||||
packet.setFlag(Packet.FLAG_ECHO);
|
||||
packet.setFlag(Packet.FLAG_SIGNATURE_INCLUDED);
|
||||
packet.setOptionalFrom(_session.getMyDestination());
|
||||
if ( (keyToUse != null) && (tagsToSend != null) ) {
|
||||
packet.setKeyUsed(keyToUse);
|
||||
packet.setTagsSent(tagsToSend);
|
||||
}
|
||||
//if ( (keyToUse != null) && (tagsToSend != null) ) {
|
||||
// packet.setKeyUsed(keyToUse);
|
||||
// packet.setTagsSent(tagsToSend);
|
||||
//}
|
||||
|
||||
PingRequest req = new PingRequest(peer, packet, notifier);
|
||||
|
||||
@ -435,7 +446,7 @@ public class ConnectionManager {
|
||||
}
|
||||
public void pong() {
|
||||
_log.debug("Ping successful");
|
||||
_context.sessionKeyManager().tagsDelivered(_peer.getPublicKey(), _packet.getKeyUsed(), _packet.getTagsSent());
|
||||
//_context.sessionKeyManager().tagsDelivered(_peer.getPublicKey(), _packet.getKeyUsed(), _packet.getTagsSent());
|
||||
synchronized (ConnectionManager.PingRequest.this) {
|
||||
_ponged = true;
|
||||
ConnectionManager.PingRequest.this.notifyAll();
|
||||
|
@ -263,12 +263,12 @@ public class ConnectionPacketHandler {
|
||||
numResends++;
|
||||
|
||||
// ACK the tags we delivered so we can use them
|
||||
if ( (p.getKeyUsed() != null) && (p.getTagsSent() != null)
|
||||
&& (p.getTagsSent().size() > 0) ) {
|
||||
_context.sessionKeyManager().tagsDelivered(p.getTo().getPublicKey(),
|
||||
p.getKeyUsed(),
|
||||
p.getTagsSent());
|
||||
}
|
||||
//if ( (p.getKeyUsed() != null) && (p.getTagsSent() != null)
|
||||
// && (p.getTagsSent().size() > 0) ) {
|
||||
// _context.sessionKeyManager().tagsDelivered(p.getTo().getPublicKey(),
|
||||
// p.getKeyUsed(),
|
||||
// p.getTagsSent());
|
||||
//}
|
||||
if (_log.shouldLog(Log.DEBUG))
|
||||
_log.debug("Packet acked after " + p.getAckTime() + "ms: " + p);
|
||||
}
|
||||
|
@ -47,11 +47,31 @@ public class PacketLocal extends Packet implements MessageOutputStream.WriteStat
|
||||
public Destination getTo() { return _to; }
|
||||
public void setTo(Destination to) { _to = to; }
|
||||
|
||||
/**
|
||||
* @deprecated should always return null
|
||||
*/
|
||||
public SessionKey getKeyUsed() { return _keyUsed; }
|
||||
public void setKeyUsed(SessionKey key) { _keyUsed = key; }
|
||||
|
||||
/**
|
||||
* @deprecated I2PSession throws out the tags
|
||||
*/
|
||||
public void setKeyUsed(SessionKey key) {
|
||||
if (key != null)
|
||||
_log.error("Who is sending tags thru the streaming lib?");
|
||||
_keyUsed = key;
|
||||
}
|
||||
|
||||
/**
|
||||
* @deprecated should always return null or an empty set
|
||||
*/
|
||||
public Set getTagsSent() { return _tagsSent; }
|
||||
|
||||
/**
|
||||
* @deprecated I2PSession throws out the tags
|
||||
*/
|
||||
public void setTagsSent(Set tags) {
|
||||
if (tags != null && tags.size() > 0)
|
||||
_log.error("Who is sending tags thru the streaming lib? " + tags.size());
|
||||
if ( (_tagsSent != null) && (_tagsSent.size() > 0) && (tags.size() > 0) ) {
|
||||
//int old = _tagsSent.size();
|
||||
//_tagsSent.addAll(tags);
|
||||
|
@ -36,16 +36,18 @@ public class PacketQueue {
|
||||
|
||||
/**
|
||||
* Add a new packet to be sent out ASAP
|
||||
*
|
||||
* keys and tags disabled since dropped in I2PSession
|
||||
*/
|
||||
public void enqueue(PacketLocal packet) {
|
||||
packet.prepare();
|
||||
|
||||
SessionKey keyUsed = packet.getKeyUsed();
|
||||
if (keyUsed == null)
|
||||
keyUsed = new SessionKey();
|
||||
Set tagsSent = packet.getTagsSent();
|
||||
if (tagsSent == null)
|
||||
tagsSent = new HashSet(0);
|
||||
//SessionKey keyUsed = packet.getKeyUsed();
|
||||
//if (keyUsed == null)
|
||||
// keyUsed = new SessionKey();
|
||||
//Set tagsSent = packet.getTagsSent();
|
||||
//if (tagsSent == null)
|
||||
// tagsSent = new HashSet(0);
|
||||
|
||||
// cache this from before sendMessage
|
||||
String conStr = null;
|
||||
@ -92,13 +94,19 @@ public class PacketQueue {
|
||||
// I2PSessionImpl2
|
||||
//sent = _session.sendMessage(packet.getTo(), buf, 0, size, keyUsed, tagsSent, expires);
|
||||
// I2PSessionMuxedImpl
|
||||
sent = _session.sendMessage(packet.getTo(), buf, 0, size, keyUsed, tagsSent, expires,
|
||||
//sent = _session.sendMessage(packet.getTo(), buf, 0, size, keyUsed, tagsSent, expires,
|
||||
// I2PSession.PROTO_STREAMING, I2PSession.PORT_UNSPECIFIED, I2PSession.PORT_UNSPECIFIED);
|
||||
// I2PSessionMuxedImpl no tags
|
||||
sent = _session.sendMessage(packet.getTo(), buf, 0, size, null, null, expires,
|
||||
I2PSession.PROTO_STREAMING, I2PSession.PORT_UNSPECIFIED, I2PSession.PORT_UNSPECIFIED);
|
||||
else
|
||||
// I2PSessionImpl2
|
||||
//sent = _session.sendMessage(packet.getTo(), buf, 0, size, keyUsed, tagsSent, 0);
|
||||
// I2PSessionMuxedImpl
|
||||
sent = _session.sendMessage(packet.getTo(), buf, 0, size, keyUsed, tagsSent,
|
||||
//sent = _session.sendMessage(packet.getTo(), buf, 0, size, keyUsed, tagsSent,
|
||||
// I2PSession.PROTO_STREAMING, I2PSession.PORT_UNSPECIFIED, I2PSession.PORT_UNSPECIFIED);
|
||||
// I2PSessionMuxedImpl no tags
|
||||
sent = _session.sendMessage(packet.getTo(), buf, 0, size, null, null,
|
||||
I2PSession.PROTO_STREAMING, I2PSession.PORT_UNSPECIFIED, I2PSession.PORT_UNSPECIFIED);
|
||||
end = _context.clock().now();
|
||||
|
||||
@ -129,13 +137,11 @@ public class PacketQueue {
|
||||
if (c != null) // handle race on b0rk
|
||||
c.disconnect(false);
|
||||
} else {
|
||||
packet.setKeyUsed(keyUsed);
|
||||
packet.setTagsSent(tagsSent);
|
||||
//packet.setKeyUsed(keyUsed);
|
||||
//packet.setTagsSent(tagsSent);
|
||||
packet.incrementSends();
|
||||
if (_log.shouldLog(Log.DEBUG)) {
|
||||
String msg = "SEND " + packet + (tagsSent.size() > 0
|
||||
? " with " + tagsSent.size() + " tags"
|
||||
: "")
|
||||
String msg = "SEND " + packet
|
||||
+ " send # " + packet.getNumSends()
|
||||
+ " sendTime: " + (end-begin)
|
||||
+ " con: " + conStr;
|
||||
|
@ -93,6 +93,10 @@ class I2CPMessageProducer {
|
||||
/**
|
||||
* Package up and send the payload to the router for delivery
|
||||
*
|
||||
* @param tag unused - no end-to-end crypto
|
||||
* @param tags unused - no end-to-end crypto
|
||||
* @param key unused - no end-to-end crypto
|
||||
* @param newKey unused - no end-to-end crypto
|
||||
*/
|
||||
public void sendMessage(I2PSessionImpl session, Destination dest, long nonce, byte[] payload, SessionTag tag,
|
||||
SessionKey key, Set tags, SessionKey newKey, long expires) throws I2PSessionException {
|
||||
@ -135,6 +139,10 @@ class I2CPMessageProducer {
|
||||
/**
|
||||
* Create a new signed payload and send it off to the destination
|
||||
*
|
||||
* @param tag unused - no end-to-end crypto
|
||||
* @param tags unused - no end-to-end crypto
|
||||
* @param key unused - no end-to-end crypto
|
||||
* @param newKey unused - no end-to-end crypto
|
||||
*/
|
||||
private Payload createPayload(Destination dest, byte[] payload, SessionTag tag, SessionKey key, Set tags,
|
||||
SessionKey newKey) throws I2PSessionException {
|
||||
|
@ -355,17 +355,23 @@ abstract class I2PSessionImpl implements I2PSession, I2CPMessageReader.I2CPMessa
|
||||
*/
|
||||
public abstract boolean sendMessage(Destination dest, byte[] payload) throws I2PSessionException;
|
||||
|
||||
/**
|
||||
* @param keyUsed unused - no end-to-end crypto
|
||||
* @param tagsSent unused - no end-to-end crypto
|
||||
*/
|
||||
public abstract boolean sendMessage(Destination dest, byte[] payload, SessionKey keyUsed,
|
||||
Set tagsSent) throws I2PSessionException;
|
||||
|
||||
public abstract void receiveStatus(int msgId, long nonce, int status);
|
||||
|
||||
/****** no end-to-end crypto
|
||||
protected static final Set createNewTags(int num) {
|
||||
Set tags = new HashSet();
|
||||
for (int i = 0; i < num; i++)
|
||||
tags.add(new SessionTag(true));
|
||||
return tags;
|
||||
}
|
||||
*******/
|
||||
|
||||
/**
|
||||
* Recieve a payload message and let the app know its available
|
||||
|
@ -133,14 +133,28 @@ class I2PSessionImpl2 extends I2PSessionImpl {
|
||||
return sendMessage(dest, payload, offset, size, null, null, 0);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param keyUsed unused - no end-to-end crypto
|
||||
* @param tagsSent unused - no end-to-end crypto
|
||||
*/
|
||||
@Override
|
||||
public boolean sendMessage(Destination dest, byte[] payload, SessionKey keyUsed, Set tagsSent) throws I2PSessionException {
|
||||
return sendMessage(dest, payload, 0, payload.length, keyUsed, tagsSent, 0);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param keyUsed unused - no end-to-end crypto
|
||||
* @param tagsSent unused - no end-to-end crypto
|
||||
*/
|
||||
public boolean sendMessage(Destination dest, byte[] payload, int offset, int size, SessionKey keyUsed, Set tagsSent)
|
||||
throws I2PSessionException {
|
||||
return sendMessage(dest, payload, offset, size, keyUsed, tagsSent, 0);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param keyUsed unused - no end-to-end crypto
|
||||
* @param tagsSent unused - no end-to-end crypto
|
||||
*/
|
||||
public boolean sendMessage(Destination dest, byte[] payload, int offset, int size, SessionKey keyUsed, Set tagsSent, long expires)
|
||||
throws I2PSessionException {
|
||||
if (_log.shouldLog(Log.DEBUG)) _log.debug("sending message");
|
||||
@ -196,13 +210,17 @@ class I2PSessionImpl2 extends I2PSessionImpl {
|
||||
|
||||
private static final int NUM_TAGS = 50;
|
||||
|
||||
/**
|
||||
* @param keyUsed unused - no end-to-end crypto
|
||||
* @param tagsSent unused - no end-to-end crypto
|
||||
*/
|
||||
protected boolean sendBestEffort(Destination dest, byte payload[], SessionKey keyUsed, Set tagsSent, long expires)
|
||||
throws I2PSessionException {
|
||||
SessionKey key = null;
|
||||
SessionKey newKey = null;
|
||||
SessionTag tag = null;
|
||||
Set sentTags = null;
|
||||
int oldTags = 0;
|
||||
//SessionKey key = null;
|
||||
//SessionKey newKey = null;
|
||||
//SessionTag tag = null;
|
||||
//Set sentTags = null;
|
||||
//int oldTags = 0;
|
||||
long begin = _context.clock().now();
|
||||
/***********
|
||||
if (I2CPMessageProducer.END_TO_END_CRYPTO) {
|
||||
@ -256,27 +274,27 @@ class I2PSessionImpl2 extends I2PSessionImpl {
|
||||
long nonce = _context.random().nextInt(Integer.MAX_VALUE);
|
||||
if (_log.shouldLog(Log.DEBUG)) _log.debug("before sync state");
|
||||
MessageState state = new MessageState(_context, nonce, getPrefix());
|
||||
state.setKey(key);
|
||||
state.setTags(sentTags);
|
||||
state.setNewKey(newKey);
|
||||
//state.setKey(key);
|
||||
//state.setTags(sentTags);
|
||||
//state.setNewKey(newKey);
|
||||
state.setTo(dest);
|
||||
if (_log.shouldLog(Log.DEBUG)) _log.debug(getPrefix() + "Setting key = " + key);
|
||||
//if (_log.shouldLog(Log.DEBUG)) _log.debug(getPrefix() + "Setting key = " + key);
|
||||
|
||||
if (keyUsed != null) {
|
||||
//if (keyUsed != null) {
|
||||
//if (I2CPMessageProducer.END_TO_END_CRYPTO) {
|
||||
// if (newKey != null)
|
||||
// keyUsed.setData(newKey.getData());
|
||||
// else
|
||||
// keyUsed.setData(key.getData());
|
||||
//} else {
|
||||
keyUsed.setData(SessionKey.INVALID_KEY.getData());
|
||||
// keyUsed.setData(SessionKey.INVALID_KEY.getData());
|
||||
//}
|
||||
}
|
||||
if (tagsSent != null) {
|
||||
if (sentTags != null) {
|
||||
tagsSent.addAll(sentTags);
|
||||
}
|
||||
}
|
||||
//}
|
||||
//if (tagsSent != null) {
|
||||
// if (sentTags != null) {
|
||||
// tagsSent.addAll(sentTags);
|
||||
// }
|
||||
//}
|
||||
|
||||
if (_log.shouldLog(Log.DEBUG)) _log.debug("before sync state");
|
||||
long beforeSendingSync = _context.clock().now();
|
||||
@ -291,7 +309,8 @@ class I2PSessionImpl2 extends I2PSessionImpl {
|
||||
+ state.getNonce() + " for best effort "
|
||||
+ " sync took " + (inSendingSync-beforeSendingSync)
|
||||
+ " add took " + (afterSendingSync-inSendingSync));
|
||||
_producer.sendMessage(this, dest, nonce, payload, tag, key, sentTags, newKey, expires);
|
||||
//_producer.sendMessage(this, dest, nonce, payload, tag, key, sentTags, newKey, expires);
|
||||
_producer.sendMessage(this, dest, nonce, payload, null, null, null, null, expires);
|
||||
|
||||
// since this is 'best effort', all we're waiting for is a status update
|
||||
// saying that the router received it - in theory, that should come back
|
||||
|
@ -128,6 +128,10 @@ class I2PSessionMuxedImpl extends I2PSessionImpl2 implements I2PSession {
|
||||
return sendMessage(dest, payload, 0, payload.length, null, null, 0, proto, fromport, toport);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param keyUsed unused - no end-to-end crypto
|
||||
* @param tagsSent unused - no end-to-end crypto
|
||||
*/
|
||||
@Override
|
||||
public boolean sendMessage(Destination dest, byte[] payload, int offset, int size,
|
||||
SessionKey keyUsed, Set tagsSent, long expires)
|
||||
@ -135,6 +139,10 @@ class I2PSessionMuxedImpl extends I2PSessionImpl2 implements I2PSession {
|
||||
return sendMessage(dest, payload, offset, size, keyUsed, tagsSent, 0, PROTO_UNSPECIFIED, PORT_UNSPECIFIED, PORT_UNSPECIFIED);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param keyUsed unused - no end-to-end crypto
|
||||
* @param tagsSent unused - no end-to-end crypto
|
||||
*/
|
||||
@Override
|
||||
public boolean sendMessage(Destination dest, byte[] payload, int offset, int size, SessionKey keyUsed, Set tagsSent,
|
||||
int proto, int fromport, int toport) throws I2PSessionException {
|
||||
@ -142,6 +150,8 @@ class I2PSessionMuxedImpl extends I2PSessionImpl2 implements I2PSession {
|
||||
}
|
||||
|
||||
/**
|
||||
* @param keyUsed unused - no end-to-end crypto
|
||||
* @param tagsSent unused - no end-to-end crypto
|
||||
* @param proto 1-254 or 0 for unset; recommended:
|
||||
* I2PSession.PROTO_UNSPECIFIED
|
||||
* I2PSession.PROTO_STREAMING
|
||||
|
@ -12,7 +12,7 @@ import org.bouncycastle.crypto.macs.I2PHMac;
|
||||
|
||||
/**
|
||||
* Calculate the HMAC-SHA256 of a key+message. All the good stuff occurs
|
||||
* in {@link org.bouncycastle.crypto.macs.HMac} and
|
||||
* in {@link org.bouncycastle.crypto.macs.I2PHMac} and
|
||||
* {@link org.bouncycastle.crypto.digests.MD5Digest}.
|
||||
*
|
||||
*/
|
||||
|
@ -15,7 +15,7 @@ import org.bouncycastle.crypto.macs.I2PHMac;
|
||||
|
||||
/**
|
||||
* Calculate the HMAC-MD5 of a key+message. All the good stuff occurs
|
||||
* in {@link org.bouncycastle.crypto.macs.HMac} and
|
||||
* in {@link org.bouncycastle.crypto.macs.I2PHMac} and
|
||||
* {@link org.bouncycastle.crypto.digests.MD5Digest}.
|
||||
*
|
||||
*/
|
||||
|
Reference in New Issue
Block a user