nutomic/ensichat
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Merge branch 'aodvv2'

Browse source
This commit is contained in:
Felix Ableitner 2016-05-29 18:38:21 +02:00
commit d97986ae81
43 changed files with 1285 additions and 280 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

90
PROTOCOL.md
View file

@ -26,6 +26,9 @@ Nodes MUST NOT have a public key with the broadcast address or null
address as hash. Additionally, nodes MUST NOT connect to a node with
either address.
All integer fields are in network byte order, and unsigned (unless
specified otherwise).
Crypto
------
@ -40,22 +43,15 @@ private key, and the result written to the 'Encryption Data' part.
Routing
-------
A simple flood routing protocol is currently used. Every node forwards
all messages, unless a message with the same Origin and Sequence Number
has already been received.
The routing protocol is based on
[AODVv2](https://datatracker.ietf.org/doc/draft-ietf-manet-aodvv2/),
with various features left out.
Nodes MUST store pairs of (Origin, Sequence Number) for all received
messages. After receiving a new message, entries with the same Origin
and Sequence Number between _received_ + 1 and _received_ + 32767 MUST
be removed (with a wrap around at the maximum value). The entries MUST
NOT be cleared while the program is running. They MAY be cleared when
the program is exited.
TODO: Add Documentation for routing protocol.
There is currently no support for offline messages. If sender and
receiver are not in the same mesh, the message will not arrive.
Nodes are free implement different routing algorithms.
Messages
--------
@ -84,9 +80,7 @@ AES key is wrapped with the recipient's public RSA key.
### Header
Every message starts with one 74 byte header indicating the message
version, type and ID, followed by the length of the message. The
header is in network byte order, i.e. big endian. The header may have
6 bytes of additional data.
version, type and ID, followed by the length of the message.
0 1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
@ -220,6 +214,74 @@ After this message has been received, communication with normal messages
may start.
### Route Request (Protocol-Type = 2)
Sent to request a route to a specific Target Address.
0 1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| |
| Address (32 bytes) |
| |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| OrigSeqNum | OriginMetric |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| TargMetric |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Equivalent to the Sequence Number in the message header.
Set OrigMetric = RouterClient.Cost for the Router Client entry
which includes OrigAddr.
If an Invalid route exists in the Local Route Set matching
TargAddr using longest prefix matching and has a valid
sequence number, set TargSeqNum = LocalRoute.SeqNum.
Otherwise, set TargSeqNum = -1. This field is signed.
### Route Reply (Protocol-Type = 3)
Sent as a reply when a Route Request arrives, to inform other nodes
about a route.
0 1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| TargSeqNum | TargMetric |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Set TargMetric = RouterClient.Cost for the Router Client entry
which includes TargAddr.
### Route Error (Protocol-Type = 4)
Notifies other nodes of routes that are no longer available. The target
address MUST be set to the broadcast address.
0 1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| |
| Packet Source (32 bytes) |
| |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| |
| Address (32 bytes) |
| |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| SeqNum |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Packet Source is the source address of the message triggering this
Route Error. If the route error is not triggered by a message,
this MUST be set to the null address.
Address is the address that is no longer reachable.
SeqNum is the sequence number of the route that is no longer available
(if known). Otherwise, set TargSeqNum = -1. This field is signed.
Content Messages
----------------

14
README.md
View file

@ -18,10 +18,20 @@ To setup a development environment, just install [Android Studio](https://develo
and import the project.
Alternatively, you can use the command line. To create a debug apk, run `./gradlew assembleDevDebug`.
This requires at least Android Lollipop on your development device. If you don't have Lollipop, you
can alternatively use `./gradlew assembleRelDebug`. However, this results in considerably slower
This requires at least Android Lollipop on your development device. If you don't have 5.0 or higher,
you have to use `./gradlew assembleRelDebug`. However, this results in considerably slower
incremental builds. To create a release apk, run `./gradlew assembleRelRelease`.
Testing
-------
You can run the unit tests with `./gradlew test`. After connecting an Android device, you can run
the Android tests with `./gradlew connectedDevDebugAndroidTest` (or
`./gradlew connectedRelDebugAndroidTest` if your Android version is lower than 5.0).
To run integration tests for the core module, use `./gradlew integration:run`. If this fails (or
is very slow), try changing the value of Crypto#PublicKeySize to 512 (in the core module).
License
-------

18
android/build.gradle
View file

@ -11,12 +11,13 @@ buildscript {
}
dependencies {
compile 'com.android.support:design:23.1.1'
compile 'com.android.support:design:23.4.0'
compile 'com.android.support:multidex:1.0.1'
compile 'org.scala-lang:scala-library:2.11.7'
compile 'com.mobsandgeeks:adapter-kit:0.5.3'
compile 'com.google.zxing:android-integration:3.2.1'
compile 'com.google.zxing:core:3.2.1'
compile 'org.slf4j:slf4j-android:1.7.21'
compile project(path: ':core')
androidTestCompile 'com.android.support:multidex-instrumentation:1.0.1',
{ exclude module: 'multidex' }
@ -45,9 +46,18 @@ android {
testInstrumentationRunner "com.android.test.runner.MultiDexTestRunner"
}
buildTypes.debug {
applicationIdSuffix ".debug"
testCoverageEnabled true
buildTypes {
debug {
applicationIdSuffix ".debug"
testCoverageEnabled true
}
release {
// HACK: This shouldn't be needed, but multidex isn't working correctly.
// https://code.google.com/p/android/issues/detail?id=206131
// https://code.google.com/p/android/issues/detail?id=209084
minifyEnabled true
proguardFiles getDefaultProguardFile('proguard-android.txt'), file('proguard-rules.txt')
}
}
// Increasing minSdkVersion reduces compilation time for MultiDex.

36
android/proguard-rules.txt Normal file
View file

@ -0,0 +1,36 @@
# Add project specific ProGuard rules here.
# By default, the flags in this file are appended to flags specified
# in /home/sg/adt/sdk/tools/proguard/proguard-android.txt
# You can edit the include path and order by changing the ProGuard
# include property in project.properties.
#
# For more details, see
# http://developer.android.com/guide/developing/tools/proguard.html
# Add any project specific keep options here:
# If your project uses WebView with JS, uncomment the following
# and specify the fully qualified class name to the JavaScript interface
# class:
#-keepclassmembers class fqcn.of.javascript.interface.for.webview {
# public *;
#}
-dontobfuscate
-keep class scala.** { *; }
-keep class slick.** { *; }
-keep class org.joda.time.** { *; }
-keep class org.h2.** { *; }
-keep class java.util.** { *; }
-keepclassmembers class java.util.Comparator {
public *;
}
-dontwarn scala.**
-dontwarn slick.**
-dontwarn org.joda.time.**
-dontwarn org.h2.**
-dontwarn java.util.function.**
-ignorewarnings

4
android/src/main/scala/com/nutomic/ensichat/App.scala
View file

@ -1,14 +1,12 @@
package com.nutomic.ensichat
import android.support.multidex.MultiDexApplication
import com.nutomic.ensichat.core.interfaces.Log
import com.nutomic.ensichat.util.{Logging, PRNGFixes}
import com.nutomic.ensichat.util.PRNGFixes
class App extends MultiDexApplication {
override def onCreate(): Unit = {
super.onCreate()
Log.setLogInstance(new Logging())
PRNGFixes.apply()
}

17
android/src/main/scala/com/nutomic/ensichat/bluetooth/BluetoothInterface.scala
View file

@ -170,10 +170,11 @@ class BluetoothInterface(context: Context, mainHandler: Handler,
* Removes device from active connections.
*/
def onConnectionClosed(device: Device, socket: BluetoothSocket): Unit = {
val address = getAddressForDevice(device.id)
devices -= device.id
connections -= device.id
connectionHandler.onConnectionClosed()
addressDeviceMap = addressDeviceMap.filterNot(_._2 == device.id)
connectionHandler.onConnectionClosed(address)
}
/**
@ -192,15 +193,18 @@ class BluetoothInterface(context: Context, mainHandler: Handler,
if (!connectionHandler.onConnectionOpened(msg))
addressDeviceMap -= address
case _ =>
connectionHandler.onMessageReceived(msg)
connectionHandler.onMessageReceived(msg, getAddressForDevice(device))
}
private def getAddressForDevice(device: Device.ID) =
addressDeviceMap.find(_._2 == device).get._1
/**
* Sends the message to nextHop.
*/
override def send(nextHop: Address, msg: Message): Unit = {
addressDeviceMap
.find(_._1 == nextHop)
.find(_._1 == nextHop || Address.Broadcast == nextHop)
.map(i => connections.get(i._2))
.getOrElse(None)
.foreach(_.send(msg))
@ -210,11 +214,6 @@ class BluetoothInterface(context: Context, mainHandler: Handler,
* Returns all active Bluetooth connections.
*/
override def getConnections: Set[Address] =
connections.flatMap { x =>
addressDeviceMap
.find(_._2 == x._1)
.map(_._1)
}
.toSet
connections.map( c => getAddressForDevice(c._1)).toSet
}

13
android/src/main/scala/com/nutomic/ensichat/util/Logging.scala
View file

@ -1,13 +0,0 @@
package com.nutomic.ensichat.util
import android.util
import com.nutomic.ensichat.core.interfaces.Log
class Logging extends Log {
def v(tag: String, message: String, tr: Throwable = null) = util.Log.v(tag, message, tr)
def d(tag: String, message: String, tr: Throwable = null) = util.Log.d(tag, message, tr)
def i(tag: String, message: String, tr: Throwable = null) = util.Log.i(tag, message, tr)
def w(tag: String, message: String, tr: Throwable = null) = util.Log.w(tag, message, tr)
def e(tag: String, message: String, tr: Throwable = null) = util.Log.e(tag, message, tr)
}

2
core/build.gradle
View file

@ -4,6 +4,8 @@ dependencies {
compile 'org.scala-lang:scala-library:2.11.7'
compile 'com.h2database:h2:1.4.191'
compile 'com.typesafe.slick:slick_2.11:3.1.1'
compile 'com.typesafe.scala-logging:scala-logging_2.11:3.4.0'
compile 'joda-time:joda-time:2.9.3'
testCompile 'junit:junit:4.12'
}

15
core/src/main/resources/logback.xml Normal file
View file

@ -0,0 +1,15 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
<target>System.out</target>
<encoder>
<pattern>%d{HH:mm:ss} %level/%logger{0}: %msg%n</pattern>
</encoder>
</appender>
<logger name="slick" level="INFO" />
<root level="DEBUG">
<appender-ref ref="CONSOLE"/>
</root>
</configuration>

5
core/src/main/scala/com/nutomic/ensichat/core/Address.scala
View file

@ -56,4 +56,9 @@ final case class Address(bytes: Array[Byte]) {
.grouped(Address.GroupLength)
.reduce(_ + "-" + _)
/**
* Returns shortened address, useful for debugging.
*/
def short = toString.split("-").head
}

230
core/src/main/scala/com/nutomic/ensichat/core/ConnectionHandler.scala
View file

@ -1,14 +1,18 @@
package com.nutomic.ensichat.core
import java.security.InvalidKeyException
import java.util.Date
import com.nutomic.ensichat.core.body.{ConnectionInfo, MessageBody, UserInfo}
import com.nutomic.ensichat.core.header.ContentHeader
import com.nutomic.ensichat.core.body._
import com.nutomic.ensichat.core.header.{ContentHeader, MessageHeader}
import com.nutomic.ensichat.core.interfaces._
import com.nutomic.ensichat.core.internet.InternetInterface
import com.nutomic.ensichat.core.util.{Database, FutureHelper}
import com.nutomic.ensichat.core.util.{Database, FutureHelper, LocalRoutesInfo, RouteMessageInfo}
import com.typesafe.scalalogging.Logger
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future
import scala.concurrent.duration._
/**
* High-level handling of all message transfers and callbacks.
@ -18,16 +22,33 @@ import scala.concurrent.ExecutionContext.Implicits.global
*/
final class ConnectionHandler(settings: SettingsInterface, database: Database,
callbacks: CallbackInterface, crypto: Crypto,
maxInternetConnections: Int) {
maxInternetConnections: Int,
port: Int = InternetInterface.DefaultPort) {
private val Tag = "ConnectionHandler"
private val logger = Logger(this.getClass)
private val MissingRouteMessageTimeout = 5.minutes
private var transmissionInterfaces = Set[TransmissionInterface]()
private lazy val router = new Router(connections, sendVia)
private lazy val seqNumGenerator = new SeqNumGenerator(settings)
private val localRoutesInfo = new LocalRoutesInfo(connections)
private val routeMessageInfo = new RouteMessageInfo()
private lazy val router = new Router(localRoutesInfo,
(a, m) => transmissionInterfaces.foreach(_.send(a, m)),
noRouteFound)
/**
* Contains messages that couldn't be forwarded because we don't know a route.
*
* These will be buffered until we receive a [[RouteReply]] for the target, or when until the
* message has couldn't be forwarded after [[MissingRouteMessageTimeout]].
*/
private var missingRouteMessages = Set[(Message, Date)]()
/**
* Holds all known users.
*
@ -41,14 +62,15 @@ final class ConnectionHandler(settings: SettingsInterface, database: Database,
* @param additionalInterfaces Instances of [[TransmissionInterface]] to transfer data over
* platform specific interfaces (eg Bluetooth).
*/
def start(additionalInterfaces: Set[TransmissionInterface] = Set()): Unit = {
def start(additionalInterfaces: Set[TransmissionInterface] = Set()): Future[Unit] = {
additionalInterfaces.foreach(transmissionInterfaces += _)
FutureHelper {
crypto.generateLocalKeys()
Log.i(Tag, "Service started, address is " + crypto.localAddress)
Log.i(Tag, "Local user is " + settings.get(SettingsInterface.KeyUserName, "none") +
logger.info("Service started, address is " + crypto.localAddress)
logger.info("Local user is " + settings.get(SettingsInterface.KeyUserName, "none") +
" with status '" + settings.get(SettingsInterface.KeyUserStatus, "") + "'")
transmissionInterfaces += new InternetInterface(this, crypto, settings, maxInternetConnections)
transmissionInterfaces +=
new InternetInterface(this, crypto, settings, maxInternetConnections, port)
transmissionInterfaces.foreach(_.create())
}
}
@ -62,6 +84,7 @@ final class ConnectionHandler(settings: SettingsInterface, database: Database,
* Sends a new message to the given target address.
*/
def sendTo(target: Address, body: MessageBody): Unit = {
assert(body.contentType != -1)
FutureHelper {
val messageId = settings.get("message_id", 0L)
val header = new ContentHeader(crypto.localAddress, target, seqNumGenerator.next(),
@ -75,23 +98,165 @@ final class ConnectionHandler(settings: SettingsInterface, database: Database,
}
}
private def sendVia(nextHop: Address, msg: Message) =
transmissionInterfaces.foreach(_.send(nextHop, msg))
private def requestRoute(target: Address): Unit = {
assert(localRoutesInfo.getRoute(target).isEmpty)
val seqNum = seqNumGenerator.next()
val targetSeqNum = localRoutesInfo.getRoute(target).map(_.seqNum).getOrElse(-1)
val body = new RouteRequest(target, seqNum, targetSeqNum, 0)
val header = new MessageHeader(body.protocolType, crypto.localAddress, Address.Broadcast, seqNum)
val signed = crypto.sign(new Message(header, body))
router.forwardMessage(signed)
}
private def replyRoute(target: Address, replyTo: Address): Unit = {
val seqNum = seqNumGenerator.next()
val body = new RouteReply(seqNum, 0)
val header = new MessageHeader(body.protocolType, crypto.localAddress, replyTo, seqNum)
val signed = crypto.sign(new Message(header, body))
router.forwardMessage(signed)
}
private def routeError(address: Address, packetSource: Option[Address]): Unit = {
val destination = packetSource.getOrElse(Address.Broadcast)
val header = new MessageHeader(RouteError.Type, crypto.localAddress, destination,
seqNumGenerator.next())
val seqNum = localRoutesInfo.getRoute(address).map(_.seqNum).getOrElse(-1)
val body = new RouteError(address, seqNum)
val signed = crypto.sign(new Message(header, body))
router.forwardMessage(signed)
}
/**
* Force connect to a sepcific internet.
*
* @param address An address in the format IP;port or hostname:port.
*/
def connect(address: String): Unit = {
transmissionInterfaces
.find(_.isInstanceOf[InternetInterface])
.map(_.asInstanceOf[InternetInterface])
.foreach(_.openConnection(address))
}
/**
* Decrypts and verifies incoming messages, forwards valid ones to [[onNewMessage()]].
*/
def onMessageReceived(msg: Message): Unit = {
def onMessageReceived(msg: Message, previousHop: Address): Unit = {
if (router.isMessageSeen(msg)) {
Log.v(Tag, "Ignoring message from " + msg.header.origin + " that we already received")
} else if (msg.header.target == crypto.localAddress) {
crypto.verifyAndDecrypt(msg) match {
case Some(m) => onNewMessage(m)
case None => Log.i(Tag, "Ignoring message with invalid signature from " + msg.header.origin)
}
} else {
router.forwardMessage(msg)
logger.trace("Ignoring message from " + msg.header.origin + " that we already received")
return
}
msg.body match {
case rreq: RouteRequest =>
localRoutesInfo.addRoute(msg.header.origin, rreq.originSeqNum, previousHop, rreq.originMetric)
// TODO: Respecting this causes the RERR test to fail. We have to fix the implementation
// of isMessageRedundant() without breaking the test.
if (routeMessageInfo.isMessageRedundant(msg)) {
logger.info("Sending redundant RREQ")
//return
}
if (crypto.localAddress == rreq.requested)
replyRoute(rreq.requested, msg.header.origin)
else {
val body = rreq.copy(originMetric = rreq.originMetric + 1)
val forwardMsg = crypto.sign(new Message(msg.header, body))
localRoutesInfo.getRoute(rreq.requested) match {
case Some(route) => router.forwardMessage(forwardMsg, Option(route.nextHop))
case None => router.forwardMessage(forwardMsg, Option(Address.Broadcast))
}
}
return
case rrep: RouteReply =>
localRoutesInfo.addRoute(msg.header.origin, rrep.originSeqNum, previousHop, 0)
// TODO: See above (in RREQ handler).
if (routeMessageInfo.isMessageRedundant(msg)) {
logger.debug("Sending redundant RREP")
//return
}
resendMissingRouteMessages()
if (msg.header.target == crypto.localAddress)
return
val existingRoute = localRoutesInfo.getRoute(msg.header.target)
val states = Set(LocalRoutesInfo.RouteStates.Active, LocalRoutesInfo.RouteStates.Idle)
if (existingRoute.isEmpty || !states.contains(existingRoute.get.state)) {
routeError(msg.header.target, Option(msg.header.origin))
return
}
val body = rrep.copy(originMetric = rrep.originMetric + 1)
val forwardMsg = crypto.sign(new Message(msg.header, body))
router.forwardMessage(forwardMsg)
return
case rerr: RouteError =>
localRoutesInfo.getRoute(rerr.address).foreach { route =>
if (route.nextHop == msg.header.origin && (rerr.seqNum == 0 || rerr.seqNum > route.seqNum)) {
localRoutesInfo.connectionClosed(rerr.address)
.foreach(routeError(_, None))
}
}
case _ =>
}
if (msg.header.target != crypto.localAddress) {
router.forwardMessage(msg)
return
}
val plainMsg =
try {
if (!crypto.verify(msg)) {
logger.warn(s"Received message with invalid signature from ${msg.header.origin}")
return
}
if (msg.header.isContentMessage)
crypto.decrypt(msg)
else
msg
} catch {
case e: InvalidKeyException =>
logger.warn(s"Failed to verify or decrypt message $msg", e)
return
}
onNewMessage(plainMsg)
}
/**
* Tries to send messages in [[missingRouteMessages]] again, after we acquired a new route.
*
* Before checking [[missingRouteMessages]], those older than [[MissingRouteMessageTimeout]]
* are removed.
*/
private def resendMissingRouteMessages(): Unit = {
// resend messages if possible
val date = new Date()
missingRouteMessages = missingRouteMessages.filter { e =>
val removeTime = new Date(e._2.getTime + MissingRouteMessageTimeout.toMillis)
removeTime.after(date)
}
val m = missingRouteMessages.filter(m => localRoutesInfo.getRoute(m._1.header.target).isDefined)
m.foreach( m => router.forwardMessage(m._1))
missingRouteMessages --= m
}
private def noRouteFound(message: Message): Unit = {
if (message.header.origin == crypto.localAddress) {
missingRouteMessages += ((message, new Date()))
requestRoute(message.header.target)
} else
routeError(message.header.target, Option(message.header.origin))
}
/**
@ -127,34 +292,34 @@ final class ConnectionHandler(settings: SettingsInterface, database: Database,
val maxConnections = settings.get(SettingsInterface.KeyMaxConnections,
SettingsInterface.DefaultMaxConnections.toString).toInt
if (connections().size == maxConnections) {
Log.i(Tag, "Maximum number of connections reached")
logger.info("Maximum number of connections reached")
return false
}
val info = msg.body.asInstanceOf[ConnectionInfo]
val sender = crypto.calculateAddress(info.key)
if (sender == Address.Broadcast || sender == Address.Null) {
Log.i(Tag, "Ignoring ConnectionInfo message with invalid sender " + sender)
logger.info("Ignoring ConnectionInfo message with invalid sender " + sender)
return false
}
if (crypto.havePublicKey(sender) && !crypto.verify(msg, Option(crypto.getPublicKey(sender)))) {
Log.i(Tag, "Ignoring ConnectionInfo message with invalid signature")
logger.info("Ignoring ConnectionInfo message with invalid signature")
return false
}
synchronized {
if (!crypto.havePublicKey(sender)) {
crypto.addPublicKey(sender, info.key)
Log.i(Tag, "Added public key for new device " + sender.toString)
logger.info("Added public key for new device " + sender.toString)
}
}
// Log with username if we know it.
if (allKnownUsers().map(_.address).contains(sender))
Log.i(Tag, "Node " + getUser(sender).name + " (" + sender + ") connected")
logger.info("Node " + getUser(sender).name + " (" + sender + ") connected")
else
Log.i(Tag, "Node " + sender + " connected")
logger.info("Node " + sender + " connected")
sendTo(sender, new UserInfo(settings.get(SettingsInterface.KeyUserName, ""),
settings.get(SettingsInterface.KeyUserStatus, "")))
@ -162,7 +327,11 @@ final class ConnectionHandler(settings: SettingsInterface, database: Database,
true
}
def onConnectionClosed() = callbacks.onConnectionsChanged()
def onConnectionClosed(address: Address): Unit = {
localRoutesInfo.connectionClosed(address)
.foreach(routeError(_, None))
callbacks.onConnectionsChanged()
}
def connections(): Set[Address] = transmissionInterfaces.flatMap(_.getConnections)
@ -176,6 +345,9 @@ final class ConnectionHandler(settings: SettingsInterface, database: Database,
.find(_.address == address)
.getOrElse(new User(address, address.toString(), ""))
/**
* This method should be called when the local device's internet connection has changed in any way.
*/
def internetConnectionChanged(): Unit = {
transmissionInterfaces
.find(_.isInstanceOf[InternetInterface])

31
core/src/main/scala/com/nutomic/ensichat/core/Crypto.scala
View file

@ -9,7 +9,8 @@ import javax.crypto.{Cipher, CipherOutputStream, KeyGenerator, SecretKey}
import com.nutomic.ensichat.core.Crypto._
import com.nutomic.ensichat.core.body._
import com.nutomic.ensichat.core.header.ContentHeader
import com.nutomic.ensichat.core.interfaces.{Log, SettingsInterface}
import com.nutomic.ensichat.core.interfaces.SettingsInterface
import com.typesafe.scalalogging.Logger
object Crypto {
@ -76,7 +77,7 @@ object Crypto {
*/
class Crypto(settings: SettingsInterface, keyFolder: File) {
private val Tag = "Crypto"
private val logger = Logger(this.getClass)
/**
* Generates a new key pair using [[keyFolder]] with [[PublicKeySize]] bits and stores the
@ -104,7 +105,7 @@ class Crypto(settings: SettingsInterface, keyFolder: File) {
saveKey(PrivateKeyAlias, keyPair.getPrivate)
saveKey(PublicKeyAlias, keyPair.getPublic)
Log.i(Tag, "Generated cryptographic keys, address is " + address)
logger.info("Generated cryptographic keys, address is " + address)
}
/**
@ -118,7 +119,7 @@ class Crypto(settings: SettingsInterface, keyFolder: File) {
* @throws RuntimeException If the key does not exist.
*/
@throws[RuntimeException]
private[core] def getPublicKey(address: Address): PublicKey = {
def getPublicKey(address: Address): PublicKey = {
loadKey(address.toString, classOf[PublicKey])
}
@ -128,7 +129,7 @@ class Crypto(settings: SettingsInterface, keyFolder: File) {
* @throws RuntimeException If a key already exists for this address.
*/
@throws[RuntimeException]
private[core] def addPublicKey(address: Address, key: PublicKey): Unit = {
def addPublicKey(address: Address, key: PublicKey): Unit = {
if (havePublicKey(address))
throw new RuntimeException("Already have key for " + address + ", not overwriting")
@ -184,7 +185,7 @@ class Crypto(settings: SettingsInterface, keyFolder: File) {
fos = Option(new FileOutputStream(path))
fos.foreach(_.write(key.getEncoded))
} catch {
case e: IOException => Log.w(Tag, "Failed to save key for alias " + alias, e)
case e: IOException => logger.warn("Failed to save key for alias " + alias, e)
} finally {
fos.foreach(_.close())
}
@ -212,7 +213,7 @@ class Crypto(settings: SettingsInterface, keyFolder: File) {
data = new Array[Byte](path.length().asInstanceOf[Int])
fis.foreach(_.read(data))
} catch {
case e: IOException => Log.e(Tag, "Failed to load key for alias " + alias, e)
case e: IOException => logger.error("Failed to load key for alias " + alias, e)
} finally {
fis.foreach(_.close())
}
@ -231,20 +232,6 @@ class Crypto(settings: SettingsInterface, keyFolder: File) {
sign(encrypt(msg, key))
}
private[core] def verifyAndDecrypt(msg: Message, key: Option[PublicKey] = None): Option[Message] = {
// Catch exception to avoid crash if we receive invalid message.
try {
if (verify(msg, key))
Option(decrypt(msg))
else
None
} catch {
case e: InvalidKeyException =>
Log.w(Tag, "Failed to verify or decrypt message", e)
None
}
}
private def encrypt(msg: Message, key: Option[PublicKey] = None): Message = {
// Symmetric encryption of data
val secretKey = makeSecretKey()
@ -262,7 +249,7 @@ class Crypto(settings: SettingsInterface, keyFolder: File) {
}
@throws[InvalidKeyException]
private def decrypt(msg: Message): Message = {
def decrypt(msg: Message): Message = {
// Asymmetric decryption of secret key
val asymmetricCipher = Cipher.getInstance(CipherAlgorithm)
asymmetricCipher.init(Cipher.UNWRAP_MODE, loadKey(PrivateKeyAlias, classOf[PrivateKey]))

12
core/src/main/scala/com/nutomic/ensichat/core/Message.scala
View file

@ -3,7 +3,7 @@ package com.nutomic.ensichat.core
import java.io.InputStream
import java.security.spec.InvalidKeySpecException
import com.nutomic.ensichat.core.body.{ConnectionInfo, CryptoData, EncryptedBody, MessageBody}
import com.nutomic.ensichat.core.body._
import com.nutomic.ensichat.core.header.{AbstractHeader, ContentHeader, MessageHeader}
object Message {
@ -50,6 +50,9 @@ object Message {
val body =
header.protocolType match {
case ConnectionInfo.Type => ConnectionInfo.read(remaining)
case RouteRequest.Type => RouteRequest.read(remaining)
case RouteReply.Type => RouteReply.read(remaining)
case RouteError.Type => RouteError.read(remaining)
case _ => new EncryptedBody(remaining)
}
@ -80,6 +83,11 @@ case class Message(header: AbstractHeader, crypto: CryptoData, body: MessageBody
def this(header: AbstractHeader, body: MessageBody) =
this(header, new CryptoData(None, None), body)
def write = header.write(body.length + crypto.length) ++ crypto.write ++ body.write
def write = {
header.write(body.length + crypto.length) ++ crypto.write ++ body.write
}
override def toString =
s"Message(${header.origin.short}(${header.seqNum}) -> ${header.target.short}: $body)"
}

62
core/src/main/scala/com/nutomic/ensichat/core/Router.scala
View file

@ -1,18 +1,44 @@
package com.nutomic.ensichat.core
import java.util.Comparator
import com.nutomic.ensichat.core.header.{ContentHeader, MessageHeader}
import com.nutomic.ensichat.core.util.LocalRoutesInfo
object Router extends Comparator[Int] {
/**
* Compares which sequence number is newer.
*
* @return 1 if lhs is newer, -1 if rhs is newer, 0 if they are equal.
*/
override def compare(lhs: Int, rhs: Int): Int = {
if (lhs == rhs)
0
// True if [[rhs]] is between {{{MessageHeader.SeqNumRange.size / 2}}} and
// [[MessageHeader.SeqNumRange.size]].
else if (lhs > ContentHeader.SeqNumRange.size / 2) {
// True if [[rhs]] is between {{{lhs - MessageHeader.SeqNumRange.size / 2}}} and [[lhs]].
if (lhs - ContentHeader.SeqNumRange.size / 2 < rhs && rhs < lhs) 1 else -1
} else {
// True if [[rhs]] is *not* between [[lhs]] and {{{lhs + MessageHeader.SeqNumRange.size / 2}}}.
if (rhs < lhs || rhs > lhs + ContentHeader.SeqNumRange.size / 2) 1 else -1
}
}
}
/**
* Forwards messages to all connected devices.
*/
final private[core] class Router(activeConnections: () => Set[Address], send: (Address, Message) => Unit) {
private[core] class Router(routesInfo: LocalRoutesInfo, send: (Address, Message) => Unit,
noRouteFound: (Message) => Unit) {
private var messageSeen = Set[(Address, Int)]()
/**
* Returns true if we have received the same message before.
*/
def isMessageSeen(msg: Message): Boolean = {
private[core] def isMessageSeen(msg: Message): Boolean = {
val info = (msg.header.origin, msg.header.seqNum)
val seen = messageSeen.contains(info)
markMessageSeen(info)
@ -23,15 +49,24 @@ final private[core] class Router(activeConnections: () => Set[Address], send: (A
* Sends message to all connected devices. Should only be called if [[isMessageSeen()]] returns
* true.
*/
def forwardMessage(msg: Message): Unit = {
val info = (msg.header.origin, msg.header.seqNum)
val updated = incHopCount(msg)
if (updated.header.hopCount >= updated.header.hopLimit)
def forwardMessage(msg: Message, nextHopOption: Option[Address] = None): Unit = {
if (msg.header.hopCount + 1 >= msg.header.hopLimit)
return
activeConnections().foreach(a => send(a, updated))
val nextHop = nextHopOption.getOrElse(msg.header.target)
markMessageSeen(info)
if (nextHop == Address.Broadcast) {
send(nextHop, msg)
return
}
routesInfo.getRoute(nextHop).map(_.nextHop) match {
case Some(a) =>
send(a, incHopCount(msg))
markMessageSeen((msg.header.origin, msg.header.seqNum))
case None =>
noRouteFound(msg)
}
}
private def markMessageSeen(info: (Address, Int)): Unit = {
@ -64,15 +99,8 @@ final private[core] class Router(activeConnections: () => Set[Address], send: (A
if (a1 != a2)
true
// True if [[s2]] is between {{{MessageHeader.SeqNumRange.size / 2}}} and
// [[MessageHeader.SeqNumRange.size]].
if (s1 > ContentHeader.SeqNumRange.size / 2) {
// True if [[s2]] is between {{{s1 - MessageHeader.SeqNumRange.size / 2}}} and [[s1]].
s1 - ContentHeader.SeqNumRange.size / 2 < s2 && s2 < s1
} else {
// True if [[s2]] is *not* between [[s1]] and {{{s1 + MessageHeader.SeqNumRange.size / 2}}}.
s2 < s1 || s2 > s1 + ContentHeader.SeqNumRange.size / 2
}
else
Router.compare(s1, s2) > 0
}
}

2
core/src/main/scala/com/nutomic/ensichat/core/body/EncryptedBody.scala
View file

@ -12,4 +12,6 @@ final case class EncryptedBody(data: Array[Byte]) extends MessageBody {
def write = data
override def length = data.length
override def toString = "EncryptedBody"
}

39
core/src/main/scala/com/nutomic/ensichat/core/body/RouteError.scala Normal file
View file

@ -0,0 +1,39 @@
package com.nutomic.ensichat.core.body
import java.nio.ByteBuffer
import com.nutomic.ensichat.core.Address
import com.nutomic.ensichat.core.util.BufferUtils
private[core] object RouteError {
val Type = 4
/**
* Constructs [[RouteError]] instance from byte array.
*/
def read(array: Array[Byte]): RouteError = {
val b = ByteBuffer.wrap(array)
val address = new Address(BufferUtils.getByteArray(b, Address.Length))
val seqNum = b.getInt
new RouteError(address, seqNum)
}
}
private[core] case class RouteError(address: Address, seqNum: Int) extends MessageBody {
override def protocolType = RouteReply.Type
override def contentType = -1
override def write: Array[Byte] = {
val b = ByteBuffer.allocate(length)
b.put(address.bytes)
b.putInt(seqNum)
b.array()
}
override def length = Address.Length + 4
}

50
core/src/main/scala/com/nutomic/ensichat/core/body/RouteReply.scala Normal file
View file

@ -0,0 +1,50 @@
package com.nutomic.ensichat.core.body
import java.nio.ByteBuffer
import com.nutomic.ensichat.core.util.BufferUtils
private[core] object RouteReply {
val Type = 3
/**
* Constructs [[RouteReply]] instance from byte array.
*/
def read(array: Array[Byte]): RouteReply = {
val b = ByteBuffer.wrap(array)
val targSeqNum = BufferUtils.getUnsignedShort(b)
val targMetric = BufferUtils.getUnsignedShort(b)
new RouteReply(targSeqNum, targMetric)
}
}
/**
* Sends information about a route.
*
* Note that the fields are named different than described in AODVv2. There, targSeqNum and
* targMetric are used to describe the seqNum and metric of the node sending the route reply. In
* Ensichat, we use originSeqNum and originMetric instead, to stay consistent with the header
* fields. That means header.origin, originSeqNum and originMetric all refer to the node sending
* this message.
*
* @param originSeqNum The current sequence number of the node sending this message.
* @param originMetric The metric of the current route to the sending node.
*/
private[core] case class RouteReply(originSeqNum: Int, originMetric: Int) extends MessageBody {
override def protocolType = RouteReply.Type
override def contentType = -1
override def write: Array[Byte] = {
val b = ByteBuffer.allocate(length)
BufferUtils.putUnsignedShort(b, originSeqNum)
BufferUtils.putUnsignedShort(b, originMetric)
b.array()
}
override def length = 4
}

44
core/src/main/scala/com/nutomic/ensichat/core/body/RouteRequest.scala Normal file
View file

@ -0,0 +1,44 @@
package com.nutomic.ensichat.core.body
import java.nio.ByteBuffer
import com.nutomic.ensichat.core.Address
import com.nutomic.ensichat.core.util.BufferUtils
private[core] object RouteRequest {
val Type = 2
/**
* Constructs [[RouteRequest]] instance from byte array.
*/
def read(array: Array[Byte]): RouteRequest = {
val b = ByteBuffer.wrap(array)
val requested = new Address(BufferUtils.getByteArray(b, Address.Length))
val origSeqNum = BufferUtils.getUnsignedShort(b)
val originMetric = BufferUtils.getUnsignedShort(b)
val targSeqNum = b.getInt()
new RouteRequest(requested, origSeqNum, targSeqNum, originMetric)
}
}
private[core] case class RouteRequest(requested: Address, originSeqNum: Int, targSeqNum: Int, originMetric: Int)
extends MessageBody {
override def protocolType = RouteRequest.Type
override def contentType = -1
override def write: Array[Byte] = {
val b = ByteBuffer.allocate(length)
b.put(requested.bytes)
BufferUtils.putUnsignedShort(b, originSeqNum)
BufferUtils.putUnsignedShort(b, originMetric)
b.putInt(targSeqNum)
b.array()
}
override def length = 8 + Address.Length
}

25
core/src/main/scala/com/nutomic/ensichat/core/interfaces/Log.scala
View file

@ -1,25 +0,0 @@
package com.nutomic.ensichat.core.interfaces
object Log {
def setLogInstance(log: Log) = instance = Option(log)
private var instance: Option[Log] = None
def v(tag: String, message: String, tr: Throwable = null) = instance.foreach(_.v(tag, message, tr))
def d(tag: String, message: String, tr: Throwable = null) = instance.foreach(_.d(tag, message, tr))
def i(tag: String, message: String, tr: Throwable = null) = instance.foreach(_.i(tag, message, tr))
def w(tag: String, message: String, tr: Throwable = null) = instance.foreach(_.w(tag, message, tr))
def e(tag: String, message: String, tr: Throwable = null) = instance.foreach(_.e(tag, message, tr))
}
trait Log {
def v(tag: String, message: String, tr: Throwable = null)
def d(tag: String, message: String, tr: Throwable = null)
def i(tag: String, message: String, tr: Throwable = null)
def w(tag: String, message: String, tr: Throwable = null)
def e(tag: String, message: String, tr: Throwable = null)
}

25
core/src/main/scala/com/nutomic/ensichat/core/internet/InternetConnectionThread.scala
View file

@ -6,23 +6,25 @@ import java.net.{InetAddress, Socket}
import com.nutomic.ensichat.core.Message.ReadMessageException
import com.nutomic.ensichat.core.body.ConnectionInfo
import com.nutomic.ensichat.core.header.MessageHeader
import com.nutomic.ensichat.core.interfaces.Log
import com.nutomic.ensichat.core.{Address, Crypto, Message}
import com.typesafe.scalalogging.Logger
/**
* Encapsulates an active connection to another node.
*/
class InternetConnectionThread(socket: Socket, crypto: Crypto, onDisconnected: (InternetConnectionThread) => Unit,
onReceive: (Message, InternetConnectionThread) => Unit) extends Thread {
private[core] class InternetConnectionThread(socket: Socket, crypto: Crypto,
onDisconnected: (InternetConnectionThread) => Unit,
onReceive: (Message, InternetConnectionThread) => Unit)
extends Thread {
private val Tag = "InternetConnectionThread"
private val logger = Logger(this.getClass)
private val inStream: InputStream =
try {
socket.getInputStream
} catch {
case e: IOException =>
Log.e(Tag, "Failed to open stream", e)
logger.error("Failed to open stream", e)
close()
null
}
@ -32,7 +34,7 @@ class InternetConnectionThread(socket: Socket, crypto: Crypto, onDisconnected: (
socket.getOutputStream
} catch {
case e: IOException =>
Log.e(Tag, "Failed to open stream", e)
logger.error("Failed to open stream", e)
close()
null
}
@ -42,7 +44,7 @@ class InternetConnectionThread(socket: Socket, crypto: Crypto, onDisconnected: (
}
override def run(): Unit = {
Log.i(Tag, "Connection opened to " + socket.getInetAddress)
logger.info("Connection opened to " + socket.getInetAddress)
send(crypto.sign(new Message(new MessageHeader(ConnectionInfo.Type,
Address.Null, Address.Null, 0), new ConnectionInfo(crypto.getLocalPublicKey))))
@ -51,13 +53,13 @@ class InternetConnectionThread(socket: Socket, crypto: Crypto, onDisconnected: (
socket.setKeepAlive(true)
while (socket.isConnected) {
val msg = Message.read(inStream)
Log.v(Tag, "Received " + msg)
logger.trace("Received " + msg)
onReceive(msg, this)
}
} catch {
case e @ (_: ReadMessageException | _: IOException) =>
Log.w(Tag, "Failed to read incoming message", e)
logger.warn("Failed to read incoming message", e)
close()
return
}
@ -68,7 +70,7 @@ class InternetConnectionThread(socket: Socket, crypto: Crypto, onDisconnected: (
try {
outStream.write(msg.write)
} catch {
case e: IOException => Log.e(Tag, "Failed to write message", e)
case e: IOException => logger.error("Failed to write message", e)
}
}
@ -76,9 +78,8 @@ class InternetConnectionThread(socket: Socket, crypto: Crypto, onDisconnected: (
try {
socket.close()
} catch {
case e: IOException => Log.w(Tag, "Failed to close socket", e)
case e: IOException => logger.warn("Failed to close socket", e)
}
Log.d(Tag, "Connection to " + socket.getInetAddress + " closed")
onDisconnected(this)
}

55
core/src/main/scala/com/nutomic/ensichat/core/internet/InternetInterface.scala
View file

@ -1,19 +1,20 @@
package com.nutomic.ensichat.core.internet
import java.io.IOException
import java.net.{InetAddress, Socket}
import com.nutomic.ensichat.core.body.ConnectionInfo
import com.nutomic.ensichat.core.interfaces.{Log, SettingsInterface, TransmissionInterface}
import com.nutomic.ensichat.core.interfaces.{SettingsInterface, TransmissionInterface}
import com.nutomic.ensichat.core.util.FutureHelper
import com.nutomic.ensichat.core.{Address, ConnectionHandler, Crypto, Message}
import com.typesafe.scalalogging.Logger
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future
import scala.util.Random
object InternetInterface {
private[core] object InternetInterface {
val ServerPort = 26344
val DefaultPort = 26344
}
@ -22,14 +23,14 @@ object InternetInterface {
*
* @param maxConnections Maximum number of concurrent connections that should be opened.
*/
class InternetInterface(connectionHandler: ConnectionHandler, crypto: Crypto,
settings: SettingsInterface, maxConnections: Int)
private[core] class InternetInterface(connectionHandler: ConnectionHandler, crypto: Crypto,
settings: SettingsInterface, maxConnections: Int, port: Int)
extends TransmissionInterface {
private val Tag = "InternetInterface"
private val logger = Logger(this.getClass)
private lazy val serverThread =
new InternetServerThread(crypto, onConnected, onDisconnected, onReceiveMessage)
new InternetServerThread(crypto, port, onConnected, onDisconnected, onReceiveMessage)
private var connections = Set[InternetConnectionThread]()
@ -43,10 +44,8 @@ class InternetInterface(connectionHandler: ConnectionHandler, crypto: Crypto,
.replace("46.101.249.188:26344", SettingsInterface.DefaultAddresses)
settings.put(SettingsInterface.KeyAddresses, servers)
FutureHelper {
serverThread.start()
openAllConnections(maxConnections)
}
serverThread.start()
openAllConnections(maxConnections)
}
/**
@ -68,13 +67,13 @@ class InternetInterface(connectionHandler: ConnectionHandler, crypto: Crypto,
.foreach(openConnection)
}
private def openConnection(addressPort: String): Unit = {
def openConnection(addressPort: String): Unit = {
val (address, port) =
if (addressPort.contains(":")) {
val split = addressPort.split(":")
(split(0), split(1).toInt)
} else
(addressPort, InternetInterface.ServerPort)
(addressPort, InternetInterface.DefaultPort)
openConnection(address, port)
}
@ -83,15 +82,14 @@ class InternetInterface(connectionHandler: ConnectionHandler, crypto: Crypto,
* Opens connection to the specified IP address in client mode.
*/
private def openConnection(address: String, port: Int): Unit = {
Log.i(Tag, s"Attempting connection to $address:$port")
try {
logger.info(s"Attempting connection to $address:$port")
Future {
val socket = new Socket(InetAddress.getByName(address), port)
val ct = new InternetConnectionThread(socket, crypto, onDisconnected, onReceiveMessage)
connections += ct
ct.start()
} catch {
case e: IOException =>
Log.w(Tag, "Failed to open connection to " + address + ":" + port, e)
}.onFailure { case e =>
logger.warn("Failed to open connection to " + address + ":" + port, e)
}
}
@ -100,11 +98,11 @@ class InternetInterface(connectionHandler: ConnectionHandler, crypto: Crypto,
}
private def onDisconnected(connectionThread: InternetConnectionThread): Unit = {
addressDeviceMap.find(_._2 == connectionThread).foreach { ad =>
Log.d(Tag, "Connection closed to " + ad._1)
getAddressForThread(connectionThread).foreach { ad =>
logger.trace("Connection closed to " + ad)
connections -= connectionThread
addressDeviceMap -= ad._1
connectionHandler.onConnectionClosed()
addressDeviceMap -= ad
connectionHandler.onConnectionClosed(ad)
}
}
@ -112,7 +110,7 @@ class InternetInterface(connectionHandler: ConnectionHandler, crypto: Crypto,
case info: ConnectionInfo =>
val address = crypto.calculateAddress(info.key)
if (address == crypto.localAddress) {
Log.i(Tag, "Address " + address + " is me, not connecting to myself")
logger.info("Address " + address + " is me, not connecting to myself")
thread.close()
return
}
@ -122,15 +120,18 @@ class InternetInterface(connectionHandler: ConnectionHandler, crypto: Crypto,
if (!connectionHandler.onConnectionOpened(msg))
addressDeviceMap -= address
case _ =>
connectionHandler.onMessageReceived(msg)
connectionHandler.onMessageReceived(msg, getAddressForThread(thread).get)
}
private def getAddressForThread(thread: InternetConnectionThread) =
addressDeviceMap.find(_._2 == thread).map(_._1)
/**
* Sends the message to nextHop.
*/
override def send(nextHop: Address, msg: Message): Unit = {
addressDeviceMap
.find(_._1 == nextHop)
.filter(_._1 == nextHop || Address.Broadcast == nextHop)
.foreach(_._2.send(msg))
}
@ -141,7 +142,7 @@ class InternetInterface(connectionHandler: ConnectionHandler, crypto: Crypto,
def connectionChanged(): Unit = {
FutureHelper {
Log.i(Tag, "Network has changed. Closing all connections and connecting to bootstrap nodes again")
logger.info("Network has changed. Closing all connections and connecting to bootstrap nodes again")
connections.foreach(_.close())
openAllConnections(maxConnections)
}

18
core/src/main/scala/com/nutomic/ensichat/core/internet/InternetServerThread.scala
View file

@ -3,19 +3,21 @@ package com.nutomic.ensichat.core.internet
import java.io.IOException
import java.net.ServerSocket
import com.nutomic.ensichat.core.interfaces.Log
import com.nutomic.ensichat.core.{Crypto, Message}
import com.typesafe.scalalogging.Logger
class InternetServerThread(crypto: Crypto, onConnected: (InternetConnectionThread) => Unit,
onDisconnected: (InternetConnectionThread) => Unit, onReceive: (Message, InternetConnectionThread) => Unit) extends Thread {
class InternetServerThread(crypto: Crypto, port: Int,
onConnected: (InternetConnectionThread) => Unit,
onDisconnected: (InternetConnectionThread) => Unit,
onReceive: (Message, InternetConnectionThread) => Unit) extends Thread {
private val Tag = "InternetServerThread"
private val logger = Logger(this.getClass)
private lazy val socket: Option[ServerSocket] = try {
Option(new ServerSocket(InternetInterface.ServerPort))
Option(new ServerSocket(port))
} catch {
case e: IOException =>
Log.w(Tag, "Failed to create server socket", e)
logger.warn("Failed to create server socket", e)
None
}
@ -27,7 +29,7 @@ class InternetServerThread(crypto: Crypto, onConnected: (InternetConnectionThrea
connection.start()
}
} catch {
case e: IOException => Log.w(Tag, "Failed to accept connection", e)
case e: IOException => logger.warn("Failed to accept connection", e)
}
}
@ -35,7 +37,7 @@ class InternetServerThread(crypto: Crypto, onConnected: (InternetConnectionThrea
try {
socket.get.close()
} catch {
case e: IOException => Log.w(Tag, "Failed to close socket", e)
case e: IOException => logger.warn("Failed to close socket", e)
}
}

13
core/src/main/scala/com/nutomic/ensichat/core/util/Database.scala
View file

@ -5,10 +5,10 @@ import java.util.Date
import com.nutomic.ensichat.core.body.Text
import com.nutomic.ensichat.core.header.ContentHeader
import com.nutomic.ensichat.core.interfaces.{Log, CallbackInterface}
import com.nutomic.ensichat.core.interfaces.CallbackInterface
import com.nutomic.ensichat.core.{Address, Message, User}
import com.typesafe.scalalogging.Logger
import slick.driver.H2Driver.api._
import slick.jdbc.meta.MTable
import scala.concurrent.Await
import scala.concurrent.duration.Duration
@ -21,7 +21,7 @@ import scala.concurrent.duration.Duration
*/
class Database(path: File, callbackInterface: CallbackInterface) {
private val Tag = "Database"
private val logger = Logger(this.getClass)
private class Messages(tag: Tag) extends Table[Message](tag, "MESSAGES") {
def id = primaryKey("id", (origin, messageId))
@ -63,7 +63,7 @@ class Database(path: File, callbackInterface: CallbackInterface) {
// H2 appends a .mv.db suffix to the path which we can't change, so we have to check that file.
val dbFile = new File(path.getAbsolutePath + ".mv.db")
if (!dbFile.exists()) {
Log.i(Tag, "Database does not exist, creating tables")
logger.info("Database does not exist, creating tables")
Await.result(db.run((messages.schema ++ contacts.schema).create), Duration.Inf)
}
}
@ -75,8 +75,9 @@ class Database(path: File, callbackInterface: CallbackInterface) {
/**
* Inserts the given new message into the database.
*/
def onMessageReceived(msg: Message): Unit = {
Await.result(db.run(messages += msg), Duration.Inf)
def onMessageReceived(msg: Message): Unit = msg.body match {
case _: Text => Await.result(db.run(messages += msg), Duration.Inf)
case _ =>
}
def getMessages(address: Address): Seq[Message] = {

6
core/src/main/scala/com/nutomic/ensichat/core/util/FutureHelper.scala
View file

@ -1,6 +1,6 @@
package com.nutomic.ensichat.core.util
import com.nutomic.ensichat.core.interfaces.Log
import com.typesafe.scalalogging.Logger
import scala.concurrent.{ExecutionContext, Future}
@ -11,7 +11,7 @@ import scala.concurrent.{ExecutionContext, Future}
*/
object FutureHelper {
private val Tag = "FutureHelper"
private val logger = Logger(this.getClass)
def apply[A](action: => A)(implicit executor: ExecutionContext): Future[A] = {
val f = Future(action)
@ -20,7 +20,7 @@ object FutureHelper {
// HACK: Android does not close app when crash occurs in background thread, and there's no
// cross-platform way to execute on the foreground thread.
// We use this to make sure exceptions are not hidden in the logs.
Log.e(Tag, "Exception in Future", e)
logger.error("Exception in Future", e)
//System.exit(-1)
}
f

119
core/src/main/scala/com/nutomic/ensichat/core/util/LocalRoutesInfo.scala Normal file
View file

@ -0,0 +1,119 @@
package com.nutomic.ensichat.core.util
import com.nutomic.ensichat.core.Address
import com.nutomic.ensichat.core.util.LocalRoutesInfo._
import com.typesafe.scalalogging.Logger
import org.joda.time.{DateTime, Duration}
private[core] object LocalRoutesInfo {
private val ActiveInterval = Duration.standardSeconds(5)
/**
* [[RouteStates.Idle]]:
* A route that is known, but has not been used in the last [[ActiveInterval.
* [[RouteStates.Active]]:
* A route that is known, and has been used in the last [[ActiveInterval]].
* [[RouteStates.Invalid]]:
* A route that has been expired or lost, may not be used for forwarding.
* RouteStates.Unconfirmed is not required as connections are always bidirectional.
*/
object RouteStates extends Enumeration {
type RouteStates = Value
val Idle, Active, Invalid = Value
}
}
/**
* This class contains information about routes available to this node.
*
* See AODVv2-13 4.5 (Local Route Set), -> implemented
* 6.9 (Local Route Set Maintenance) -> implemented (hopefully correct)
*/
private[core] class LocalRoutesInfo(activeConnections: () => Set[Address]) {
import RouteStates._
private val MaxSeqnumLifetime = Duration.standardSeconds(300)
// TODO: this can probably be much higher because of infrequent topology changes between internet nodes
private val MaxIdleTime = Duration.standardSeconds(300)
/**
* Holds information about a local route.
*
* @param destination The destination address that can be reached with this route.
* @param seqNum Sequence number of the last route message that updated this entry.
* @param nextHop The next hop on the path towards destination.
* @param lastUsed The time this route was last used to forward a message.
* @param lastSeqNumUpdate The time seqNum was last updated.
* @param metric The number of hops towards destination using this route.
* @param state The last known state of the route.
*/
case class RouteEntry(destination: Address, seqNum: Int, nextHop: Address, lastUsed: DateTime,
lastSeqNumUpdate: DateTime, metric: Int, state: RouteStates)
private var routes = Set[RouteEntry]()
def addRoute(destination: Address, seqNum: Int, nextHop: Address, metric: Int): Unit = {
val entry = RouteEntry(destination, seqNum, nextHop, new DateTime(0), DateTime.now, metric, Idle)
routes += entry
}
def getRoute(destination: Address): Option[RouteEntry] = {
if (activeConnections().contains(destination))
return Option(new RouteEntry(destination, 0, destination, DateTime.now, DateTime.now, 1, Idle))
handleTimeouts()
val r = routes.toList
.sortWith(_.metric < _.metric)
.find( r => r.destination == destination && r.state != Invalid)
if (r.isDefined)
routes = routes -- r + r.get.copy(state = Active, lastUsed = DateTime.now)
r
}
/**
*
* @param address The address which can't be reached any more.
* @return The set of active destinations that can't be reached anymore.
*/
def connectionClosed(address: Address): Set[Address] = {
handleTimeouts()
val affectedDestinations =
routes
.filter(r => r.state == Active && (r.nextHop == address || r.destination == address))
.map(_.destination)
routes = routes.map { r =>
if (r.nextHop == address || r.destination == address)
r.copy(state = Invalid)
else
r
}
affectedDestinations
}
private def handleTimeouts(): Unit = {
routes = routes
// Delete routes after max lifetime.
.map { r =>
if (DateTime.now.isAfter(r.lastSeqNumUpdate.plus(MaxSeqnumLifetime)))
r.copy(seqNum = 0)
else
r
}
// Set routes to invalid after max idle time.
.map { r =>
if (DateTime.now.isAfter(r.lastSeqNumUpdate.plus(MaxIdleTime)))
r.copy(state = Invalid)
else
r
}
}
}

74
core/src/main/scala/com/nutomic/ensichat/core/util/RouteMessageInfo.scala Normal file
View file

@ -0,0 +1,74 @@
package com.nutomic.ensichat.core.util
import com.nutomic.ensichat.core.body.{RouteReply, RouteRequest}
import com.nutomic.ensichat.core.{Address, Message, Router}
import org.joda.time.{DateTime, Duration}
/**
* Contains information about AODVv2 control messages that have been received.
*
* This class handles Route Request and Route Reply messages (referred to as "route messages").
*
* See AODVv2-13 4.6 (Multicast Route Message Table), -> implemented
* 6.8 (Surpressing Redundant Messages Using the Multicast Route Message Table) -> implemented (hopefully correct)
*/
private[core] class RouteMessageInfo {
private val MaxSeqnumLifetime = Duration.standardSeconds(300)
/**
* @param messageType Either [[RouteRequest.Type]] or [[RouteReply.Type]].
* @param origAddress Source address of the route message triggering the route request.
* @param targAddress Destination address of the route message triggering the route request.
* @param origSeqNum Sequence number associated with the route to [[origAddress]], if route
* message is an RREQ.
* @param targSeqNum Sequence number associated with the route to [[targAddress]], if present in
* the route message.
* @param metric Metric value received in the route message.
* @param timestamp Last time this entry was updated.
*/
private case class RouteMessageEntry(messageType: Int, origAddress: Address,
targAddress: Address, origSeqNum: Int, targSeqNum: Int,
metric: Int, timestamp: DateTime)
private var entries = Set[RouteMessageEntry]()
private def addEntry(msg: Message): Unit = msg.body match {
case rreq: RouteRequest =>
entries += new RouteMessageEntry(RouteRequest.Type, msg.header.origin, msg.header.target,
msg.header.seqNum, rreq.targSeqNum, rreq.originMetric,
DateTime.now)
case rrep: RouteReply =>
entries += new RouteMessageEntry(RouteReply.Type, msg.header.origin, msg.header.target,
msg.header.seqNum, rrep.originSeqNum, rrep.originMetric,
DateTime.now)
}
def isMessageRedundant(msg: Message): Boolean = {
handleTimeouts()
val existingEntry =
entries.find { e =>
val haveEntry = e.messageType == msg.header.protocolType &&
e.origAddress == msg.header.origin && e.targAddress == msg.header.target
val (metric, seqNumComparison) = msg.body match {
case rreq: RouteRequest => (rreq.originMetric, Router.compare(rreq.originSeqNum, e.origSeqNum))
case rrep: RouteReply => (rrep.originMetric, Router.compare(rrep.originSeqNum, e.targSeqNum))
}
val isMetricBetter = e.metric < metric
haveEntry && (seqNumComparison > 0 || (seqNumComparison == 0 && isMetricBetter))
}
if (existingEntry.isDefined)
entries = entries - existingEntry.get
addEntry(msg)
existingEntry.isDefined
}
private def handleTimeouts(): Unit = {
entries = entries.filter { e =>
DateTime.now.isBefore(e.timestamp.plus(MaxSeqnumLifetime))
}
}
}

5
core/src/test/scala/com/nutomic/ensichat/core/CryptoTest.scala
View file

@ -41,8 +41,9 @@ class CryptoTest extends TestCase {
def testEncryptDecrypt(): Unit = {
MessageTest.messages.foreach{ m =>
val encrypted = crypto.encryptAndSign(m, Option(crypto.getLocalPublicKey))
val decrypted = crypto.verifyAndDecrypt(encrypted, Option(crypto.getLocalPublicKey))
assertEquals(m.body, decrypted.get.body)
assertTrue(crypto.verify(encrypted, Option(crypto.getLocalPublicKey)))
val decrypted = crypto.decrypt(encrypted)
assertEquals(m.body, decrypted.body)
assertEquals(m.header, encrypted.header)
}
}

7
core/src/test/scala/com/nutomic/ensichat/core/MessageTest.scala
View file

@ -58,9 +58,10 @@ class MessageTest extends TestCase {
val read = Message.read(new ByteArrayInputStream(bytes))
assertEquals(encrypted.crypto, read.crypto)
val decrypted = crypto.verifyAndDecrypt(read, Option(crypto.getLocalPublicKey))
assertEquals(m.header, decrypted.get.header)
assertEquals(m.body, decrypted.get.body)
assertTrue(crypto.verify(read, Option(crypto.getLocalPublicKey)))
val decrypted = crypto.decrypt(read)
assertEquals(m.header, decrypted.header)
assertEquals(m.body, decrypted.body)
}
}

59
core/src/test/scala/com/nutomic/ensichat/core/RouterTest.scala
View file

@ -1,34 +1,45 @@
package com.nutomic.ensichat.core
import java.util.concurrent.{CountDownLatch, TimeUnit}
import java.util.{Date, GregorianCalendar}
import com.nutomic.ensichat.core.body.{Text, UserInfo}
import com.nutomic.ensichat.core.header.ContentHeader
import com.nutomic.ensichat.core.util.LocalRoutesInfo
import junit.framework.TestCase
import org.junit.Assert._
class RouterTest extends TestCase {
private def neighbors() = Set[Address](AddressTest.a1, AddressTest.a2, AddressTest.a3)
private def neighbors() = Set[Address](AddressTest.a1, AddressTest.a2, AddressTest.a4)
private val msg = generateMessage(AddressTest.a1, AddressTest.a4, 1)
def testNoRouteFound(): Unit = {
val msg = generateMessage(AddressTest.a2, AddressTest.a3, 1)
val latch = new CountDownLatch(1)
val router = new Router(new LocalRoutesInfo(neighbors),
(_, _) => fail("Message shouldn't be forwarded"), m => {
assertEquals(msg, m)
latch.countDown()
})
router.forwardMessage(msg)
assertTrue(latch.await(1, TimeUnit.SECONDS))
}
/**
* Messages should be sent to all neighbors.
*/
def testFlooding(): Unit = {
def testNextHop(): Unit = {
val msg = generateMessage(AddressTest.a1, AddressTest.a4, 1)
var sentTo = Set[Address]()
val router: Router = new Router(neighbors,
val router = new Router(new LocalRoutesInfo(neighbors),
(a, m) => {
sentTo += a
})
}, _ => ())
router.forwardMessage(msg)
assertEquals(neighbors(), sentTo)
assertEquals(Set(AddressTest.a4), sentTo)
}
def testMessageSame(): Unit = {
val router: Router = new Router(neighbors,
val msg = generateMessage(AddressTest.a1, AddressTest.a4, 1)
val router = new Router(new LocalRoutesInfo(neighbors),
(a, m) => {
assertEquals(msg.header.origin, m.header.origin)
assertEquals(msg.header.target, m.header.target)
@ -38,7 +49,7 @@ class RouterTest extends TestCase {
assertEquals(msg.header.hopLimit, m.header.hopLimit)
assertEquals(msg.body, m.body)
assertEquals(msg.crypto, m.crypto)
})
}, _ => ())
router.forwardMessage(msg)
}
@ -47,26 +58,32 @@ class RouterTest extends TestCase {
*/
def testDifferentSenders(): Unit = {
var sentTo = Set[Address]()
val router: Router = new Router(neighbors, (a, m) => sentTo += a)
val router = new Router(new LocalRoutesInfo(neighbors), (a, m) => sentTo += a, _ => ())
router.forwardMessage(msg)
assertEquals(neighbors(), sentTo)
router.forwardMessage(generateMessage(AddressTest.a1, AddressTest.a4, 1))
assertEquals(Set(AddressTest.a4), sentTo)
sentTo = Set[Address]()
router.forwardMessage(generateMessage(AddressTest.a2, AddressTest.a4, 1))
assertEquals(neighbors(), sentTo)
assertEquals(Set(AddressTest.a4), sentTo)
}
def testSeqNumComparison(): Unit = {
Router.compare(1, ContentHeader.SeqNumRange.last)
Router.compare(ContentHeader.SeqNumRange.last / 2, ContentHeader.SeqNumRange.last)
Router.compare(ContentHeader.SeqNumRange.last / 2, 1)
}
def testDiscardOldIgnores(): Unit = {
def test(first: Int, second: Int) {
var sentTo = Set[Address]()
val router: Router = new Router(neighbors, (a, m) => sentTo += a)
router.forwardMessage(generateMessage(AddressTest.a1, AddressTest.a3, first))
router.forwardMessage(generateMessage(AddressTest.a1, AddressTest.a3, second))
val router = new Router(new LocalRoutesInfo(neighbors), (a, m) => sentTo += a, _ => ())
router.forwardMessage(generateMessage(AddressTest.a1, AddressTest.a4, first))
router.forwardMessage(generateMessage(AddressTest.a1, AddressTest.a4, second))
sentTo = Set[Address]()
router.forwardMessage(generateMessage(AddressTest.a1, AddressTest.a3, first))
assertEquals(neighbors(), sentTo)
router.forwardMessage(generateMessage(AddressTest.a1, AddressTest.a4, first))
assertEquals(Set(AddressTest.a4), sentTo)
}
test(1, ContentHeader.SeqNumRange.last)
@ -77,7 +94,7 @@ class RouterTest extends TestCase {
def testHopLimit(): Unit = Range(19, 22).foreach { i =>
val msg = new Message(
new ContentHeader(AddressTest.a1, AddressTest.a2, 1, 1, Some(1), Some(new Date()), i), new Text(""))
val router: Router = new Router(neighbors, (a, m) => fail())
val router = new Router(new LocalRoutesInfo(neighbors), (a, m) => fail(), _ => ())
router.forwardMessage(msg)
}

16
core/src/test/scala/com/nutomic/ensichat/core/body/RouteErrorTest.scala Normal file
View file

@ -0,0 +1,16 @@
package com.nutomic.ensichat.core.body
import com.nutomic.ensichat.core.AddressTest
import junit.framework.TestCase
import org.junit.Assert._
class RouteErrorTest extends TestCase {
def testWriteRead(): Unit = {
val rerr = new RouteError(AddressTest.a2, 62000)
val bytes = rerr.write
val parsed = RouteError.read(bytes)
assertEquals(rerr, parsed)
}
}

16
core/src/test/scala/com/nutomic/ensichat/core/body/RouteReplyTest.scala Normal file
View file

@ -0,0 +1,16 @@
package com.nutomic.ensichat.core.body
import com.nutomic.ensichat.core.AddressTest
import junit.framework.TestCase
import org.junit.Assert._
class RouteReplyTest extends TestCase {
def testWriteRead(): Unit = {
val rrep = new RouteReply(61000, 123)
val bytes = rrep.write
val parsed = RouteReply.read(bytes)
assertEquals(rrep, parsed)
}
}

16
core/src/test/scala/com/nutomic/ensichat/core/body/RouteRequestTest.scala Normal file
View file

@ -0,0 +1,16 @@
package com.nutomic.ensichat.core.body
import com.nutomic.ensichat.core.AddressTest
import junit.framework.TestCase
import org.junit.Assert._
class RouteRequestTest extends TestCase {
def testWriteRead(): Unit = {
val rreq = new RouteRequest(AddressTest.a2, 60000, 60001, 60002)
val bytes = rreq.write
val parsed = RouteRequest.read(bytes)
assertEquals(rreq, parsed)
}
}

45
core/src/test/scala/com/nutomic/ensichat/core/util/LocalRoutesInfoTest.scala Normal file
View file

@ -0,0 +1,45 @@
package com.nutomic.ensichat.core.util
import com.nutomic.ensichat.core.AddressTest
import junit.framework.TestCase
import org.joda.time.{DateTime, DateTimeUtils, Duration}
import org.junit.Assert._
class LocalRoutesInfoTest extends TestCase {
private def connections() = Set(AddressTest.a1, AddressTest.a2)
def testRoute(): Unit = {
val routesInfo = new LocalRoutesInfo(connections)
routesInfo.addRoute(AddressTest.a3, 0, AddressTest.a1, 1)
val route = routesInfo.getRoute(AddressTest.a3)
assertEquals(AddressTest.a1, route.get.nextHop)
}
def testBestMetric(): Unit = {
val routesInfo = new LocalRoutesInfo(connections)
routesInfo.addRoute(AddressTest.a3, 0, AddressTest.a1, 1)
routesInfo.addRoute(AddressTest.a3, 0, AddressTest.a2, 2)
val route = routesInfo.getRoute(AddressTest.a3)
assertEquals(AddressTest.a1, route.get.nextHop)
}
def testConnectionClosed(): Unit = {
val routesInfo = new LocalRoutesInfo(connections)
routesInfo.addRoute(AddressTest.a3, 0, AddressTest.a1, 1)
routesInfo.addRoute(AddressTest.a4, 0, AddressTest.a1, 1)
// Mark the route as active, because only active routes are returned.
routesInfo.getRoute(AddressTest.a3)
val unreachable = routesInfo.connectionClosed(AddressTest.a1)
assertEquals(Set(AddressTest.a3), unreachable)
}
def testTimeout(): Unit = {
DateTimeUtils.setCurrentMillisFixed(new DateTime().getMillis)
val routesInfo = new LocalRoutesInfo(connections)
routesInfo.addRoute(AddressTest.a3, 0, AddressTest.a1, 1)
DateTimeUtils.setCurrentMillisFixed(DateTime.now.plus(Duration.standardSeconds(400)).getMillis)
assertEquals(None, routesInfo.getRoute(AddressTest.a3))
}
}

79
core/src/test/scala/com/nutomic/ensichat/core/util/RouteMessageInfoTest.scala Normal file
View file

@ -0,0 +1,79 @@
package com.nutomic.ensichat.core.util
import com.nutomic.ensichat.core.body.{RouteReply, RouteRequest}
import com.nutomic.ensichat.core.header.MessageHeader
import com.nutomic.ensichat.core.{AddressTest, Message}
import junit.framework.TestCase
import org.joda.time.{DateTime, DateTimeUtils, Duration}
import org.junit.Assert._
class RouteMessageInfoTest extends TestCase {
/**
* Test case in which we have an entry with the same type, origin and target.
*/
def testSameMessage(): Unit = {
val header = new MessageHeader(RouteRequest.Type, AddressTest.a1, AddressTest.a2, 1)
val msg = new Message(header, new RouteRequest(AddressTest.a3, 2, 3, 1))
val rmi = new RouteMessageInfo()
assertFalse(rmi.isMessageRedundant(msg))
assertTrue(rmi.isMessageRedundant(msg))
}
/**
* Forward a message with a seqnum that is older than the latest.
*/
def testSeqNumOlder(): Unit = {
val header1 = new MessageHeader(RouteRequest.Type, AddressTest.a1, AddressTest.a2, 1)
val msg1 = new Message(header1, new RouteRequest(AddressTest.a3, 0, 0, 0))
val rmi = new RouteMessageInfo()
assertFalse(rmi.isMessageRedundant(msg1))
val header2 = new MessageHeader(RouteRequest.Type, AddressTest.a1, AddressTest.a2, 3)
val msg2 = new Message(header2, new RouteRequest(AddressTest.a3, 2, 0, 0))
assertTrue(rmi.isMessageRedundant(msg2))
}
/**
* Announce a route with a metric that is worse than the existing one.
*/
def testMetricWorse(): Unit = {
val header1 = new MessageHeader(RouteRequest.Type, AddressTest.a1, AddressTest.a2, 1)
val msg1 = new Message(header1, new RouteRequest(AddressTest.a3, 1, 0, 2))
val rmi = new RouteMessageInfo()
assertFalse(rmi.isMessageRedundant(msg1))
val header2 = new MessageHeader(RouteRequest.Type, AddressTest.a1, AddressTest.a2, 2)
val msg2 = new Message(header2, new RouteRequest(AddressTest.a3, 1, 0, 4))
assertTrue(rmi.isMessageRedundant(msg2))
}
/**
* Announce route with a better metric.
*/
def testMetricBetter(): Unit = {
val header1 = new MessageHeader(RouteRequest.Type, AddressTest.a1, AddressTest.a2, 1)
val msg1 = new Message(header1, new RouteReply(0, 4))
val rmi = new RouteMessageInfo()
assertFalse(rmi.isMessageRedundant(msg1))
val header2 = new MessageHeader(RouteRequest.Type, AddressTest.a1, AddressTest.a2, 2)
val msg2 = new Message(header2, new RouteReply(0, 2))
assertFalse(rmi.isMessageRedundant(msg2))
}
/**
* Test that entries are removed after [[RouteMessageInfo.MaxSeqnumLifetime]].
*/
def testTimeout(): Unit = {
val rmi = new RouteMessageInfo()
DateTimeUtils.setCurrentMillisFixed(DateTime.now.getMillis)
val header = new MessageHeader(RouteRequest.Type, AddressTest.a1, AddressTest.a2, 1)
val msg = new Message(header, new RouteRequest(AddressTest.a3, 0, 0, 0))
assertFalse(rmi.isMessageRedundant(msg))
DateTimeUtils.setCurrentMillisFixed(DateTime.now.plus(Duration.standardSeconds(400)).getMillis)
assertFalse(rmi.isMessageRedundant(msg))
}
}

1
integration/.gitignore vendored Normal file
View file

@ -0,0 +1 @@
/build

12
integration/build.gradle Normal file
View file

@ -0,0 +1,12 @@
apply plugin: 'scala'
apply plugin: 'application'
dependencies {
compile 'org.scala-lang:scala-library:2.11.7'
compile 'com.github.scala-incubator.io:scala-io-file_2.11:0.4.3'
compile project(path: ':core')
}
mainClassName = 'com.nutomic.ensichat.integration.Main'
version = "0.2.3"
applicationName = 'ensichat-server'

84
integration/src/main/scala/com.nutomic.ensichat.integration/LocalNode.scala Normal file
View file

@ -0,0 +1,84 @@
package com.nutomic.ensichat.integration
import java.io.File
import java.util.concurrent.{LinkedBlockingDeque, LinkedBlockingQueue}
import com.nutomic.ensichat.core.body.{RouteError, RouteRequest, RouteReply}
import com.nutomic.ensichat.core.interfaces.{CallbackInterface, SettingsInterface}
import com.nutomic.ensichat.core.util.Database
import com.nutomic.ensichat.core.{ConnectionHandler, Crypto, Message}
import com.nutomic.ensichat.integration.LocalNode._
import scala.concurrent.Await
import scala.concurrent.duration.Duration
import scalax.file.Path
object LocalNode {
private final val StartingPort = 21000
object EventType extends Enumeration {
type EventType = Value
val MessageReceived, ConnectionsChanged, ContactsUpdated = Value
}
class FifoStream[A]() {
private val queue = new LinkedBlockingQueue[Option[A]]()
def toStream: Stream[A] = queue.take match {
case Some(a) => Stream.cons(a, toStream)
case None => Stream.empty
}
def close() = queue add None
def enqueue(a: A) = queue.put(Option(a))
}
}
/**
* Runs an ensichat node on localhost.
*
* Received messages can be accessed through [[eventQueue]].
*
* @param index Number of this node. The server port is opened on port [[StartingPort]] + index.
* @param configFolder Folder where keys and configuration should be stored.
*/
class LocalNode(val index: Int, configFolder: File) extends CallbackInterface {
import com.nutomic.ensichat.integration.LocalNode.EventType._
private val databaseFile = new File(configFolder, "database")
private val keyFolder = new File(configFolder, "keys")
private val database = new Database(databaseFile, this)
private val settings = new SettingsInterface {
private var values = Map[String, Any]()
override def get[T](key: String, default: T): T = values.get(key).map(_.asInstanceOf[T]).getOrElse(default)
override def put[T](key: String, value: T): Unit = values += (key -> value.asInstanceOf[Any])
}
val crypto = new Crypto(settings, keyFolder)
val connectionHandler = new ConnectionHandler(settings, database, this, crypto, 0, port)
val eventQueue = new FifoStream[(EventType.EventType, Option[Message])]()
configFolder.mkdirs()
keyFolder.mkdirs()
settings.put(SettingsInterface.KeyAddresses, "")
Await.result(connectionHandler.start(), Duration.Inf)
def port = StartingPort + index
def stop(): Unit = {
connectionHandler.stop()
Path(configFolder).deleteRecursively()
}
def onMessageReceived(msg: Message): Unit = {
eventQueue.enqueue((EventType.MessageReceived, Option(msg)))
}
def onConnectionsChanged(): Unit =
eventQueue.enqueue((EventType.ConnectionsChanged, None))
def onContactsUpdated(): Unit =
eventQueue.enqueue((EventType.ContactsUpdated, None))
}

135
integration/src/main/scala/com.nutomic.ensichat.integration/Main.scala Normal file
View file

@ -0,0 +1,135 @@
package com.nutomic.ensichat.integration
import java.io.File
import java.util.concurrent.{CountDownLatch, TimeUnit}
import com.nutomic.ensichat.core.Crypto
import com.nutomic.ensichat.core.body.Text
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration.Duration
import scala.concurrent.{Await, Future}
import scala.util.Try
/**
* Creates some local nodes, connects them and sends messages between them.
*
* If the test runs slow or fails, changing [[Crypto.PublicKeySize]] to 512 should help.
*/
object Main extends App {
val nodes = createMesh()
System.out.println("\n\nAll nodes connected!\n\n")
sendMessages(nodes)
System.out.println("\n\nAll messages sent!\n\n")
// Stop node 1, forcing route errors and messages to use the (longer) path via nodes 7 and 8.
nodes(1).connectionHandler.stop()
System.out.println("node 1 stopped")
sendMessages(nodes)
/**
* Creates a new mesh with a predefined layout.
*
* Graphical representation:
* 8 —— 7
* / \
* 0———1———3———4
* \ / | |
* 2 5———6
*
* @return List of [[LocalNode]]s, ordered from 0 to 7.
*/
private def createMesh(): Seq[LocalNode] = {
val nodes = Await.result(Future.sequence(0.to(8).map(createNode)), Duration.Inf)
sys.addShutdownHook(nodes.foreach(_.stop()))
connectNodes(nodes(0), nodes(1))
connectNodes(nodes(0), nodes(2))
connectNodes(nodes(1), nodes(2))
connectNodes(nodes(1), nodes(3))
connectNodes(nodes(3), nodes(4))
connectNodes(nodes(3), nodes(5))
connectNodes(nodes(4), nodes(6))
connectNodes(nodes(5), nodes(6))
connectNodes(nodes(3), nodes(7))
connectNodes(nodes(0), nodes(8))
connectNodes(nodes(7), nodes(8))
nodes.foreach(n => System.out.println(s"Node ${n.index} has address ${n.crypto.localAddress}"))
nodes
}
private def createNode(index: Int): Future[LocalNode] = {
val configFolder = new File(s"build/node$index/")
assert(!configFolder.exists(), s"stale config exists in $configFolder")
Future(new LocalNode(index, configFolder))
}
private def connectNodes(first: LocalNode, second: LocalNode): Unit = {
first.connectionHandler.connect(s"localhost:${second.port}")
first.eventQueue.toStream.find(_._1 == LocalNode.EventType.ConnectionsChanged)
second.eventQueue.toStream.find(_._1 == LocalNode.EventType.ConnectionsChanged)
val firstAddress = first.crypto.localAddress
val secondAddress = second.crypto.localAddress
val firstConnections = first.connectionHandler.connections()
val secondConnections = second.connectionHandler.connections()
assert(firstConnections.contains(secondAddress),
s"${first.index} is not connected to ${second.index}")
assert(secondConnections.contains(firstAddress),
s"${second.index} is not connected to ${second.index}")
System.out.println(s"${first.index} and ${second.index} connected")
}
private def sendMessages(nodes: Seq[LocalNode]): Unit = {
sendMessage(nodes(0), nodes(2))
sendMessage(nodes(2), nodes(0))
sendMessage(nodes(4), nodes(3))
sendMessage(nodes(3), nodes(5))
sendMessage(nodes(4), nodes(6))
sendMessage(nodes(2), nodes(3))
sendMessage(nodes(0), nodes(3))
sendMessage(nodes(3), nodes(6))
sendMessage(nodes(3), nodes(2))
}
private def sendMessage(from: LocalNode, to: LocalNode): Unit = {
addKey(to.crypto, from.crypto)
addKey(from.crypto, to.crypto)
System.out.println(s"sendMessage(${from.index}, ${to.index})")
val text = s"${from.index} to ${to.index}"
from.connectionHandler.sendTo(to.crypto.localAddress, new Text(text))
val latch = new CountDownLatch(1)
Future {
val exists =
to.eventQueue.toStream.exists { event =>
if (event._1 != LocalNode.EventType.MessageReceived)
false
else {
event._2.get.body match {
case t: Text => t.text == text
case _ => false
}
}
}
assert(exists, s"message from ${from.index} did not arrive at ${to.index}")
latch.countDown()
}
assert(latch.await(1000, TimeUnit.MILLISECONDS))
}
private def addKey(addTo: Crypto, addFrom: Crypto): Unit = {
if (Try(addTo.getPublicKey(addFrom.localAddress)).isFailure)
addTo.addPublicKey(addFrom.localAddress, addFrom.getLocalPublicKey)
}
}

1
server/build.gradle
View file

@ -5,6 +5,7 @@ dependencies {
compile 'org.scala-lang:scala-library:2.11.7'
compile project(path: ':core')
compile 'com.github.scopt:scopt_2.10:3.3.0'
compile 'ch.qos.logback:logback-classic:1.1.7'
}
mainClassName = 'com.nutomic.ensichat.server.Main'

42
server/src/main/scala/com/nutomic/ensichat/server/Logging.scala
View file

@ -1,42 +0,0 @@
package com.nutomic.ensichat.server
import java.io.{PrintWriter, StringWriter}
import java.text.DateFormat
import java.util.{Date, Locale}
import com.nutomic.ensichat.core.interfaces.Log
import scala.collection.mutable
class Logging extends Log {
private val logs = new mutable.Queue[String]()
def dequeue(): Seq[String] = logs.dequeueAll((String) => true)
private def enqueue(tag: String, message: String, tr: Option[Throwable]): Unit = {
val df = DateFormat.getTimeInstance(DateFormat.DEFAULT, Locale.UK)
val throwableString = tr.map { tr =>
val sw = new StringWriter()
tr.printStackTrace(new PrintWriter(sw))
"\n" + sw.toString
}
logs.enqueue(df.format(new Date()) + " " + tag + ": " + message + throwableString.getOrElse(""))
}
def v(tag: String, message: String, tr: Throwable = null): Unit =
enqueue("V/" + tag, message, Option(tr))
def d(tag: String, message: String, tr: Throwable = null): Unit =
enqueue("D/" + tag, message, Option(tr))
def i(tag: String, message: String, tr: Throwable = null): Unit =
enqueue("I/" + tag, message, Option(tr))
def w(tag: String, message: String, tr: Throwable = null): Unit =
enqueue("W/" + tag, message, Option(tr))
def e(tag: String, message: String, tr: Throwable = null): Unit =
enqueue("E/" + tag, message, Option(tr))
}

17
server/src/main/scala/com/nutomic/ensichat/server/Main.scala
View file

@ -5,24 +5,21 @@ import java.nio.file.Paths
import java.util.concurrent.TimeUnit
import com.nutomic.ensichat.core.body.Text
import com.nutomic.ensichat.core.interfaces.SettingsInterface._
import com.nutomic.ensichat.core.interfaces.{CallbackInterface, Log, SettingsInterface}
import com.nutomic.ensichat.core.interfaces.{CallbackInterface, SettingsInterface}
import com.nutomic.ensichat.core.util.Database
import com.nutomic.ensichat.core.{ConnectionHandler, Crypto, Message}
import com.typesafe.scalalogging.Logger
import scopt.OptionParser
object Main extends App with CallbackInterface {
private val Tag = "Main"
private val logger = Logger(this.getClass)
private val ConfigFolder = Paths.get("").toFile.getAbsoluteFile
private val ConfigFile = new File(ConfigFolder, "config.properties")
private val DatabaseFile = new File(ConfigFolder, "database")
private val KeyFolder = new File(ConfigFolder, "keys")
private val LogInterval = TimeUnit.SECONDS.toMillis(1)
private lazy val logInstance = new Logging()
private lazy val settings = new Settings(ConfigFile)
private lazy val crypto = new Crypto(settings, KeyFolder)
private lazy val database = new Database(DatabaseFile, this)
@ -38,7 +35,6 @@ object Main extends App with CallbackInterface {
private def init(): Unit = {
ConfigFolder.mkdirs()
KeyFolder.mkdirs()
Log.setLogInstance(logInstance)
sys.addShutdownHook(connectionHandler.stop())
val parser = new OptionParser[Config]("ensichat") {
@ -64,8 +60,7 @@ object Main extends App with CallbackInterface {
// Keep alive and print logs
while (true) {
Thread.sleep(LogInterval)
logInstance.dequeue().foreach(System.out.println)
Thread.sleep(TimeUnit.SECONDS.toMillis(1))
}
}
@ -78,9 +73,9 @@ object Main extends App with CallbackInterface {
val address = msg.header.origin
val name = connectionHandler.getUser(address).name
connectionHandler.sendTo(address, new Text("Hello " + name))
Log.i(Tag, "Received text: " + text.text)
logger.info("Received text: " + text.text)
case _ =>
Log.i(Tag, "Received msg: " + msg.body)
logger.info("Received msg: " + msg.body)
}
}

9
server/src/main/scala/com/nutomic/ensichat/server/Settings.scala
View file

@ -3,13 +3,14 @@ package com.nutomic.ensichat.server
import java.io._
import java.util.Properties
import com.nutomic.ensichat.core.interfaces.{Log, SettingsInterface}
import com.nutomic.ensichat.core.interfaces.SettingsInterface
import com.typesafe.scalalogging.Logger
import scala.collection.JavaConverters._
class Settings(file: File) extends SettingsInterface {
private val Tag = "Settings"
private val logger = Logger(this.getClass)
if (!file.exists()) {
file.createNewFile()
@ -23,7 +24,7 @@ class Settings(file: File) extends SettingsInterface {
p.load(fis)
fis.close()
} catch {
case e: IOException => Log.w(Tag, "Failed to load settings from " + file, e)
case e: IOException => logger.warn("Failed to load settings from " + file, e)
}
p
}
@ -35,7 +36,7 @@ class Settings(file: File) extends SettingsInterface {
props.store(fos, "")
fos.close()
} catch {
case e: IOException => Log.w(Tag, "Failed to write preference for key " + key, e)
case e: IOException => logger.warn("Failed to write preference for key " + key, e)
}
}

2
settings.gradle
View file

@ -1 +1 @@
include ':android', ':core', ':server'
include ':android', ':core', ':server', ':integration'