Add ConnectionManager method for incoming mailbox connections.

This commit is contained in:
akwizgran
2022-06-01 14:46:23 +01:00
parent de63a50662
commit 6aa24af94c
4 changed files with 85 additions and 9 deletions

View File

@@ -16,6 +16,17 @@ public interface ConnectionManager {
*/ */
void manageIncomingConnection(TransportId t, TransportConnectionReader r); void manageIncomingConnection(TransportId t, TransportConnectionReader r);
/**
* Manages an incoming connection from a contact via a mailbox.
* <p>
* This method does not mark the tag as recognised until after the data
* has been read from the {@link TransportConnectionReader}, at which
* point the {@link TagController} is called to decide whether the tag
* should be marked as recognised.
*/
void manageIncomingConnection(TransportId t, TransportConnectionReader r,
TagController c);
/** /**
* Manages an incoming connection from a contact over a duplex transport. * Manages an incoming connection from a contact over a duplex transport.
*/ */
@@ -46,4 +57,21 @@ public interface ConnectionManager {
*/ */
void manageOutgoingConnection(PendingContactId p, TransportId t, void manageOutgoingConnection(PendingContactId p, TransportId t,
DuplexTransportConnection d); DuplexTransportConnection d);
/**
* An interface for controlling whether a tag should be marked as
* recognised.
*/
interface TagController {
/**
* This method is only called if a tag was read from the corresponding
* {@link TransportConnectionReader} and recognised.
*
* @param exception True if an exception was thrown while reading from
* the {@link TransportConnectionReader}, after successfully reading
* and recognising the tag.
* @return True if the tag should be marked as recognised.
*/
boolean shouldMarkTagAsRecognised(boolean exception);
}
} }

View File

