briar/briar-desktop/src/org/briarproject/plugins/bluetooth/BluetoothPlugin.java

package org.briarproject.plugins.bluetooth;

import org.briarproject.api.TransportId;
import org.briarproject.api.contact.ContactId;
import org.briarproject.api.crypto.PseudoRandom;
import org.briarproject.api.keyagreement.KeyAgreementConnection;
import org.briarproject.api.keyagreement.KeyAgreementListener;
import org.briarproject.api.keyagreement.TransportDescriptor;
import org.briarproject.api.plugins.Backoff;
import org.briarproject.api.plugins.duplex.DuplexPlugin;
import org.briarproject.api.plugins.duplex.DuplexPluginCallback;
import org.briarproject.api.plugins.duplex.DuplexTransportConnection;
import org.briarproject.api.properties.TransportProperties;
import org.briarproject.util.OsUtils;
import org.briarproject.util.StringUtils;

import java.io.IOException;
import java.io.InputStream;
import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.logging.Logger;

import javax.bluetooth.BluetoothStateException;
import javax.bluetooth.DiscoveryAgent;
import javax.bluetooth.LocalDevice;
import javax.microedition.io.Connector;
import javax.microedition.io.StreamConnection;
import javax.microedition.io.StreamConnectionNotifier;

import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.logging.Level.INFO;
import static java.util.logging.Level.WARNING;
import static javax.bluetooth.DiscoveryAgent.GIAC;

class BluetoothPlugin implements DuplexPlugin {

	// Share an ID with the Android Bluetooth plugin
	static final TransportId ID = new TransportId("bt");

	private static final Logger LOG =
			Logger.getLogger(BluetoothPlugin.class.getName());
	private static final int UUID_BYTES = 16;

	private static final String PROP_ADDRESS = "address";
	private static final String PROP_UUID = "uuid";

	private final Executor ioExecutor;
	private final SecureRandom secureRandom;
	private final Backoff backoff;
	private final DuplexPluginCallback callback;
	private final int maxLatency;
	private final Semaphore discoverySemaphore = new Semaphore(1);

	private volatile boolean running = false;
	private volatile StreamConnectionNotifier socket = null;
	private volatile LocalDevice localDevice = null;

	BluetoothPlugin(Executor ioExecutor, SecureRandom secureRandom,
			Backoff backoff, DuplexPluginCallback callback, int maxLatency) {
		this.ioExecutor = ioExecutor;
		this.secureRandom = secureRandom;
		this.backoff = backoff;
		this.callback = callback;
		this.maxLatency = maxLatency;
	}

	public TransportId getId() {
		return ID;
	}

	public int getMaxLatency() {
		return maxLatency;
	}

	public int getMaxIdleTime() {
		// Bluetooth detects dead connections so we don't need keepalives
		return Integer.MAX_VALUE;
	}

	public boolean start() throws IOException {
		// Initialise the Bluetooth stack
		try {
			localDevice = LocalDevice.getLocalDevice();
		} catch (UnsatisfiedLinkError e) {
			// On Linux the user may need to install libbluetooth-dev
			if (OsUtils.isLinux())
				callback.showMessage("BLUETOOTH_INSTALL_LIBS");
			return false;
		}
		if (LOG.isLoggable(INFO))
			LOG.info("Local address " + localDevice.getBluetoothAddress());
		running = true;
		bind();
		return true;
	}

	private void bind() {
		ioExecutor.execute(new Runnable() {
			public void run() {
				if (!running) return;
				// Advertise the Bluetooth address to contacts
				TransportProperties p = new TransportProperties();
				p.put(PROP_ADDRESS, localDevice.getBluetoothAddress());
				callback.mergeLocalProperties(p);
				// Bind a server socket to accept connections from contacts
				String url = makeUrl("localhost", getUuid());
				StreamConnectionNotifier ss;
				try {
					ss = (StreamConnectionNotifier) Connector.open(url);
				} catch (IOException e) {
					if (LOG.isLoggable(WARNING))
						LOG.log(WARNING, e.toString(), e);
					return;
				}
				if (!running) {
					tryToClose(ss);
					return;
				}
				socket = ss;
				backoff.reset();
				callback.transportEnabled();
				acceptContactConnections(ss);
			}
		});
	}

