Merge branch '205-unit-tests-for-keymanagerimpl-and-transportkeymanager' into 'master'

Add unit tests for KeyManagerImpl

This also creates a `TransportKeyManager` interface and a factory for that to be able to test things separately.

Closes #205

See merge request !380

briar-core/src/org/briarproject/transport/KeyManagerImpl.java

@@ -3,7 +3,6 @@ package org.briarproject.transport;
 import org.briarproject.api.TransportId;
 import org.briarproject.api.contact.Contact;
 import org.briarproject.api.contact.ContactId;
-import org.briarproject.api.crypto.CryptoComponent;
 import org.briarproject.api.crypto.SecretKey;
 import org.briarproject.api.db.DatabaseComponent;
 import org.briarproject.api.db.DatabaseExecutor;
@@ -18,7 +17,6 @@ import org.briarproject.api.lifecycle.ServiceException;
 import org.briarproject.api.plugins.PluginConfig;
 import org.briarproject.api.plugins.duplex.DuplexPluginFactory;
 import org.briarproject.api.plugins.simplex.SimplexPluginFactory;
-import org.briarproject.api.system.Clock;
 import org.briarproject.api.transport.KeyManager;
 import org.briarproject.api.transport.StreamContext;
 
@@ -27,7 +25,6 @@ import java.util.Map;
 import java.util.Map.Entry;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Logger;
 
