Add handshake keys to TransportKeyManagerImpl.

This commit is contained in:
akwizgran
2019-05-07 10:33:26 +01:00
parent 3f51ad6c07
commit 5adc9d8dbd
8 changed files with 180 additions and 111 deletions

View File

@@ -1,34 +1,52 @@
package org.briarproject.bramble.api.transport;
import org.briarproject.bramble.api.contact.ContactId;
import org.briarproject.bramble.api.contact.PendingContactId;
import org.briarproject.bramble.api.crypto.SecretKey;
import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
import org.briarproject.bramble.api.plugin.TransportId;
import javax.annotation.Nullable;
import javax.annotation.concurrent.Immutable;
@Immutable
@NotNullByDefault
public class StreamContext {
@Nullable
private final ContactId contactId;
@Nullable
private final PendingContactId pendingContactId;
private final TransportId transportId;
private final SecretKey tagKey, headerKey;
private final long streamNumber;
private final boolean handshakeMode;
public StreamContext(ContactId contactId, TransportId transportId,
SecretKey tagKey, SecretKey headerKey, long streamNumber) {
public StreamContext(@Nullable ContactId contactId,
@Nullable PendingContactId pendingContactId,
TransportId transportId, SecretKey tagKey, SecretKey headerKey,
long streamNumber, boolean handshakeMode) {
if ((contactId == null) == (pendingContactId == null))
throw new IllegalArgumentException();
this.contactId = contactId;
this.pendingContactId = pendingContactId;
this.transportId = transportId;
this.tagKey = tagKey;
this.headerKey = headerKey;
this.streamNumber = streamNumber;
this.handshakeMode = handshakeMode;
}
@Nullable
public ContactId getContactId() {
return contactId;
}
@Nullable
public PendingContactId getPendingContactId() {
return pendingContactId;
}
public TransportId getTransportId() {
return transportId;
}
@@ -44,4 +62,8 @@ public class StreamContext {
public long getStreamNumber() {
return streamNumber;
}
public boolean isHandshakeMode() {
return handshakeMode;
}
}

View File

@@ -24,6 +24,8 @@ public class TransportKeySet {
public TransportKeySet(KeySetId keySetId, @Nullable ContactId contactId,
@Nullable PendingContactId pendingContactId, TransportKeys keys) {
if ((contactId == null) == (pendingContactId == null))
throw new IllegalArgumentException();
this.keySetId = keySetId;
this.contactId = contactId;
this.pendingContactId = pendingContactId;

View File

@@ -96,6 +96,7 @@ class ConnectionManagerImpl implements ConnectionManager {
TransportConnectionReader r) throws IOException {
InputStream streamReader = streamReaderFactory.createStreamReader(
r.getInputStream(), ctx);
// TODO: Pending contacts, handshake mode
return syncSessionFactory.createIncomingSession(ctx.getContactId(),
streamReader);
}
@@ -104,6 +105,7 @@ class ConnectionManagerImpl implements ConnectionManager {
TransportConnectionWriter w) throws IOException {
StreamWriter streamWriter = streamWriterFactory.createStreamWriter(
w.getOutputStream(), ctx);
// TODO: Pending contacts, handshake mode
return syncSessionFactory.createSimplexOutgoingSession(
ctx.getContactId(), w.getMaxLatency(), streamWriter);
}
@@ -112,6 +114,7 @@ class ConnectionManagerImpl implements ConnectionManager {
TransportConnectionWriter w) throws IOException {
StreamWriter streamWriter = streamWriterFactory.createStreamWriter(
w.getOutputStream(), ctx);
// TODO: Pending contacts, handshake mode
return syncSessionFactory.createDuplexOutgoingSession(
ctx.getContactId(), w.getMaxLatency(), w.getMaxIdleTime(),
streamWriter);
@@ -145,6 +148,7 @@ class ConnectionManagerImpl implements ConnectionManager {
disposeReader(false, false);
return;
}
// TODO: Pending contacts
ContactId contactId = ctx.getContactId();
connectionRegistry.registerConnection(contactId, transportId, true);
try {
@@ -388,7 +392,7 @@ class ConnectionManagerImpl implements ConnectionManager {
return;
}
// Check that the stream comes from the expected contact
if (!ctx.getContactId().equals(contactId)) {
if (!contactId.equals(ctx.getContactId())) {
LOG.warning("Wrong contact ID for returning stream");
disposeReader(true, true);
return;

View File

@@ -1,30 +0,0 @@
package org.briarproject.bramble.transport;
import org.briarproject.bramble.api.contact.ContactId;
import org.briarproject.bramble.api.transport.KeySetId;
class MutableKeySet {
private final KeySetId keySetId;
private final ContactId contactId;
private final MutableTransportKeys transportKeys;
MutableKeySet(KeySetId keySetId, ContactId contactId,
MutableTransportKeys transportKeys) {
this.keySetId = keySetId;
this.contactId = contactId;
this.transportKeys = transportKeys;
}
KeySetId getKeySetId() {
return keySetId;
}
ContactId getContactId() {
return contactId;
}
MutableTransportKeys getTransportKeys() {
return transportKeys;
}
}

View File

@@ -0,0 +1,50 @@
package org.briarproject.bramble.transport;
import org.briarproject.bramble.api.contact.ContactId;
import org.briarproject.bramble.api.contact.PendingContactId;
import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
import org.briarproject.bramble.api.transport.KeySetId;
import javax.annotation.Nullable;
import javax.annotation.concurrent.NotThreadSafe;
@NotThreadSafe
@NotNullByDefault
class MutableTransportKeySet {
private final KeySetId keySetId;
@Nullable
private final ContactId contactId;
@Nullable
private final PendingContactId pendingContactId;
private final MutableTransportKeys keys;
MutableTransportKeySet(KeySetId keySetId, @Nullable ContactId contactId,
@Nullable PendingContactId pendingContactId,
MutableTransportKeys keys) {
if ((contactId == null) == (pendingContactId == null))
throw new IllegalArgumentException();
this.keySetId = keySetId;
this.contactId = contactId;
this.pendingContactId = pendingContactId;
this.keys = keys;
}
KeySetId getKeySetId() {
return keySetId;
}
@Nullable
ContactId getContactId() {
return contactId;
}
@Nullable
PendingContactId getPendingContactId() {
return pendingContactId;
}
MutableTransportKeys getKeys() {
return keys;
}
}

View File

@@ -2,6 +2,7 @@ package org.briarproject.bramble.transport;
import org.briarproject.bramble.api.Bytes;
import org.briarproject.bramble.api.contact.ContactId;
import org.briarproject.bramble.api.contact.PendingContactId;
import org.briarproject.bramble.api.crypto.SecretKey;
import org.briarproject.bramble.api.crypto.TransportCrypto;
import org.briarproject.bramble.api.db.DatabaseComponent;
@@ -28,6 +29,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;
import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.ThreadSafe;
@@ -58,11 +60,12 @@ class TransportKeyManagerImpl implements TransportKeyManager {
private final ReentrantLock lock = new ReentrantLock();
@GuardedBy("lock")
private final Map<KeySetId, MutableKeySet> keys = new HashMap<>();
private final Map<KeySetId, MutableTransportKeySet> keys = new HashMap<>();
@GuardedBy("lock")
private final Map<Bytes, TagContext> inContexts = new HashMap<>();
@GuardedBy("lock")
private final Map<ContactId, MutableKeySet> outContexts = new HashMap<>();
private final Map<ContactId, MutableTransportKeySet> outContexts =
new HashMap<>();
TransportKeyManagerImpl(DatabaseComponent db,
TransportCrypto transportCrypto, Executor dbExecutor,
@@ -86,23 +89,23 @@ class TransportKeyManagerImpl implements TransportKeyManager {
// Load the transport keys from the DB
Collection<TransportKeySet> loaded =
db.getTransportKeys(txn, transportId);
// Rotate the keys to the current time period
RotationResult rotationResult = rotateKeys(loaded, now);
// Update the keys to the current time period
UpdateResult updateResult = updateKeys(loaded, now);
// Initialise mutable state for all contacts
addKeys(rotationResult.current);
// Write any rotated keys back to the DB
if (!rotationResult.rotated.isEmpty())
db.updateTransportKeys(txn, rotationResult.rotated);
addKeys(updateResult.current);
// Write any updated keys back to the DB
if (!updateResult.updated.isEmpty())
db.updateTransportKeys(txn, updateResult.updated);
} finally {
lock.unlock();
}
// Schedule the next key rotation
scheduleKeyRotation(now);
// Schedule the next key update
scheduleKeyUpdate(now);
}
private RotationResult rotateKeys(Collection<TransportKeySet> keys,
private UpdateResult updateKeys(Collection<TransportKeySet> keys,
long now) {
RotationResult rotationResult = new RotationResult();
UpdateResult updateResult = new UpdateResult();
long timePeriod = now / timePeriodLength;
for (TransportKeySet ks : keys) {
TransportKeys k = ks.getKeys();
@@ -111,38 +114,45 @@ class TransportKeyManagerImpl implements TransportKeyManager {
TransportKeySet ks1 = new TransportKeySet(ks.getKeySetId(),
ks.getContactId(), null, k1);
if (k1.getTimePeriod() > k.getTimePeriod())
rotationResult.rotated.add(ks1);
rotationResult.current.add(ks1);
updateResult.updated.add(ks1);
updateResult.current.add(ks1);
}
return rotationResult;
return updateResult;
}
@GuardedBy("lock")
private void addKeys(Collection<TransportKeySet> keys) {
for (TransportKeySet ks : keys) {
// TODO: Keys may be for a pending contact
addKeys(ks.getKeySetId(), ks.getContactId(),
ks.getPendingContactId(),
new MutableTransportKeys(ks.getKeys()));
}
}
@GuardedBy("lock")
private void addKeys(KeySetId keySetId, ContactId contactId,
MutableTransportKeys m) {
MutableKeySet ks = new MutableKeySet(keySetId, contactId, m);
keys.put(keySetId, ks);
encodeTags(keySetId, contactId, m.getPreviousIncomingKeys());
encodeTags(keySetId, contactId, m.getCurrentIncomingKeys());
encodeTags(keySetId, contactId, m.getNextIncomingKeys());
private void addKeys(KeySetId keySetId, @Nullable ContactId contactId,
@Nullable PendingContactId pendingContactId,
MutableTransportKeys keys) {
MutableTransportKeySet ks = new MutableTransportKeySet(keySetId,
contactId, pendingContactId, keys);
this.keys.put(keySetId, ks);
boolean handshakeMode = keys.isHandshakeMode();
encodeTags(keySetId, contactId, pendingContactId,
keys.getPreviousIncomingKeys(), handshakeMode);
encodeTags(keySetId, contactId, pendingContactId,
keys.getCurrentIncomingKeys(), handshakeMode);
encodeTags(keySetId, contactId, pendingContactId,
keys.getNextIncomingKeys(), handshakeMode);
considerReplacingOutgoingKeys(ks);
}
@GuardedBy("lock")
private void encodeTags(KeySetId keySetId, ContactId contactId,
MutableIncomingKeys inKeys) {
private void encodeTags(KeySetId keySetId, @Nullable ContactId contactId,
@Nullable PendingContactId pendingContactId,
MutableIncomingKeys inKeys, boolean handshakeMode) {
for (long streamNumber : inKeys.getWindow().getUnseen()) {
TagContext tagCtx =
new TagContext(keySetId, contactId, inKeys, streamNumber);
TagContext tagCtx = new TagContext(keySetId, contactId,
pendingContactId, inKeys, streamNumber, handshakeMode);
byte[] tag = new byte[TAG_LENGTH];
transportCrypto.encodeTag(tag, inKeys.getTagKey(), PROTOCOL_VERSION,
streamNumber);
@@ -151,26 +161,29 @@ class TransportKeyManagerImpl implements TransportKeyManager {
}
@GuardedBy("lock")
private void considerReplacingOutgoingKeys(MutableKeySet ks) {
private void considerReplacingOutgoingKeys(MutableTransportKeySet ks) {
// Use the active outgoing keys with the highest key set ID
if (ks.getTransportKeys().getCurrentOutgoingKeys().isActive()) {
MutableKeySet old = outContexts.get(ks.getContactId());
ContactId c = ks.getContactId();
if (c != null && ks.getKeys().getCurrentOutgoingKeys().isActive()) {
MutableTransportKeySet old = outContexts.get(c);
if (old == null ||
(old.getKeys().isHandshakeMode() &&
!ks.getKeys().isHandshakeMode()) ||
old.getKeySetId().getInt() < ks.getKeySetId().getInt()) {
outContexts.put(ks.getContactId(), ks);
outContexts.put(c, ks);
}
}
}
private void scheduleKeyRotation(long now) {
private void scheduleKeyUpdate(long now) {
long delay = timePeriodLength - now % timePeriodLength;
scheduler.schedule((Runnable) this::rotateKeys, delay, MILLISECONDS);
scheduler.schedule((Runnable) this::updateKeys, delay, MILLISECONDS);
}
private void rotateKeys() {
private void updateKeys() {
dbExecutor.execute(() -> {
try {
db.transaction(false, this::rotateKeys);
db.transaction(false, this::updateKeys);
} catch (DbException e) {
logException(LOG, WARNING, e);
}
@@ -187,13 +200,13 @@ class TransportKeyManagerImpl implements TransportKeyManager {
// Derive the transport keys
TransportKeys k = transportCrypto.deriveRotationKeys(transportId,
rootKey, timePeriod, alice, active);
// Rotate the keys to the current time period if necessary
// Update the keys to the current time period if necessary
timePeriod = clock.currentTimeMillis() / timePeriodLength;
k = transportCrypto.updateTransportKeys(k, timePeriod);
// Write the keys back to the DB
KeySetId keySetId = db.addTransportKeys(txn, c, k);
// Initialise mutable state for the contact
addKeys(keySetId, c, new MutableTransportKeys(k));
addKeys(keySetId, c, null, new MutableTransportKeys(k));
return keySetId;
} finally {
lock.unlock();
@@ -204,9 +217,9 @@ class TransportKeyManagerImpl implements TransportKeyManager {
public void activateKeys(Transaction txn, KeySetId k) throws DbException {
lock.lock();
try {
MutableKeySet ks = keys.get(k);
MutableTransportKeySet ks = keys.get(k);
if (ks == null) throw new IllegalArgumentException();
MutableTransportKeys m = ks.getTransportKeys();
MutableTransportKeys m = ks.getKeys();
m.getCurrentOutgoingKeys().activate();
considerReplacingOutgoingKeys(ks);
db.setTransportKeysActive(txn, m.getTransportId(), k);
@@ -221,13 +234,11 @@ class TransportKeyManagerImpl implements TransportKeyManager {
try {
// Remove mutable state for the contact
Iterator<TagContext> it = inContexts.values().iterator();
while (it.hasNext()) if (it.next().contactId.equals(c)) it.remove();
while (it.hasNext()) if (c.equals(it.next().contactId)) it.remove();
outContexts.remove(c);
Iterator<MutableKeySet> it1 = keys.values().iterator();
while (it1.hasNext()) {
ContactId c1 = it1.next().getContactId();
if (c1 != null && c1.equals(c)) it1.remove();
}
Iterator<MutableTransportKeySet> it1 = keys.values().iterator();
while (it1.hasNext())
if (c.equals(it1.next().getContactId())) it1.remove();
} finally {
lock.unlock();
}
@@ -237,10 +248,10 @@ class TransportKeyManagerImpl implements TransportKeyManager {
public boolean canSendOutgoingStreams(ContactId c) {
lock.lock();
try {
MutableKeySet ks = outContexts.get(c);
MutableTransportKeySet ks = outContexts.get(c);
if (ks == null) return false;
MutableOutgoingKeys outKeys =
ks.getTransportKeys().getCurrentOutgoingKeys();
ks.getKeys().getCurrentOutgoingKeys();
if (!outKeys.isActive()) throw new AssertionError();
return outKeys.getStreamCounter() <= MAX_32_BIT_UNSIGNED;
} finally {
@@ -254,16 +265,16 @@ class TransportKeyManagerImpl implements TransportKeyManager {
lock.lock();
try {
// Look up the outgoing keys for the contact
MutableKeySet ks = outContexts.get(c);
MutableTransportKeySet ks = outContexts.get(c);
if (ks == null) return null;
MutableOutgoingKeys outKeys =
ks.getTransportKeys().getCurrentOutgoingKeys();
MutableTransportKeys keys = ks.getKeys();
MutableOutgoingKeys outKeys = keys.getCurrentOutgoingKeys();
if (!outKeys.isActive()) throw new AssertionError();
if (outKeys.getStreamCounter() > MAX_32_BIT_UNSIGNED) return null;
// Create a stream context
StreamContext ctx = new StreamContext(c, transportId,
StreamContext ctx = new StreamContext(c, null, transportId,
outKeys.getTagKey(), outKeys.getHeaderKey(),
outKeys.getStreamCounter());
outKeys.getStreamCounter(), keys.isHandshakeMode());
// Increment the stream counter and write it back to the DB
outKeys.incrementStreamCounter();
db.incrementStreamCounter(txn, transportId, ks.getKeySetId());
@@ -283,9 +294,10 @@ class TransportKeyManagerImpl implements TransportKeyManager {
if (tagCtx == null) return null;
MutableIncomingKeys inKeys = tagCtx.inKeys;
// Create a stream context
StreamContext ctx = new StreamContext(tagCtx.contactId, transportId,
StreamContext ctx = new StreamContext(tagCtx.contactId,
tagCtx.pendingContactId, transportId,
inKeys.getTagKey(), inKeys.getHeaderKey(),
tagCtx.streamNumber);
tagCtx.streamNumber, tagCtx.handshakeMode);
// Update the reordering window
ReorderingWindow window = inKeys.getWindow();
Change change = window.setSeen(tagCtx.streamNumber);
@@ -295,7 +307,8 @@ class TransportKeyManagerImpl implements TransportKeyManager {
transportCrypto.encodeTag(addTag, inKeys.getTagKey(),
PROTOCOL_VERSION, streamNumber);
TagContext tagCtx1 = new TagContext(tagCtx.keySetId,
tagCtx.contactId, inKeys, streamNumber);
tagCtx.contactId, tagCtx.pendingContactId, inKeys,
streamNumber, tagCtx.handshakeMode);
inContexts.put(new Bytes(addTag), tagCtx1);
}
// Remove tags for any stream numbers removed from the window
@@ -311,9 +324,9 @@ class TransportKeyManagerImpl implements TransportKeyManager {
inKeys.getTimePeriod(), window.getBase(),
window.getBitmap());
// If the outgoing keys are inactive, activate them
MutableKeySet ks = keys.get(tagCtx.keySetId);
MutableTransportKeySet ks = keys.get(tagCtx.keySetId);
MutableOutgoingKeys outKeys =
ks.getTransportKeys().getCurrentOutgoingKeys();
ks.getKeys().getCurrentOutgoingKeys();
if (!outKeys.isActive()) {
LOG.info("Activating outgoing keys");
outKeys.activate();
@@ -326,52 +339,60 @@ class TransportKeyManagerImpl implements TransportKeyManager {
}
}
private void rotateKeys(Transaction txn) throws DbException {
private void updateKeys(Transaction txn) throws DbException {
long now = clock.currentTimeMillis();
lock.lock();
try {
// Rotate the keys to the current time period
// Update the keys to the current time period
Collection<TransportKeySet> snapshot = new ArrayList<>(keys.size());
for (MutableKeySet ks : keys.values()) {
for (MutableTransportKeySet ks : keys.values()) {
snapshot.add(new TransportKeySet(ks.getKeySetId(),
ks.getContactId(), null,
ks.getTransportKeys().snapshot()));
ks.getContactId(), ks.getPendingContactId(),
ks.getKeys().snapshot()));
}
RotationResult rotationResult = rotateKeys(snapshot, now);
UpdateResult updateResult = updateKeys(snapshot, now);
// Rebuild the mutable state for all contacts
inContexts.clear();
outContexts.clear();
keys.clear();
addKeys(rotationResult.current);
// Write any rotated keys back to the DB
if (!rotationResult.rotated.isEmpty())
db.updateTransportKeys(txn, rotationResult.rotated);
addKeys(updateResult.current);
// Write any updated keys back to the DB
if (!updateResult.updated.isEmpty())
db.updateTransportKeys(txn, updateResult.updated);
} finally {
lock.unlock();
}
// Schedule the next key rotation
scheduleKeyRotation(now);
// Schedule the next key update
scheduleKeyUpdate(now);
}
private static class TagContext {
private final KeySetId keySetId;
@Nullable
private final ContactId contactId;
@Nullable
private final PendingContactId pendingContactId;
private final MutableIncomingKeys inKeys;
private final long streamNumber;
private final boolean handshakeMode;
private TagContext(KeySetId keySetId, ContactId contactId,
MutableIncomingKeys inKeys, long streamNumber) {
private TagContext(KeySetId keySetId, @Nullable ContactId contactId,
@Nullable PendingContactId pendingContactId,
MutableIncomingKeys inKeys, long streamNumber,
boolean handshakeMode) {
this.keySetId = keySetId;
this.contactId = contactId;
this.pendingContactId = pendingContactId;
this.inKeys = inKeys;
this.streamNumber = streamNumber;
this.handshakeMode = handshakeMode;
}
}
private static class RotationResult {
private static class UpdateResult {
private final Collection<TransportKeySet> current = new ArrayList<>();
private final Collection<TransportKeySet> rotated = new ArrayList<>();
private final Collection<TransportKeySet> updated = new ArrayList<>();
}
}

View File

@@ -101,8 +101,8 @@ public class SyncIntegrationTest extends BrambleTestCase {
private byte[] write() throws Exception {
ByteArrayOutputStream out = new ByteArrayOutputStream();
StreamContext ctx = new StreamContext(contactId, transportId, tagKey,
headerKey, streamNumber);
StreamContext ctx = new StreamContext(contactId, null, transportId,
tagKey, headerKey, streamNumber, false);
StreamWriter streamWriter = streamWriterFactory.createStreamWriter(out,
ctx);
SyncRecordWriter recordWriter = recordWriterFactory.createRecordWriter(
@@ -131,8 +131,8 @@ public class SyncIntegrationTest extends BrambleTestCase {
assertArrayEquals(expectedTag, tag);
// Create the readers
StreamContext ctx = new StreamContext(contactId, transportId, tagKey,
headerKey, streamNumber);
StreamContext ctx = new StreamContext(contactId, null, transportId,
tagKey, headerKey, streamNumber, false);
InputStream streamReader = streamReaderFactory.createStreamReader(in,
ctx);
SyncRecordReader recordReader = recordReaderFactory.createRecordReader(

View File

@@ -47,7 +47,7 @@ public class KeyManagerImplTest extends BrambleMockTestCase {
private final TransportId transportId = getTransportId();
private final TransportId unknownTransportId = getTransportId();
private final StreamContext streamContext = new StreamContext(contactId,
transportId, getSecretKey(), getSecretKey(), 1);
null, transportId, getSecretKey(), getSecretKey(), 1, false);
private final byte[] tag = getRandomBytes(TAG_LENGTH);
private final Random random = new Random();