	private String makeUrl(String address, String uuid) {
		return "btspp://" + address + ":" + uuid + ";name=RFCOMM";
	}

	private String getUuid() {
		String uuid = callback.getLocalProperties().get(PROP_UUID);
		if (uuid == null) {
			byte[] random = new byte[UUID_BYTES];
			secureRandom.nextBytes(random);
			uuid = UUID.nameUUIDFromBytes(random).toString();
			TransportProperties p = new TransportProperties();
			p.put(PROP_UUID, uuid);
			callback.mergeLocalProperties(p);
		}
		return uuid;
	}

	private void tryToClose(StreamConnectionNotifier ss) {
		try {
			if (ss != null) ss.close();
		} catch (IOException e) {
			if (LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
		} finally {
			callback.transportDisabled();
		}
	}

	private void acceptContactConnections(StreamConnectionNotifier ss) {
		while (true) {
			StreamConnection s;
			try {
				s = ss.acceptAndOpen();
			} catch (IOException e) {
				// This is expected when the socket is closed
				if (LOG.isLoggable(INFO)) LOG.info(e.toString());
				return;
			}
			backoff.reset();
			callback.incomingConnectionCreated(wrapSocket(s));
			if (!running) return;
		}
	}

	private DuplexTransportConnection wrapSocket(StreamConnection s) {
		return new BluetoothTransportConnection(this, s);
	}

	public void stop() {
		running = false;
		tryToClose(socket);
	}

	public boolean isRunning() {
		return running;
	}

	public boolean shouldPoll() {
		return true;
	}

	public int getPollingInterval() {
		return backoff.getPollingInterval();
	}

	public void poll(final Collection<ContactId> connected) {
		if (!running) return;
		backoff.increment();
		// Try to connect to known devices in parallel
		Map<ContactId, TransportProperties> remote =
				callback.getRemoteProperties();
		for (Entry<ContactId, TransportProperties> e : remote.entrySet()) {
			final ContactId c = e.getKey();
			if (connected.contains(c)) continue;
			final String address = e.getValue().get(PROP_ADDRESS);
			if (StringUtils.isNullOrEmpty(address)) continue;
			final String uuid = e.getValue().get(PROP_UUID);
			if (StringUtils.isNullOrEmpty(uuid)) continue;
			ioExecutor.execute(new Runnable() {
				public void run() {
					if (!running) return;
					StreamConnection s = connect(makeUrl(address, uuid));
					if (s != null) {
						backoff.reset();
						callback.outgoingConnectionCreated(c, wrapSocket(s));
					}
				}
			});
		}
	}

	private StreamConnection connect(String url) {
		if (LOG.isLoggable(INFO)) LOG.info("Connecting to " + url);
		try {
			StreamConnection s = (StreamConnection) Connector.open(url);
			if (LOG.isLoggable(INFO)) LOG.info("Connected to " + url);
			return s;
		} catch (IOException e) {
			if (LOG.isLoggable(INFO)) LOG.info("Could not connect to " + url);
			return null;
		}
	}

	public DuplexTransportConnection createConnection(ContactId c) {
		if (!running) return null;
		TransportProperties p = callback.getRemoteProperties().get(c);
		if (p == null) return null;
		String address = p.get(PROP_ADDRESS);
		if (StringUtils.isNullOrEmpty(address)) return null;
		String uuid = p.get(PROP_UUID);
		if (StringUtils.isNullOrEmpty(uuid)) return null;
		String url = makeUrl(address, uuid);
		StreamConnection s = connect(url);
		if (s == null) return null;
		return new BluetoothTransportConnection(this, s);
	}

	public boolean supportsInvitations() {
		return true;
	}

	public DuplexTransportConnection createInvitationConnection(PseudoRandom r,
			long timeout, boolean alice) {
		if (!running) return null;
		// Use the invitation codes to generate the UUID
		byte[] b = r.nextBytes(UUID_BYTES);
		String uuid = UUID.nameUUIDFromBytes(b).toString();
		String url = makeUrl("localhost", uuid);
		// Make the device discoverable if possible
		makeDeviceDiscoverable();
		// Bind a server socket for receiving invitation connections
		final StreamConnectionNotifier ss;
		try {
			ss = (StreamConnectionNotifier) Connector.open(url);
		} catch (IOException e) {
			if (LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
			return null;
		}
		if (!running) {
			tryToClose(ss);
			return null;
		}
		// Create the background tasks
		CompletionService<StreamConnection> complete =
				new ExecutorCompletionService<StreamConnection>(ioExecutor);
		List<Future<StreamConnection>> futures =
				new ArrayList<Future<StreamConnection>>();
		if (alice) {
			// Return the first connected socket
			futures.add(complete.submit(new ListeningTask(ss)));
			futures.add(complete.submit(new DiscoveryTask(uuid)));
		} else {
			// Return the first socket with readable data
			futures.add(complete.submit(new ReadableTask(
					new ListeningTask(ss))));
			futures.add(complete.submit(new ReadableTask(
					new DiscoveryTask(uuid))));
		}
		StreamConnection chosen = null;
		try {
			Future<StreamConnection> f = complete.poll(timeout, MILLISECONDS);
			if (f == null) return null; // No task completed within the timeout
			chosen = f.get();
			return new BluetoothTransportConnection(this, chosen);
		} catch (InterruptedException e) {
			LOG.info("Interrupted while exchanging invitations");
			Thread.currentThread().interrupt();
			return null;
		} catch (ExecutionException e) {
			if (LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
			return null;
		} finally {
			// Closing the socket will terminate the listener task
			tryToClose(ss);
			closeSockets(futures, chosen);
		}
	}

	private void closeSockets(final List<Future<StreamConnection>> futures,
			final StreamConnection chosen) {
		ioExecutor.execute(new Runnable() {
			public void run() {
				for (Future<StreamConnection> f : futures) {
					try {
						if (f.cancel(true)) {
							LOG.info("Cancelled task");
						} else {
							StreamConnection s = f.get();
							if (s != null && s != chosen) {
								LOG.info("Closing unwanted socket");
								s.close();
							}
						}
					} catch (InterruptedException e) {
						LOG.info("Interrupted while closing sockets");
						return;
					} catch (ExecutionException e) {
						if (LOG.isLoggable(INFO)) LOG.info(e.toString());
					} catch (IOException e) {
						if (LOG.isLoggable(INFO)) LOG.info(e.toString());
					}
				}
			}
		});
	}

	public boolean supportsKeyAgreement() {
		return true;
	}

	public KeyAgreementListener createKeyAgreementListener(
			byte[] localCommitment) {
		// No truncation necessary because COMMIT_LENGTH = 16
		String uuid = UUID.nameUUIDFromBytes(localCommitment).toString();
		if (LOG.isLoggable(INFO)) LOG.info("Key agreement UUID " + uuid);
		String url = makeUrl("localhost", uuid);
		// Make the device discoverable if possible
		makeDeviceDiscoverable();
		// Bind a server socket for receiving invitation connections
		final StreamConnectionNotifier ss;
		try {
			ss = (StreamConnectionNotifier) Connector.open(url);
		} catch (IOException e) {
			if (LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
			return null;
		}
		if (!running) {
			tryToClose(ss);
			return null;
		}
		TransportProperties p = new TransportProperties();
		p.put(PROP_ADDRESS, localDevice.getBluetoothAddress());
		TransportDescriptor d = new TransportDescriptor(ID, p);
		return new BluetoothKeyAgreementListener(d, ss);
	}

	public DuplexTransportConnection createKeyAgreementConnection(
			byte[] remoteCommitment, TransportDescriptor d, long timeout) {
		if (!isRunning()) return null;
		if (!ID.equals(d.getIdentifier())) return null;
		TransportProperties p = d.getProperties();
		if (p == null) return null;
		String address = p.get(PROP_ADDRESS);
		if (StringUtils.isNullOrEmpty(address)) return null;
		// No truncation necessary because COMMIT_LENGTH = 16
		String uuid = UUID.nameUUIDFromBytes(remoteCommitment).toString();
		if (LOG.isLoggable(INFO))
			LOG.info("Connecting to key agreement UUID " + uuid);
		String url = makeUrl(address, uuid);
		StreamConnection s = connect(url);
		if (s == null) return null;
		return new BluetoothTransportConnection(this, s);
	}

	private void makeDeviceDiscoverable() {
		// Try to make the device discoverable (requires root on Linux)
		try {
			localDevice.setDiscoverable(GIAC);
		} catch (BluetoothStateException e) {
			if (LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
		}
	}

	private class DiscoveryTask implements Callable<StreamConnection> {

		private final String uuid;

		private DiscoveryTask(String uuid) {
			this.uuid = uuid;
		}

		@Override
		public StreamConnection call() throws Exception {
			// Repeat discovery until we connect or get interrupted
			DiscoveryAgent discoveryAgent = localDevice.getDiscoveryAgent();
			while (true) {
				if (!discoverySemaphore.tryAcquire())
					throw new Exception("Discovery is already in progress");
				try {
					InvitationListener listener =
							new InvitationListener(discoveryAgent, uuid);
					discoveryAgent.startInquiry(GIAC, listener);
					String url = listener.waitForUrl();
					if (url != null) {
						StreamConnection s = connect(url);
						if (s != null) {
							LOG.info("Outgoing connection");
							return s;
						}
					}
				} finally {
					discoverySemaphore.release();
				}
			}
		}
	}

	private static class ListeningTask implements Callable<StreamConnection> {

		private final StreamConnectionNotifier serverSocket;

		private ListeningTask(StreamConnectionNotifier serverSocket) {
			this.serverSocket = serverSocket;
		}

		@Override
		public StreamConnection call() throws Exception  {
			StreamConnection s = serverSocket.acceptAndOpen();
			LOG.info("Incoming connection");
			return s;
		}
	}

	private static class ReadableTask implements Callable<StreamConnection> {

		private final Callable<StreamConnection> connectionTask;

		private ReadableTask(Callable<StreamConnection> connectionTask) {
			this.connectionTask = connectionTask;
		}

		@Override
		public StreamConnection call() throws Exception {
			StreamConnection s = connectionTask.call();
			InputStream in = s.openInputStream();
			while (in.available() == 0) {
				LOG.info("Waiting for data");
				Thread.sleep(1000);
			}
			LOG.info("Data available");
			return s;
		}
	}

	private class BluetoothKeyAgreementListener extends KeyAgreementListener {

		private final StreamConnectionNotifier ss;

		public BluetoothKeyAgreementListener(TransportDescriptor descriptor,
				StreamConnectionNotifier ss) {
			super(descriptor);
			this.ss = ss;
		}

		@Override
		public Callable<KeyAgreementConnection> listen() {
			return new Callable<KeyAgreementConnection>() {
				@Override
				public KeyAgreementConnection call() throws Exception {
					StreamConnection s = ss.acceptAndOpen();
					if (LOG.isLoggable(INFO))
						LOG.info(ID.getString() + ": Incoming connection");
					return new KeyAgreementConnection(
							new BluetoothTransportConnection(
									BluetoothPlugin.this, s), ID);
				}
			};
		}

		@Override
		public void close() {
			try {
				ss.close();
			} catch (IOException e) {
				if (LOG.isLoggable(WARNING)) LOG.log(WARNING, e.toString(), e);
			}
		}
	}
}