@@ -54,7 +54,7 @@ abstract class Connection {
} }
} }
private byte[] readTag(InputStream in) throws IOException { byte[] readTag(InputStream in) throws IOException {
byte[] tag = new byte[TAG_LENGTH]; byte[] tag = new byte[TAG_LENGTH];
read(in, tag); read(in, tag);
return tag; return tag;

View File

@@ -67,7 +67,15 @@ class ConnectionManagerImpl implements ConnectionManager {
TransportConnectionReader r) { TransportConnectionReader r) {
ioExecutor.execute(new IncomingSimplexSyncConnection(keyManager, ioExecutor.execute(new IncomingSimplexSyncConnection(keyManager,
connectionRegistry, streamReaderFactory, streamWriterFactory, connectionRegistry, streamReaderFactory, streamWriterFactory,
syncSessionFactory, transportPropertyManager, t, r)); syncSessionFactory, transportPropertyManager, t, r, null));
}
@Override
public void manageIncomingConnection(TransportId t,
TransportConnectionReader r, TagController c) {
ioExecutor.execute(new IncomingSimplexSyncConnection(keyManager,
connectionRegistry, streamReaderFactory, streamWriterFactory,
syncSessionFactory, transportPropertyManager, t, r, c));
} }
@Override @Override

View File

@@ -1,7 +1,9 @@
package org.briarproject.bramble.connection; package org.briarproject.bramble.connection;
import org.briarproject.bramble.api.connection.ConnectionManager.TagController;
import org.briarproject.bramble.api.connection.ConnectionRegistry; import org.briarproject.bramble.api.connection.ConnectionRegistry;
import org.briarproject.bramble.api.contact.ContactId; import org.briarproject.bramble.api.contact.ContactId;
import org.briarproject.bramble.api.db.DbException;
import org.briarproject.bramble.api.nullsafety.NotNullByDefault; import org.briarproject.bramble.api.nullsafety.NotNullByDefault;
import org.briarproject.bramble.api.plugin.TransportConnectionReader; import org.briarproject.bramble.api.plugin.TransportConnectionReader;
import org.briarproject.bramble.api.plugin.TransportId; import org.briarproject.bramble.api.plugin.TransportId;
@@ -15,7 +17,10 @@ import org.briarproject.bramble.api.transport.StreamWriterFactory;
import java.io.IOException; import java.io.IOException;
import javax.annotation.Nullable;
import static java.util.logging.Level.WARNING; import static java.util.logging.Level.WARNING;
import static org.briarproject.bramble.api.nullsafety.NullSafety.requireNonNull;
import static org.briarproject.bramble.util.LogUtils.logException; import static org.briarproject.bramble.util.LogUtils.logException;
@NotNullByDefault @NotNullByDefault
@@ -23,6 +28,8 @@ class IncomingSimplexSyncConnection extends SyncConnection implements Runnable {
private final TransportId transportId; private final TransportId transportId;
private final TransportConnectionReader reader; private final TransportConnectionReader reader;
@Nullable
private final TagController tagController;
IncomingSimplexSyncConnection(KeyManager keyManager, IncomingSimplexSyncConnection(KeyManager keyManager,
ConnectionRegistry connectionRegistry, ConnectionRegistry connectionRegistry,
@@ -30,33 +37,50 @@ class IncomingSimplexSyncConnection extends SyncConnection implements Runnable {
StreamWriterFactory streamWriterFactory, StreamWriterFactory streamWriterFactory,
SyncSessionFactory syncSessionFactory, SyncSessionFactory syncSessionFactory,
TransportPropertyManager transportPropertyManager, TransportPropertyManager transportPropertyManager,
TransportId transportId, TransportConnectionReader reader) { TransportId transportId,
TransportConnectionReader reader,
@Nullable TagController tagController) {
super(keyManager, connectionRegistry, streamReaderFactory, super(keyManager, connectionRegistry, streamReaderFactory,
streamWriterFactory, syncSessionFactory, streamWriterFactory, syncSessionFactory,
transportPropertyManager); transportPropertyManager);
this.transportId = transportId; this.transportId = transportId;
this.reader = reader; this.reader = reader;
this.tagController = tagController;
} }
@Override @Override
public void run() { public void run() {
// Read and recognise the tag // Read and recognise the tag
StreamContext ctx = recogniseTag(reader, transportId); byte[] tag = null;
StreamContext ctx;
try {
tag = readTag(reader.getInputStream());
// If we have a tag controller, defer marking the tag as recognised
if (tagController == null) {
ctx = keyManager.getStreamContext(transportId, tag);
} else {
ctx = keyManager.getStreamContextOnly(transportId, tag);
}
} catch (IOException | DbException e) {
logException(LOG, WARNING, e);
onError(false, tag);
return;
}
if (ctx == null) { if (ctx == null) {
LOG.info("Unrecognised tag"); LOG.info("Unrecognised tag");
onError(false); onError(false, tag);
return; return;
} }
ContactId contactId = ctx.getContactId(); ContactId contactId = ctx.getContactId();
if (contactId == null) { if (contactId == null) {
LOG.warning("Received rendezvous stream, expected contact"); LOG.warning("Received rendezvous stream, expected contact");
onError(true); onError(true, tag);
return; return;
} }
if (ctx.isHandshakeMode()) { if (ctx.isHandshakeMode()) {
// TODO: Support handshake mode for contacts // TODO: Support handshake mode for contacts
LOG.warning("Received handshake tag, expected rotation mode"); LOG.warning("Received handshake tag, expected rotation mode");
onError(true); onError(true, tag);
return; return;
} }
try { try {
@@ -65,15 +89,31 @@ class IncomingSimplexSyncConnection extends SyncConnection implements Runnable {
LOG.info("Ignoring priority for simplex connection"); LOG.info("Ignoring priority for simplex connection");
// Create and run the incoming session // Create and run the incoming session
createIncomingSession(ctx, reader, handler).run(); createIncomingSession(ctx, reader, handler).run();
// Success
markTagAsRecognisedIfRequired(false, tag);
reader.dispose(false, true); reader.dispose(false, true);
} catch (IOException e) { } catch (IOException e) {
logException(LOG, WARNING, e); logException(LOG, WARNING, e);
onError(true); onError(true, tag);
} }
} }
private void onError(boolean recognised) { private void onError(boolean recognised, @Nullable byte[] tag) {
if (recognised) {
markTagAsRecognisedIfRequired(true, requireNonNull(tag));
}
disposeOnError(reader, recognised); disposeOnError(reader, recognised);
} }
private void markTagAsRecognisedIfRequired(boolean exception, byte[] tag) {
if (tagController != null &&
tagController.shouldMarkTagAsRecognised(exception)) {
try {
keyManager.markTagAsRecognised(transportId, tag);
} catch (DbException e) {
logException(LOG, WARNING, e);
}
}
}
} }