@@ -41,26 +38,21 @@ class KeyManagerImpl implements KeyManager, Service, EventListener {
 			Logger.getLogger(KeyManagerImpl.class.getName());
 
 	private final DatabaseComponent db;
-	private final CryptoComponent crypto;
 	private final Executor dbExecutor;
-	private final ScheduledExecutorService scheduler;
 	private final PluginConfig pluginConfig;
-	private final Clock clock;
+	private final TransportKeyManagerFactory transportKeyManagerFactory;
 	private final Map<ContactId, Boolean> activeContacts;
 	private final ConcurrentHashMap<TransportId, TransportKeyManager> managers;
 	private final AtomicBoolean used = new AtomicBoolean(false);
 
 	@Inject
-	KeyManagerImpl(DatabaseComponent db, CryptoComponent crypto,
-			@DatabaseExecutor Executor dbExecutor,
-			ScheduledExecutorService scheduler, PluginConfig pluginConfig,
-			Clock clock) {
+	KeyManagerImpl(DatabaseComponent db, @DatabaseExecutor Executor dbExecutor,
+			PluginConfig pluginConfig,
+			TransportKeyManagerFactory transportKeyManagerFactory) {
 		this.db = db;
-		this.crypto = crypto;
 		this.dbExecutor = dbExecutor;
-		this.scheduler = scheduler;
 		this.pluginConfig = pluginConfig;
-		this.clock = clock;
+		this.transportKeyManagerFactory = transportKeyManagerFactory;
 		// Use a ConcurrentHashMap as a thread-safe set
 		activeContacts = new ConcurrentHashMap<ContactId, Boolean>();
 		managers = new ConcurrentHashMap<TransportId, TransportKeyManager>();
@@ -83,9 +75,9 @@ class KeyManagerImpl implements KeyManager, Service, EventListener {
 				for (Entry<TransportId, Integer> e : transports.entrySet())
 					db.addTransport(txn, e.getKey(), e.getValue());
 				for (Entry<TransportId, Integer> e : transports.entrySet()) {
-					TransportKeyManager m = new TransportKeyManager(db, crypto,
-							dbExecutor, scheduler, clock, e.getKey(),
-							e.getValue());
+					TransportKeyManager m = transportKeyManagerFactory
+							.createTransportKeyManager(e.getKey(),
+									e.getValue());
 					managers.put(e.getKey(), m);
 					m.start(txn);
 				}

briar-core/src/org/briarproject/transport/TransportKeyManager.java

@@ -1,303 +1,24 @@
 package org.briarproject.transport;
 
-import org.briarproject.api.Bytes;
-import org.briarproject.api.TransportId;
 import org.briarproject.api.contact.ContactId;
-import org.briarproject.api.crypto.CryptoComponent;
 import org.briarproject.api.crypto.SecretKey;
-import org.briarproject.api.db.DatabaseComponent;
 import org.briarproject.api.db.DbException;
 import org.briarproject.api.db.Transaction;
-import org.briarproject.api.system.Clock;
 import org.briarproject.api.transport.StreamContext;
-import org.briarproject.api.transport.TransportKeys;
-import org.briarproject.transport.ReorderingWindow.Change;
 
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.concurrent.Executor;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.locks.ReentrantLock;
-import java.util.logging.Logger;
+interface TransportKeyManager {
 
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import static java.util.logging.Level.WARNING;
-import static org.briarproject.api.transport.TransportConstants.MAX_CLOCK_DIFFERENCE;
-import static org.briarproject.api.transport.TransportConstants.TAG_LENGTH;
-import static org.briarproject.util.ByteUtils.MAX_32_BIT_UNSIGNED;
-
-class TransportKeyManager {
-
-	private static final Logger LOG =
-			Logger.getLogger(TransportKeyManager.class.getName());
-
-	private final DatabaseComponent db;
-	private final CryptoComponent crypto;
-	private final Executor dbExecutor;
-	private final ScheduledExecutorService scheduler;
-	private final Clock clock;
-	private final TransportId transportId;
-	private final long rotationPeriodLength;
-	private final ReentrantLock lock;
-
-	// The following are locking: lock
-	private final Map<Bytes, TagContext> inContexts;
-	private final Map<ContactId, MutableOutgoingKeys> outContexts;
-	private final Map<ContactId, MutableTransportKeys> keys;
-
-	TransportKeyManager(DatabaseComponent db, CryptoComponent crypto,
-			Executor dbExecutor, ScheduledExecutorService scheduler,
-			Clock clock, TransportId transportId, long maxLatency) {
-		this.db = db;
-		this.crypto = crypto;
-		this.dbExecutor = dbExecutor;
-		this.scheduler = scheduler;
-		this.clock = clock;
-		this.transportId = transportId;
-		rotationPeriodLength = maxLatency + MAX_CLOCK_DIFFERENCE;
-		lock = new ReentrantLock();
-		inContexts = new HashMap<Bytes, TagContext>();
-		outContexts = new HashMap<ContactId, MutableOutgoingKeys>();
-		keys = new HashMap<ContactId, MutableTransportKeys>();
-	}
-
-	void start(Transaction txn) throws DbException {
-		long now = clock.currentTimeMillis();
-		lock.lock();
-		try {
-			// Load the transport keys from the DB
-			Map<ContactId, TransportKeys> loaded =
-					db.getTransportKeys(txn, transportId);
-			// Rotate the keys to the current rotation period
-			RotationResult rotationResult = rotateKeys(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);
-		} finally {
-			lock.unlock();
-		}
-		// Schedule the next key rotation
-		scheduleKeyRotation(now);
-	}
-
-	private RotationResult rotateKeys(Map<ContactId, TransportKeys> keys,
-			long now) {
-		RotationResult rotationResult = new RotationResult();
-		long rotationPeriod = now / rotationPeriodLength;
-		for (Entry<ContactId, TransportKeys> e : keys.entrySet()) {
-			ContactId c = e.getKey();
-			TransportKeys k = e.getValue();
-			TransportKeys k1 = crypto.rotateTransportKeys(k, rotationPeriod);
-			if (k1.getRotationPeriod() > k.getRotationPeriod())
-				rotationResult.rotated.put(c, k1);
-			rotationResult.current.put(c, k1);
-		}
-		return rotationResult;
-	}
-
-	// Locking: lock
-	private void addKeys(Map<ContactId, TransportKeys> m) {
-		for (Entry<ContactId, TransportKeys> e : m.entrySet())
-			addKeys(e.getKey(), new MutableTransportKeys(e.getValue()));
-	}
-
-	// Locking: lock
-	private void addKeys(ContactId c, MutableTransportKeys m) {
-		encodeTags(c, m.getPreviousIncomingKeys());
-		encodeTags(c, m.getCurrentIncomingKeys());
-		encodeTags(c, m.getNextIncomingKeys());
-		outContexts.put(c, m.getCurrentOutgoingKeys());
-		keys.put(c, m);
-	}
-
-	// Locking: lock
-	private void encodeTags(ContactId c, MutableIncomingKeys inKeys) {
-		for (long streamNumber : inKeys.getWindow().getUnseen()) {
-			TagContext tagCtx = new TagContext(c, inKeys, streamNumber);
-			byte[] tag = new byte[TAG_LENGTH];
-			crypto.encodeTag(tag, inKeys.getTagKey(), streamNumber);
-			inContexts.put(new Bytes(tag), tagCtx);
-		}
-	}
-
-	private void scheduleKeyRotation(long now) {
-		Runnable task = new Runnable() {
-			@Override
-			public void run() {
-				rotateKeys();
-			}
-		};
-		long delay = rotationPeriodLength - now % rotationPeriodLength;
-		scheduler.schedule(task, delay, MILLISECONDS);
-	}
-
-	private void rotateKeys() {
-		dbExecutor.execute(new Runnable() {
-			@Override
-			public void run() {
-				try {
-					Transaction txn = db.startTransaction(false);
-					try {
-						rotateKeys(txn);
-						db.commitTransaction(txn);
-					} finally {
-						db.endTransaction(txn);
-					}
-				} catch (DbException e) {
-					if (LOG.isLoggable(WARNING))
-						LOG.log(WARNING, e.toString(), e);
-				}
-			}
-		});
-	}
+	void start(Transaction txn) throws DbException;
 
 	void addContact(Transaction txn, ContactId c, SecretKey master,
-			long timestamp, boolean alice) throws DbException {
-		lock.lock();
-		try {
-			// Work out what rotation period the timestamp belongs to
-			long rotationPeriod = timestamp / rotationPeriodLength;
-			// Derive the transport keys
-			TransportKeys k = crypto.deriveTransportKeys(transportId, master,
-					rotationPeriod, alice);
-			// Rotate the keys to the current rotation period if necessary
-			rotationPeriod = clock.currentTimeMillis() / rotationPeriodLength;
-			k = crypto.rotateTransportKeys(k, rotationPeriod);
-			// Initialise mutable state for the contact
-			addKeys(c, new MutableTransportKeys(k));
-			// Write the keys back to the DB
-			db.addTransportKeys(txn, c, k);
-		} finally {
-			lock.unlock();
-		}
-	}
+			long timestamp, boolean alice) throws DbException;
 
-	void removeContact(ContactId c) {
-		lock.lock();
-		try {
-			// Remove mutable state for the contact
-			Iterator<Entry<Bytes, TagContext>> it =
-					inContexts.entrySet().iterator();
-			while (it.hasNext())
-				if (it.next().getValue().contactId.equals(c)) it.remove();
-			outContexts.remove(c);
-			keys.remove(c);
-		} finally {
-			lock.unlock();
-		}
-	}
+	void removeContact(ContactId c);
 
 	StreamContext getStreamContext(Transaction txn, ContactId c)
-			throws DbException {
-		lock.lock();
-		try {
-			// Look up the outgoing keys for the contact
-			MutableOutgoingKeys outKeys = outContexts.get(c);
-			if (outKeys == null) return null;
-			if (outKeys.getStreamCounter() > MAX_32_BIT_UNSIGNED) return null;
-			// Create a stream context
-			StreamContext ctx = new StreamContext(c, transportId,
-					outKeys.getTagKey(), outKeys.getHeaderKey(),
-					outKeys.getStreamCounter());
-			// Increment the stream counter and write it back to the DB
-			outKeys.incrementStreamCounter();
-			db.incrementStreamCounter(txn, c, transportId,
-					outKeys.getRotationPeriod());
-			return ctx;
-		} finally {
-			lock.unlock();
-		}
-	}
+			throws DbException;
 
 	StreamContext getStreamContext(Transaction txn, byte[] tag)
-			throws DbException {
-		lock.lock();
-		try {
-			// Look up the incoming keys for the tag
-			TagContext tagCtx = inContexts.remove(new Bytes(tag));
-			if (tagCtx == null) return null;
-			MutableIncomingKeys inKeys = tagCtx.inKeys;
-			// Create a stream context
-			StreamContext ctx = new StreamContext(tagCtx.contactId, transportId,
-					inKeys.getTagKey(), inKeys.getHeaderKey(),
-					tagCtx.streamNumber);
-			// Update the reordering window
-			ReorderingWindow window = inKeys.getWindow();
-			Change change = window.setSeen(tagCtx.streamNumber);
-			// Add tags for any stream numbers added to the window
-			for (long streamNumber : change.getAdded()) {
-				byte[] addTag = new byte[TAG_LENGTH];
-				crypto.encodeTag(addTag, inKeys.getTagKey(), streamNumber);
-				inContexts.put(new Bytes(addTag), new TagContext(
-						tagCtx.contactId, inKeys, streamNumber));
-			}
-			// Remove tags for any stream numbers removed from the window
-			for (long streamNumber : change.getRemoved()) {
-				if (streamNumber == tagCtx.streamNumber) continue;
-				byte[] removeTag = new byte[TAG_LENGTH];
-				crypto.encodeTag(removeTag, inKeys.getTagKey(), streamNumber);
-				inContexts.remove(new Bytes(removeTag));
-			}
-			// Write the window back to the DB
-			db.setReorderingWindow(txn, tagCtx.contactId, transportId,
-					inKeys.getRotationPeriod(), window.getBase(),
-					window.getBitmap());
-			return ctx;
-		} finally {
-			lock.unlock();
-		}
-	}
+			throws DbException;
 
-	private void rotateKeys(Transaction txn) throws DbException {
-		long now = clock.currentTimeMillis();
-		lock.lock();
-		try {
-			// Rotate the keys to the current rotation period
-			Map<ContactId, TransportKeys> snapshot =
-					new HashMap<ContactId, TransportKeys>();
-			for (Entry<ContactId, MutableTransportKeys> e : keys.entrySet())
-				snapshot.put(e.getKey(), e.getValue().snapshot());
-			RotationResult rotationResult = rotateKeys(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);
-		} finally {
-			lock.unlock();
-		}
-		// Schedule the next key rotation
-		scheduleKeyRotation(now);
-	}
-
-	private static class TagContext {
-
-		private final ContactId contactId;
-		private final MutableIncomingKeys inKeys;
-		private final long streamNumber;
-
-		private TagContext(ContactId contactId, MutableIncomingKeys inKeys,
-				long streamNumber) {
-			this.contactId = contactId;
-			this.inKeys = inKeys;
-			this.streamNumber = streamNumber;
-		}
-	}
-
-	private static class RotationResult {
-
-		private final Map<ContactId, TransportKeys> current, rotated;
-
-		private RotationResult() {
-			current = new HashMap<ContactId, TransportKeys>();
-			rotated = new HashMap<ContactId, TransportKeys>();
-		}
-	}
 }

briar-core/src/org/briarproject/transport/TransportKeyManagerFactory.java

@@ -0,0 +1,10 @@
+package org.briarproject.transport;
+
+import org.briarproject.api.TransportId;
+
+interface TransportKeyManagerFactory {
+
+	TransportKeyManager createTransportKeyManager(TransportId transportId,
+			long maxLatency);
+
+}

briar-core/src/org/briarproject/transport/TransportKeyManagerFactoryImpl.java

@@ -0,0 +1,41 @@
+package org.briarproject.transport;
+
+import org.briarproject.api.TransportId;
+import org.briarproject.api.crypto.CryptoComponent;
+import org.briarproject.api.db.DatabaseComponent;
+import org.briarproject.api.db.DatabaseExecutor;
+import org.briarproject.api.system.Clock;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledExecutorService;
+
+import javax.inject.Inject;
+
+class TransportKeyManagerFactoryImpl implements
+		TransportKeyManagerFactory {
+
+	private final DatabaseComponent db;
+	private final CryptoComponent crypto;
+	private final Executor dbExecutor;
+	private final ScheduledExecutorService scheduler;
+	private final Clock clock;
+
+	@Inject
+	TransportKeyManagerFactoryImpl(DatabaseComponent db, CryptoComponent crypto,
+			@DatabaseExecutor Executor dbExecutor,
+			ScheduledExecutorService scheduler, Clock clock) {
+		this.db = db;
+		this.crypto = crypto;
+		this.dbExecutor = dbExecutor;
+		this.scheduler = scheduler;
+		this.clock = clock;
+	}
+
+	@Override
+	public TransportKeyManager createTransportKeyManager(
+			TransportId transportId, long maxLatency) {
+		return new TransportKeyManagerImpl(db, crypto, dbExecutor, scheduler,
+				clock, transportId, maxLatency);
+	}
+
+}

briar-core/src/org/briarproject/transport/TransportKeyManagerImpl.java

@@ -0,0 +1,308 @@
+package org.briarproject.transport;
+
+import org.briarproject.api.Bytes;
+import org.briarproject.api.TransportId;
+import org.briarproject.api.contact.ContactId;
+import org.briarproject.api.crypto.CryptoComponent;
+import org.briarproject.api.crypto.SecretKey;
+import org.briarproject.api.db.DatabaseComponent;
+import org.briarproject.api.db.DbException;
+import org.briarproject.api.db.Transaction;
+import org.briarproject.api.system.Clock;
+import org.briarproject.api.transport.StreamContext;
+import org.briarproject.api.transport.TransportKeys;
+import org.briarproject.transport.ReorderingWindow.Change;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.logging.Logger;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.logging.Level.WARNING;
+import static org.briarproject.api.transport.TransportConstants.MAX_CLOCK_DIFFERENCE;
+import static org.briarproject.api.transport.TransportConstants.TAG_LENGTH;
+import static org.briarproject.util.ByteUtils.MAX_32_BIT_UNSIGNED;
+
+class TransportKeyManagerImpl implements TransportKeyManager {
+
+	private static final Logger LOG =
+			Logger.getLogger(TransportKeyManagerImpl.class.getName());
+
+	private final DatabaseComponent db;
+	private final CryptoComponent crypto;
+	private final Executor dbExecutor;
+	private final ScheduledExecutorService scheduler;
+	private final Clock clock;
+	private final TransportId transportId;
+	private final long rotationPeriodLength;
+	private final ReentrantLock lock;
+
+	// The following are locking: lock
+	private final Map<Bytes, TagContext> inContexts;
+	private final Map<ContactId, MutableOutgoingKeys> outContexts;
+	private final Map<ContactId, MutableTransportKeys> keys;
+
+	TransportKeyManagerImpl(DatabaseComponent db, CryptoComponent crypto,
+			Executor dbExecutor, ScheduledExecutorService scheduler,
+			Clock clock, TransportId transportId, long maxLatency) {
+		this.db = db;
+		this.crypto = crypto;
+		this.dbExecutor = dbExecutor;
+		this.scheduler = scheduler;
+		this.clock = clock;
+		this.transportId = transportId;
+		rotationPeriodLength = maxLatency + MAX_CLOCK_DIFFERENCE;
+		lock = new ReentrantLock();
+		inContexts = new HashMap<Bytes, TagContext>();
+		outContexts = new HashMap<ContactId, MutableOutgoingKeys>();
+		keys = new HashMap<ContactId, MutableTransportKeys>();
+	}
+
+	@Override
+	public void start(Transaction txn) throws DbException {
+		long now = clock.currentTimeMillis();
+		lock.lock();
+		try {
+			// Load the transport keys from the DB
+			Map<ContactId, TransportKeys> loaded =
+					db.getTransportKeys(txn, transportId);
+			// Rotate the keys to the current rotation period
+			RotationResult rotationResult = rotateKeys(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);
+		} finally {
+			lock.unlock();
+		}
+		// Schedule the next key rotation
+		scheduleKeyRotation(now);
+	}
+
+	private RotationResult rotateKeys(Map<ContactId, TransportKeys> keys,
+			long now) {
+		RotationResult rotationResult = new RotationResult();
+		long rotationPeriod = now / rotationPeriodLength;
+		for (Entry<ContactId, TransportKeys> e : keys.entrySet()) {
+			ContactId c = e.getKey();
+			TransportKeys k = e.getValue();
+			TransportKeys k1 = crypto.rotateTransportKeys(k, rotationPeriod);
+			if (k1.getRotationPeriod() > k.getRotationPeriod())
+				rotationResult.rotated.put(c, k1);
+			rotationResult.current.put(c, k1);
+		}
+		return rotationResult;
+	}
+
+	// Locking: lock
+	private void addKeys(Map<ContactId, TransportKeys> m) {
+		for (Entry<ContactId, TransportKeys> e : m.entrySet())
+			addKeys(e.getKey(), new MutableTransportKeys(e.getValue()));
+	}
+
+	// Locking: lock
+	private void addKeys(ContactId c, MutableTransportKeys m) {
+		encodeTags(c, m.getPreviousIncomingKeys());
+		encodeTags(c, m.getCurrentIncomingKeys());
+		encodeTags(c, m.getNextIncomingKeys());
+		outContexts.put(c, m.getCurrentOutgoingKeys());
+		keys.put(c, m);
+	}
+
+	// Locking: lock
+	private void encodeTags(ContactId c, MutableIncomingKeys inKeys) {
+		for (long streamNumber : inKeys.getWindow().getUnseen()) {
+			TagContext tagCtx = new TagContext(c, inKeys, streamNumber);
+			byte[] tag = new byte[TAG_LENGTH];
+			crypto.encodeTag(tag, inKeys.getTagKey(), streamNumber);
+			inContexts.put(new Bytes(tag), tagCtx);
+		}
+	}
+
+	private void scheduleKeyRotation(long now) {
+		Runnable task = new Runnable() {
+			@Override
+			public void run() {
+				rotateKeys();
+			}
+		};
+		long delay = rotationPeriodLength - now % rotationPeriodLength;
+		scheduler.schedule(task, delay, MILLISECONDS);
+	}
+
+	private void rotateKeys() {
+		dbExecutor.execute(new Runnable() {
+			@Override
+			public void run() {
+				try {
+					Transaction txn = db.startTransaction(false);
+					try {
+						rotateKeys(txn);
+						db.commitTransaction(txn);
+					} finally {
+						db.endTransaction(txn);
+					}
+				} catch (DbException e) {
+					if (LOG.isLoggable(WARNING))
+						LOG.log(WARNING, e.toString(), e);
+				}
+			}
+		});
+	}
+
+	@Override
+	public void addContact(Transaction txn, ContactId c, SecretKey master,
+			long timestamp, boolean alice) throws DbException {
+		lock.lock();
+		try {
+			// Work out what rotation period the timestamp belongs to
+			long rotationPeriod = timestamp / rotationPeriodLength;
+			// Derive the transport keys
+			TransportKeys k = crypto.deriveTransportKeys(transportId, master,
+					rotationPeriod, alice);
+			// Rotate the keys to the current rotation period if necessary
+			rotationPeriod = clock.currentTimeMillis() / rotationPeriodLength;
+			k = crypto.rotateTransportKeys(k, rotationPeriod);
+			// Initialise mutable state for the contact
+			addKeys(c, new MutableTransportKeys(k));
+			// Write the keys back to the DB
+			db.addTransportKeys(txn, c, k);
+		} finally {
+			lock.unlock();
+		}
+	}
+
+	@Override
+	public void removeContact(ContactId c) {
+		lock.lock();
+		try {
+			// Remove mutable state for the contact
+			Iterator<Entry<Bytes, TagContext>> it =
+					inContexts.entrySet().iterator();
+			while (it.hasNext())
+				if (it.next().getValue().contactId.equals(c)) it.remove();
+			outContexts.remove(c);
+			keys.remove(c);
+		} finally {
+			lock.unlock();
+		}
+	}
+
+	@Override
+	public StreamContext getStreamContext(Transaction txn, ContactId c)
+			throws DbException {
+		lock.lock();
+		try {
+			// Look up the outgoing keys for the contact
+			MutableOutgoingKeys outKeys = outContexts.get(c);
+			if (outKeys == null) return null;
+			if (outKeys.getStreamCounter() > MAX_32_BIT_UNSIGNED) return null;
+			// Create a stream context
+			StreamContext ctx = new StreamContext(c, transportId,
+					outKeys.getTagKey(), outKeys.getHeaderKey(),
+					outKeys.getStreamCounter());
+			// Increment the stream counter and write it back to the DB
+			outKeys.incrementStreamCounter();
+			db.incrementStreamCounter(txn, c, transportId,
+					outKeys.getRotationPeriod());
+			return ctx;
+		} finally {
+			lock.unlock();
+		}
+	}
+
+	@Override
+	public StreamContext getStreamContext(Transaction txn, byte[] tag)
+			throws DbException {
+		lock.lock();
+		try {
+			// Look up the incoming keys for the tag
+			TagContext tagCtx = inContexts.remove(new Bytes(tag));
+			if (tagCtx == null) return null;
+			MutableIncomingKeys inKeys = tagCtx.inKeys;
+			// Create a stream context
+			StreamContext ctx = new StreamContext(tagCtx.contactId, transportId,
+					inKeys.getTagKey(), inKeys.getHeaderKey(),
+					tagCtx.streamNumber);
+			// Update the reordering window
+			ReorderingWindow window = inKeys.getWindow();
+			Change change = window.setSeen(tagCtx.streamNumber);
+			// Add tags for any stream numbers added to the window
+			for (long streamNumber : change.getAdded()) {
+				byte[] addTag = new byte[TAG_LENGTH];
+				crypto.encodeTag(addTag, inKeys.getTagKey(), streamNumber);
+				inContexts.put(new Bytes(addTag), new TagContext(
+						tagCtx.contactId, inKeys, streamNumber));
+			}
+			// Remove tags for any stream numbers removed from the window
+			for (long streamNumber : change.getRemoved()) {
+				if (streamNumber == tagCtx.streamNumber) continue;
+				byte[] removeTag = new byte[TAG_LENGTH];
+				crypto.encodeTag(removeTag, inKeys.getTagKey(), streamNumber);
+				inContexts.remove(new Bytes(removeTag));
+			}
+			// Write the window back to the DB
+			db.setReorderingWindow(txn, tagCtx.contactId, transportId,
+					inKeys.getRotationPeriod(), window.getBase(),
+					window.getBitmap());
+			return ctx;
+		} finally {
+			lock.unlock();
+		}
+	}
+
+	private void rotateKeys(Transaction txn) throws DbException {
+		long now = clock.currentTimeMillis();
+		lock.lock();
+		try {
+			// Rotate the keys to the current rotation period
+			Map<ContactId, TransportKeys> snapshot =
+					new HashMap<ContactId, TransportKeys>();
+			for (Entry<ContactId, MutableTransportKeys> e : keys.entrySet())
+				snapshot.put(e.getKey(), e.getValue().snapshot());
+			RotationResult rotationResult = rotateKeys(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);
+		} finally {
+			lock.unlock();
+		}
+		// Schedule the next key rotation
+		scheduleKeyRotation(now);
+	}
+
+	private static class TagContext {
+
+		private final ContactId contactId;
+		private final MutableIncomingKeys inKeys;
+		private final long streamNumber;
+
+		private TagContext(ContactId contactId, MutableIncomingKeys inKeys,
+				long streamNumber) {
+			this.contactId = contactId;
+			this.inKeys = inKeys;
+			this.streamNumber = streamNumber;
+		}
+	}
+
+	private static class RotationResult {
+
+		private final Map<ContactId, TransportKeys> current, rotated;
+
+		private RotationResult() {
+			current = new HashMap<ContactId, TransportKeys>();
+			rotated = new HashMap<ContactId, TransportKeys>();
+		}
+	}
+}

briar-core/src/org/briarproject/transport/TransportModule.java

@@ -34,6 +34,12 @@ public class TransportModule {
 		return new StreamWriterFactoryImpl(streamEncrypterFactory);
 	}
 
+	@Provides
+	TransportKeyManagerFactory provideTransportKeyManagerFactory(
+			TransportKeyManagerFactoryImpl transportKeyManagerFactory) {
+		return transportKeyManagerFactory;
+	}
+
 	@Provides
 	@Singleton
 	KeyManager provideKeyManager(LifecycleManager lifecycleManager,

briar-tests/src/org/briarproject/transport/KeyManagerImplTest.java

@@ -1,14 +1,203 @@
 package org.briarproject.transport;
 
 import org.briarproject.BriarTestCase;
+import org.briarproject.api.TransportId;
+import org.briarproject.api.contact.Contact;
+import org.briarproject.api.contact.ContactId;
+import org.briarproject.api.crypto.SecretKey;
+import org.briarproject.api.db.DatabaseComponent;
+import org.briarproject.api.db.Transaction;
+import org.briarproject.api.event.ContactRemovedEvent;
+import org.briarproject.api.event.ContactStatusChangedEvent;
+import org.briarproject.api.identity.Author;
+import org.briarproject.api.identity.AuthorId;
+import org.briarproject.api.plugins.PluginConfig;
+import org.briarproject.api.plugins.simplex.SimplexPluginFactory;
+import org.briarproject.api.transport.StreamContext;
+import org.jmock.Expectations;
+import org.jmock.Mockery;
+import org.jmock.lib.concurrent.DeterministicExecutor;
+import org.junit.Before;
 import org.junit.Test;
 
-import static org.junit.Assert.fail;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+
+import static org.briarproject.TestUtils.getRandomBytes;
+import static org.briarproject.TestUtils.getRandomId;
+import static org.briarproject.TestUtils.getSecretKey;
+import static org.briarproject.api.transport.TransportConstants.TAG_LENGTH;
+import static org.junit.Assert.assertEquals;
 
 public class KeyManagerImplTest extends BriarTestCase {
 
-	@Test
-	public void testUnitTestsExist() {
-		fail(); // FIXME: Write tests
+	private final Mockery context = new Mockery();
+	private final KeyManagerImpl keyManager;
+	private final DatabaseComponent db = context.mock(DatabaseComponent.class);
+	private final PluginConfig pluginConfig = context.mock(PluginConfig.class);
+	private final TransportKeyManagerFactory transportKeyManagerFactory =
+			context.mock(TransportKeyManagerFactory.class);
+	private final TransportKeyManager transportKeyManager =
+			context.mock(TransportKeyManager.class);
+	private final DeterministicExecutor executor = new DeterministicExecutor();
+	private final Transaction txn = new Transaction(null, false);
+	private final ContactId contactId = new ContactId(42);
+	private final ContactId inactiveContactId = new ContactId(43);
+	private final TransportId transportId = new TransportId("tId");
+	private final TransportId unknownTransportId = new TransportId("id");
+	private final StreamContext streamContext =
+			new StreamContext(contactId, transportId, getSecretKey(),
+					getSecretKey(), 1);
+	private final byte[] tag = getRandomBytes(TAG_LENGTH);
+
+	public KeyManagerImplTest() {
+		keyManager = new KeyManagerImpl(db, executor, pluginConfig,
+				transportKeyManagerFactory);
 	}
+
+	@Before
+	public void testStartService() throws Exception {
+		final Transaction txn = new Transaction(null, false);
+		AuthorId remoteAuthorId = new AuthorId(getRandomId());
+		Author remoteAuthor = new Author(remoteAuthorId, "author",
+				getRandomBytes(42));
+		AuthorId localAuthorId = new AuthorId(getRandomId());
+		final Collection<Contact> contacts = new ArrayList<>();
+		contacts.add(new Contact(contactId, remoteAuthor, localAuthorId, true,
+				true));
+		contacts.add(new Contact(inactiveContactId, remoteAuthor, localAuthorId,
+				true, false));
+		final SimplexPluginFactory pluginFactory =
+				context.mock(SimplexPluginFactory.class);
+		final Collection<SimplexPluginFactory> factories = Collections
+				.singletonList(pluginFactory);
+		final int maxLatency = 1337;
+
+		context.checking(new Expectations() {{
+			oneOf(pluginConfig).getSimplexFactories();
+			will(returnValue(factories));
+			oneOf(pluginFactory).getId();
+			will(returnValue(transportId));
+			oneOf(pluginFactory).getMaxLatency();
+			will(returnValue(maxLatency));
+			oneOf(db).addTransport(txn, transportId, maxLatency);
+			oneOf(transportKeyManagerFactory)
+					.createTransportKeyManager(transportId, maxLatency);
+			will(returnValue(transportKeyManager));
+			oneOf(pluginConfig).getDuplexFactories();
+			oneOf(db).startTransaction(false);
+			will(returnValue(txn));
+			oneOf(db).getContacts(txn);
+			will(returnValue(contacts));
+			oneOf(transportKeyManager).start(txn);
+			oneOf(db).commitTransaction(txn);
+			oneOf(db).endTransaction(txn);
+		}});
+
+		keyManager.startService();
+	}
+
+	@Test
+	public void testAddContact() throws Exception {
+		final SecretKey secretKey = getSecretKey();
+		final long timestamp = 42L;
+		final boolean alice =  true;
+
+		context.checking(new Expectations() {{
+			oneOf(transportKeyManager)
+					.addContact(txn, contactId, secretKey, timestamp, alice);
+		}});
+
+		keyManager.addContact(txn, contactId, secretKey, timestamp, alice);
+		context.assertIsSatisfied();
+	}
+
+	@Test
+	public void testGetStreamContextForInactiveContact() throws Exception {
+		assertEquals(null,
+				keyManager.getStreamContext(inactiveContactId, transportId));
+	}
+
+	@Test
+	public void testGetStreamContextForUnknownTransport() throws Exception {
+		assertEquals(null, keyManager
+				.getStreamContext(contactId, unknownTransportId));
+	}
+
+	@Test
+	public void testGetStreamContextForContact() throws Exception {
+		context.checking(new Expectations() {{
+			oneOf(db).startTransaction(false);
+			will(returnValue(txn));
+			oneOf(transportKeyManager).getStreamContext(txn, contactId);
+			will(returnValue(streamContext));
+			oneOf(db).commitTransaction(txn);
+			oneOf(db).endTransaction(txn);
+		}});
+
+		assertEquals(streamContext,
+				keyManager.getStreamContext(contactId, transportId));
+		context.assertIsSatisfied();
+	}
+
+	@Test
+	public void testGetStreamContextForTagAndUnknownTransport()
+			throws Exception {
+		assertEquals(null,
+				keyManager.getStreamContext(unknownTransportId, tag));
+	}
+
+	@Test
+	public void testGetStreamContextForTag() throws Exception {
+		context.checking(new Expectations() {{
+			oneOf(db).startTransaction(false);
+			will(returnValue(txn));
+			oneOf(transportKeyManager).getStreamContext(txn, tag);
+			will(returnValue(streamContext));
+			oneOf(db).commitTransaction(txn);
+			oneOf(db).endTransaction(txn);
+		}});
+
+		assertEquals(streamContext,
+				keyManager.getStreamContext(transportId, tag));
+		context.assertIsSatisfied();
+	}
+
+	@Test
+	public void testContactRemovedEvent() throws Exception {
+		ContactRemovedEvent event = new ContactRemovedEvent(contactId);
+
+		context.checking(new Expectations() {{
+			oneOf(transportKeyManager).removeContact(contactId);
+		}});
+
+		keyManager.eventOccurred(event);
+		executor.runUntilIdle();
+		assertEquals(null, keyManager.getStreamContext(contactId, transportId));
+
+		context.assertIsSatisfied();
+	}
+
+	@Test
+	public void testContactStatusChangedEvent() throws Exception {
+		ContactStatusChangedEvent event =
+				new ContactStatusChangedEvent(inactiveContactId, true);
+
+		context.checking(new Expectations() {{
+			oneOf(db).startTransaction(false);
+			will(returnValue(txn));
+			oneOf(transportKeyManager).getStreamContext(txn, inactiveContactId);
+			will(returnValue(streamContext));
+			oneOf(db).commitTransaction(txn);
+			oneOf(db).endTransaction(txn);
+		}});
+
+		keyManager.eventOccurred(event);
+		assertEquals(streamContext,
+				keyManager.getStreamContext(inactiveContactId, transportId));
+
+		context.assertIsSatisfied();
+	}
+
 }

briar-tests/src/org/briarproject/transport/TransportKeyManagerImplTest.java

@@ -40,7 +40,7 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 
-public class TransportKeyManagerTest extends BriarTestCase {
+public class TransportKeyManagerImplTest extends BriarTestCase {
 
 	private final TransportId transportId = new TransportId("id");
 	private final long maxLatency = 30 * 1000; // 30 seconds
@@ -96,7 +96,8 @@ public class TransportKeyManagerTest extends BriarTestCase {
 					with(rotationPeriodLength - 1), with(MILLISECONDS));
 		}});
 
-		TransportKeyManager transportKeyManager = new TransportKeyManager(db,
+		TransportKeyManager
+				transportKeyManager = new TransportKeyManagerImpl(db,
 				crypto, dbExecutor, scheduler, clock, transportId, maxLatency);
 		transportKeyManager.start(txn);
 
@@ -138,7 +139,8 @@ public class TransportKeyManagerTest extends BriarTestCase {
 			oneOf(db).addTransportKeys(txn, contactId, rotated);
 		}});
 
-		TransportKeyManager transportKeyManager = new TransportKeyManager(db,
+		TransportKeyManager
+				transportKeyManager = new TransportKeyManagerImpl(db,
 				crypto, dbExecutor, scheduler, clock, transportId, maxLatency);
 		// The timestamp is 1 ms before the start of rotation period 1000
 		long timestamp = rotationPeriodLength * 1000 - 1;
@@ -161,7 +163,8 @@ public class TransportKeyManagerTest extends BriarTestCase {
 
 		final Transaction txn = new Transaction(null, false);
 
-		TransportKeyManager transportKeyManager = new TransportKeyManager(db,
+		TransportKeyManager
+				transportKeyManager = new TransportKeyManagerImpl(db,
 				crypto, dbExecutor, scheduler, clock, transportId, maxLatency);
 		assertNull(transportKeyManager.getStreamContext(txn, contactId));
 
@@ -205,7 +208,8 @@ public class TransportKeyManagerTest extends BriarTestCase {
 			oneOf(db).addTransportKeys(txn, contactId, transportKeys);
 		}});
 
-		TransportKeyManager transportKeyManager = new TransportKeyManager(db,
+		TransportKeyManager
+				transportKeyManager = new TransportKeyManagerImpl(db,
 				crypto, dbExecutor, scheduler, clock, transportId, maxLatency);
 		// The timestamp is at the start of rotation period 1000
 		long timestamp = rotationPeriodLength * 1000;
@@ -254,7 +258,8 @@ public class TransportKeyManagerTest extends BriarTestCase {
 			oneOf(db).incrementStreamCounter(txn, contactId, transportId, 1000);
 		}});
 
-		TransportKeyManager transportKeyManager = new TransportKeyManager(db,
+		TransportKeyManager
+				transportKeyManager = new TransportKeyManagerImpl(db,
 				crypto, dbExecutor, scheduler, clock, transportId, maxLatency);
 		// The timestamp is at the start of rotation period 1000
 		long timestamp = rotationPeriodLength * 1000;
@@ -310,7 +315,8 @@ public class TransportKeyManagerTest extends BriarTestCase {
 			oneOf(db).addTransportKeys(txn, contactId, transportKeys);
 		}});
 
-		TransportKeyManager transportKeyManager = new TransportKeyManager(db,
+		TransportKeyManager
+				transportKeyManager = new TransportKeyManagerImpl(db,
 				crypto, dbExecutor, scheduler, clock, transportId, maxLatency);
 		// The timestamp is at the start of rotation period 1000
 		long timestamp = rotationPeriodLength * 1000;
@@ -365,7 +371,8 @@ public class TransportKeyManagerTest extends BriarTestCase {
 					1, new byte[REORDERING_WINDOW_SIZE / 8]);
 		}});
 
-		TransportKeyManager transportKeyManager = new TransportKeyManager(db,
+		TransportKeyManager
+				transportKeyManager = new TransportKeyManagerImpl(db,
 				crypto, dbExecutor, scheduler, clock, transportId, maxLatency);
 		// The timestamp is at the start of rotation period 1000
 		long timestamp = rotationPeriodLength * 1000;
@@ -456,7 +463,8 @@ public class TransportKeyManagerTest extends BriarTestCase {
 			oneOf(db).endTransaction(txn1);
 		}});
 
-		TransportKeyManager transportKeyManager = new TransportKeyManager(db,
+		TransportKeyManager
+				transportKeyManager = new TransportKeyManagerImpl(db,
 				crypto, dbExecutor, scheduler, clock, transportId, maxLatency);
 		transportKeyManager.start(txn);