progress towards multi-server handling

- bump some versions
- validate Factorio server ID & token
-

modules/events-server/build.gradle.kts

@@ -40,6 +40,8 @@ dependencies {
   implementation(libs.ktorSerialization.kotlinxJson)
 
   with(libs.ktorServer) {
+    implementation(auth)
+    implementation(authJwt)
     implementation(core)
     implementation(callId)
     implementation(callLogging)
@@ -59,9 +61,13 @@ dependencies {
   }
   //</editor-fold>
 
-
   implementation(libs.simpleSyslogParser)
 
+  // https://search.maven.org/artifact/org.bouncycastle/bcprov-jdk15on
+  implementation("org.bouncycastle:bcprov-jdk15on:1.70")
+  implementation("com.brendangoldberg:kotlin-jwt:1.1.0")
+  implementation("com.auth0:java-jwt:4.0.0")
+
   implementation(libs.kafka.kotkaStreams) {
     isChanging = false
   }

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/config/ApplicationProperties.kt

@@ -3,6 +3,9 @@ package dev.adamko.kafkatorio.server.config
 import com.sksamuel.hoplite.ConfigLoaderBuilder
 import com.sksamuel.hoplite.addEnvironmentSource
 import com.sksamuel.hoplite.addResourceOrFileSource
+import dev.adamko.kafkatorio.schema.common.FactorioServerId
+import dev.adamko.kafkatorio.schema.common.FactorioServerToken
+import dev.adamko.kafkatorio.schema.common.MaskedValue
 import java.nio.file.Path
 
 
@@ -15,6 +18,10 @@ data class ApplicationProperties(
 
   val socketServerHost: String,
   val socketServerPort: Int,
+
+  val jwtSecret: MaskedValue,
+
+  val kafkatorioServers: Map<FactorioServerToken, FactorioServerId>
 ) {
 
   companion object {
@@ -23,7 +30,7 @@ data class ApplicationProperties(
       ConfigLoaderBuilder.default()
         .addDefaults()
         .allowUnresolvedSubstitutions()
-        .allowEmptyTree()
+        .allowEmptySources()
         .addEnvironmentSource()
         .addResourceOrFileSource("/.secret.config.yml", optional = true)
         .addResourceOrFileSource("/config.yml", optional = false)

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/eventsServer.kt

@@ -2,6 +2,7 @@ package dev.adamko.kafkatorio.server
 
 import dev.adamko.kafkatorio.server.config.ApplicationProperties
 import dev.adamko.kafkatorio.server.processor.eventsProcessor
+import dev.adamko.kafkatorio.server.security.Authenticator
 import dev.adamko.kafkatorio.server.socket.SyslogSocketServer
 import dev.adamko.kafkatorio.server.web.startWebServer
 import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer
@@ -15,6 +16,7 @@ suspend fun main(): Unit = coroutineScope {
 
   val websocketServer = WebmapWebsocketServer()
   val syslogSocketServer = SyslogSocketServer(appProps)
+  val authenticator = Authenticator(appProps)
 
   launch {
     startWebServer(appProps, websocketServer)
@@ -25,6 +27,6 @@ suspend fun main(): Unit = coroutineScope {
   }
 
   launch {
-    eventsProcessor(appProps, syslogSocketServer, websocketServer)
+    eventsProcessor(appProps, syslogSocketServer, websocketServer, authenticator)
   }
 }

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/KafkatorioPacketKafkaProducer.kt

@@ -0,0 +1,172 @@
+package dev.adamko.kafkatorio.server.processor
+
+import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
+import dev.adamko.kafkatorio.schema.common.FactorioServerId
+import dev.adamko.kafkatorio.server.config.ApplicationProperties
+import dev.adamko.kafkatorio.server.config.jsonMapper
+import dev.adamko.kafkatorio.server.security.Authenticator
+import dev.adamko.kafkatorio.server.socket.SyslogMsg
+import dev.adamko.kafkatorio.server.socket.SyslogSocketServer
+import dev.adamko.kotka.kxs.kafkaSerializer
+import java.util.Properties
+import kotlin.coroutines.coroutineContext
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.job
+import org.apache.kafka.clients.producer.KafkaProducer
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.common.header.internals.RecordHeader
+import org.apache.kafka.common.serialization.Serdes
+import org.apache.kafka.streams.StreamsConfig
+
+
+/**
+ * Consume Syslog messages, verifies each, and forwards them to a Kafka topic.
+ */
+class KafkatorioPacketKafkaProducer(
+  private val appProps: ApplicationProperties,
+  private val syslogServer: SyslogSocketServer,
+  private val authenticator: Authenticator,
+) : AutoCloseable {
+
+  private val props: Properties = appProps.kafkaStreamsConfig.toProperties()
+
+  init {
+    props.updateValue(StreamsConfig.APPLICATION_ID_CONFIG) { "$it.KafkatorioPacketKafkaProducer" }
+  }
+
+  private val producer: KafkaProducer<FactorioServerId, String> = KafkaProducer(
+    props.apply {
+      updateValue(StreamsConfig.APPLICATION_ID_CONFIG) { "$it.KafkatorioPacketKafkaProducer" }
+    },
+    jsonMapper.kafkaSerializer(FactorioServerId.serializer()),
+    Serdes.String().serializer(),
+  )
+
+  private val dlqProducer: KafkaProducer<String, String> = KafkaProducer(
+    props.apply {
+      updateValue(StreamsConfig.APPLICATION_ID_CONFIG) { "$it.KafkatorioPacketKafkaProducer.dlq" }
+    },
+    Serdes.String().serializer(),
+    Serdes.String().serializer(),
+  )
+
+
+  suspend fun launch() {
+    coroutineContext.job.invokeOnCompletion {
+      this@KafkatorioPacketKafkaProducer.close()
+    }
+
+    syslogServer.messages
+      .filter {
+        (it.message ?: "").run {
+          startsWith("KafkatorioPacket:::") || startsWith("KafkatorioPacket encoded:::")
+        }
+      }.onEach { syslog ->
+        runCatching {
+          getFactorioServerId(syslog)
+        }.fold(
+          onSuccess = { serverId -> produceMessage(serverId, syslog) },
+          onFailure = { ex ->
+            produceDlqMessage(syslog, "${ex::class.java.name}: ${ex.message}")
+
+            if (ex is CancellationException) throw ex
+          }
+        )
+      }.collect()
+  }
+
+
+  private fun getFactorioServerId(syslog: SyslogMsg): FactorioServerId {
+    val token = syslog.headerAppName
+    if (token.isNullOrBlank()) {
+      error("no token in syslog message. msg: ${syslog.src}")
+    }
+
+    val serverToken = authenticator.verifyJwt(token)
+    if (serverToken.isNullOrBlank()) {
+      error("could not decode token $token. msg: ${syslog.src}")
+    }
+
+    val serverId = appProps.kafkatorioServers[serverToken]
+    if (serverId.isNullOrBlank()) {
+      error("no server ID for token $serverToken. msg: ${syslog.src}")
+    }
+
+    return serverId
+  }
+
+
+  private fun produceMessage(
+    serverId: FactorioServerId,
+    syslog: SyslogMsg,
+  ) {
+    val kafkatorioMessage = syslog.message?.substringAfter(":::")
+
+    val record = ProducerRecord<FactorioServerId, String>(
+      TOPIC_SRC_SERVER_LOG,
+      serverId,
+      kafkatorioMessage,
+    )
+
+    syslog.copyHeadersTo(record)
+
+    producer.send(record)
+  }
+
+
+  private fun produceDlqMessage(syslog: SyslogMsg, errorMessage: String) {
+    log("error: $errorMessage, $syslog")
+
+    val record = ProducerRecord(
+      TOPIC_SRC_SERVER_LOG_DLQ,
+      "unknown",
+      errorMessage,
+    )
+
+    syslog.copyHeadersTo(record)
+
+    dlqProducer.send(record)
+  }
+
+
+  override fun close() {
+    producer.close()
+  }
+
+
+  companion object {
+    private fun log(msg: String) = println("[KafkatorioPacketKafkaProducer] $msg")
+
+
+    private fun Properties.updateValue(
+      key: String,
+      update: (oldValue: String) -> String
+    ) {
+      val newValue = update(getValue(key) as String)
+      setProperty(key, newValue)
+    }
+
+
+    private fun SyslogMsg.copyHeadersTo(record: ProducerRecord<*, *>) {
+      val headers = SyslogFieldKeys.values()
+        .filter { "HEADER_" in it.name }
+        .mapNotNull { headerKey ->
+          val headerValue = src[headerKey.field]
+          when {
+            headerValue.isNullOrBlank() -> null
+            else                        -> RecordHeader(
+              headerKey.field,
+              headerValue.toByteArray()
+            )
+          }
+        }
+
+      headers.forEach { header ->
+        record.headers().add(header)
+      }
+    }
+  }
+}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/KafkatorioTopology.kt

@@ -1,6 +1,5 @@
 package dev.adamko.kafkatorio.server.processor
 
-import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
 import dev.adamko.kafkatorio.schema.common.FactorioServerId
 import dev.adamko.kafkatorio.schema.common.PrototypeHashCode
 import dev.adamko.kafkatorio.schema.common.ServerMapChunkId
@@ -31,17 +30,10 @@ import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.currentCoroutineContext
-import kotlinx.coroutines.flow.filter
-import kotlinx.coroutines.flow.launchIn
-import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.job
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.suspendCancellableCoroutine
-import org.apache.kafka.clients.producer.KafkaProducer
 import org.apache.kafka.clients.producer.ProducerConfig
-import org.apache.kafka.clients.producer.ProducerRecord
-import org.apache.kafka.common.header.internals.RecordHeader
-import org.apache.kafka.common.serialization.Serdes
 import org.apache.kafka.streams.KafkaStreams
 import org.apache.kafka.streams.StreamsBuilder
 import org.apache.kafka.streams.StreamsConfig
@@ -54,6 +46,7 @@ internal class KafkatorioTopology(
   private val appProps: ApplicationProperties,
   private val wsServer: WebmapWebsocketServer,
   private val syslogServer: SyslogSocketServer,
+  private val kafkatorioPacketKafkaProducer: KafkatorioPacketKafkaProducer,
 ) {
 
   private val builderScope = CoroutineScope(CoroutineName("KafkaTopology"))
@@ -168,60 +161,10 @@ internal class KafkatorioTopology(
 
 
   /** Receive log messages from the syslog socket, and send them to a Kafka topic */
-  private fun syslogProducer(
-    id: String = "syslog-producer"
-  ) {
-    val props: Properties = appProps.kafkaStreamsConfig.toProperties()
-
-    val appId = props.compute(StreamsConfig.APPLICATION_ID_CONFIG) { _, v -> "$v.$id" } as String
-    props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, appId)
-
-
-    val producer: KafkaProducer<String, String> = KafkaProducer(
-      props,
-      Serdes.String().serializer(),
-      Serdes.String().serializer(),
-    )
-
-    builderScope.coroutineContext.job.invokeOnCompletion {
-      producer.close()
+  private fun syslogProducer() {
+    builderScope.launch {
+      kafkatorioPacketKafkaProducer.launch()
     }
-
-    syslogServer.messages
-      .filter {
-        (it.message ?: "").run {
-          startsWith("KafkatorioPacket:::") || startsWith("KafkatorioPacket encoded:::")
-        }
-      }
-      .onEach { syslogMsg ->
-
-        val kafkatorioMessage = syslogMsg.message?.substringAfter(":::")
-
-        val record = ProducerRecord<String, String>(
-          TOPIC_SRC_SERVER_LOG,
-          "syslog-test",
-          kafkatorioMessage,
-        )
-
-        val headers = SyslogFieldKeys.values()
-          .filter { "HEADER_" in it.name }
-          .mapNotNull { headerKey ->
-            val headerValue = syslogMsg.src[headerKey.field]
-            when {
-              headerValue.isNullOrBlank() -> null
-              else                        -> RecordHeader(
-                headerKey.field,
-                headerValue.toByteArray()
-              )
-            }
-          }
-
-        headers.forEach { header ->
-          record.headers().add(header)
-        }
-
-        producer.send(record)
-      }.launchIn(builderScope)
   }
 
 
@@ -251,6 +194,7 @@ internal class KafkatorioTopology(
           when (newState) {
             KafkaStreams.State.NOT_RUNNING,
             KafkaStreams.State.ERROR -> if (cont.isActive) cont.resume(newState)
+
             else                     -> Unit // do nothing
           }
         }

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/eventProcessor.kt

@@ -3,6 +3,7 @@ package dev.adamko.kafkatorio.server.processor
 
 import dev.adamko.kafkatorio.server.config.ApplicationProperties
 import dev.adamko.kafkatorio.server.processor.admin.KafkatorioKafkaAdmin
+import dev.adamko.kafkatorio.server.security.Authenticator
 import dev.adamko.kafkatorio.server.socket.SyslogSocketServer
 import dev.adamko.kafkatorio.server.web.websocket.WebmapWebsocketServer
 import kotlinx.coroutines.awaitCancellation
@@ -14,12 +15,15 @@ suspend fun eventsProcessor(
   appProps: ApplicationProperties,
   syslogServer: SyslogSocketServer,
   wsServer: WebmapWebsocketServer,
+  authenticator: Authenticator,
 ): Unit = coroutineScope {
 
   val admin = KafkatorioKafkaAdmin(appProps)
   admin.createKafkatorioTopics()
 
-  val topology = KafkatorioTopology(appProps, wsServer, syslogServer)
+  val kafkatorioPacketKafkaProducer = KafkatorioPacketKafkaProducer(appProps, syslogServer, authenticator)
+
+  val topology = KafkatorioTopology(appProps, wsServer, syslogServer, kafkatorioPacketKafkaProducer)
 
   launch {
     topology.start()

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/kafkatorioTopics.kt

@@ -18,6 +18,7 @@ import kotlin.reflect.KClass
 private const val DOMAIN = "kafkatorio"
 
 const val TOPIC_SRC_SERVER_LOG = "$DOMAIN.src.server-log"
+const val TOPIC_SRC_SERVER_LOG_DLQ = "$DOMAIN.src.server-log.dlq"
 
 //const val TOPIC_GROUPED_MAP_CHUNKS_STATE = "$DOMAIN.state.map-chunks.grouped"
 
@@ -41,6 +42,7 @@ const val TOPIC_BROADCAST_TO_WEBSOCKET = "$DOMAIN.broadcast.websocket"
 
 fun allTopics(): Set<String> = buildSet {
   add(TOPIC_SRC_SERVER_LOG)
+  add(TOPIC_SRC_SERVER_LOG_DLQ)
 
 //  add(TOPIC_GROUPED_MAP_CHUNKS_STATE)
 

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/processor/topology/factorioServerPackets.kt

@@ -11,11 +11,11 @@ import dev.adamko.kotka.extensions.component1
 import dev.adamko.kotka.extensions.component2
 import dev.adamko.kotka.extensions.consumedAs
 import dev.adamko.kotka.extensions.producedAs
-import dev.adamko.kotka.extensions.streams.map
+import dev.adamko.kotka.extensions.streams.mapValues
 import dev.adamko.kotka.extensions.streams.to
 import dev.adamko.kotka.kxs.serde
 import java.util.zip.Inflater
-import kotlinx.serialization.decodeFromString
+import kotlinx.coroutines.CancellationException
 import okio.Buffer
 import okio.ByteString
 import okio.ByteString.Companion.decodeBase64
@@ -34,29 +34,32 @@ fun factorioServerPacketStream(
     TOPIC_SRC_SERVER_LOG,
     consumedAs(
       "read-raw-packets-from-server",
-      Serdes.String(),
+      jsonMapper.serde(FactorioServerId.serializer()),
       Serdes.String(),
     )
-  ).map("decode-packets") { serverId: String, value: String ->
-    val packet = runCatching {
-      val decoded = decodeFactorioEncodedString(value)
-      jsonMapper.decodeFromString<KafkatorioPacket>(decoded)
-    }.getOrElse { e ->
+  ).mapValues("decode-packets") { _: FactorioServerId, value: String? ->
+
+    val packet = try {
+      val decoded = decodeFactorioEncodedString(value!!)
+      jsonMapper.decodeFromString(KafkatorioPacket.serializer(), decoded)
+    } catch (e: Exception) {
+      if (e is CancellationException) throw e
+
       KafkatorioPacket(
         modVersion = "unknown",
         tick = Tick(0u),
         data = KafkatorioPacketDataError(
-          message = e.message,
+          exception = e,
           rawValue = value,
         )
       )
     }
 
     if (packet.data is KafkatorioPacketDataError) {
-      println("error parsing $TOPIC_SRC_SERVER_LOG message: ${packet.data}")
+      println("error parsing $TOPIC_SRC_SERVER_LOG message: $packet")
     }
 
-    FactorioServerId(serverId) to packet
+    packet
   }
 }
 

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/security/Authenticator.kt

@@ -0,0 +1,125 @@
+package dev.adamko.kafkatorio.server.security
+
+import com.auth0.jwt.JWT
+import com.auth0.jwt.algorithms.Algorithm
+import com.auth0.jwt.interfaces.DecodedJWT
+import dev.adamko.kafkatorio.schema.common.FactorioServerToken
+import dev.adamko.kafkatorio.server.config.ApplicationProperties
+import kotlin.reflect.jvm.jvmName
+import kotlinx.coroutines.CancellationException
+import kotlinx.serialization.SerialName
+import kotlinx.serialization.Serializable
+import kotlinx.serialization.json.Json
+import okio.ByteString.Companion.decodeBase64
+
+
+class Authenticator(
+  appProps: ApplicationProperties,
+) {
+  private val algorithm: Algorithm = Algorithm.HMAC256(appProps.jwtSecret.value)
+
+  private val verifier = JWT.require(algorithm).build()
+
+  fun verifyJwt(token: String): FactorioServerToken? {
+    return runCatching {
+      val decodedJwt: DecodedJWT? = verifier.verify(token)
+
+      val payloadJson = decodedJwt?.payload?.decodeBase64()?.string(Charsets.UTF_8) ?: "{}"
+
+      jsonMapper.decodeFromString(
+        KafkatorioJwtPayload.serializer(),
+        payloadJson,
+      ).serverToken
+
+    }.onFailure { e ->
+      if (e is CancellationException) throw e
+
+      println("[Authenticator] Could not decode JWT ${e::class.jvmName}: ${e.message}")
+    }.getOrNull()
+  }
+
+  companion object {
+    private val jsonMapper = Json {
+      ignoreUnknownKeys = true
+    }
+  }
+}
+
+
+@Serializable
+@SerialName("kafkatorio.auth.KafkatorioJwtPayload")
+private data class KafkatorioJwtPayload(
+  @SerialName("sub")
+  val serverToken: FactorioServerToken,
+  /** unix epoch seconds*/
+  @SerialName("iat")
+  val issuedAtTime: Long,
+)
+
+
+//// https://github.com/auth0/java-jwt/blob/f799e589c3d43de318c7ca0f0c36e462297db973/lib/src/test/java/com/auth0/jwt/PemUtils.java
+//private object PemUtils {
+//
+//  fun readPublicKey(filepath: String, algorithm: String): PublicKey {
+//    val bytes = parsePemFile(File(filepath))
+//    return getPublicKey(bytes, algorithm)
+//  }
+//
+////  fun readPublicKey(reader: Reader, algorithm: String): PublicKey {
+////     return getPublicKey(bytes, algorithm)
+////  }
+//
+//  fun readPrivateKey(filepath: String, algorithm: String): PrivateKey {
+//    val bytes = parsePemFile(File(filepath))
+//    return getPrivateKey(bytes, algorithm)
+//  }
+//
+//  private fun parsePemFile(pemFile: File): ByteArray {
+//    return when {
+//      !pemFile.isFile   ->
+//        throw FileNotFoundException("'${pemFile.absolutePath}' is not a file")
+//
+//      !pemFile.exists() ->
+//        throw FileNotFoundException("The file '${pemFile.absolutePath}' doesn't exist.")
+//
+//      else              ->
+//        PemReader(FileReader(pemFile)).use { reader ->
+//          reader.readPemObject().content
+//        }
+//    }
+//  }
+//
+//  private fun getPublicKey(keyBytes: ByteArray, algorithm: String): PublicKey =
+//    getKey(algorithm) {
+//      val keySpec: EncodedKeySpec = X509EncodedKeySpec(keyBytes)
+//      generatePublic(keySpec)
+//    }
+//
+//  private fun getPrivateKey(keyBytes: ByteArray, algorithm: String): PrivateKey =
+//    getKey(algorithm) {
+//      val keySpec: EncodedKeySpec = PKCS8EncodedKeySpec(keyBytes)
+//      generatePrivate(keySpec)
+//    }
+//
+//
+//  private fun <K : Key> getKey(
+//    algorithm: String,
+//    generateKey: KeyFactory.() -> K
+//  ): K = runCatching {
+//    val kf = KeyFactory.getInstance(algorithm)
+//    kf.generateKey()
+//  }.getOrElse { e ->
+//    when (e) {
+//      is NoSuchAlgorithmException ->
+//        throw IllegalStateException(
+//          "Could not reconstruct key, algorithm '$algorithm' could not be found.",
+//          e
+//        )
+//
+//      is InvalidKeySpecException  ->
+//        throw IllegalStateException("Could not reconstruct key", e)
+//
+//      else                        -> throw e
+//    }
+//  }
+//}

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/socket/SyslogMsg.kt

@@ -3,7 +3,7 @@ package dev.adamko.kafkatorio.server.socket
 import com.github.palindromicity.syslog.dsl.SyslogFieldKeys
 
 data class SyslogMsg(
-  val src: Map<String, String?>
+  val src: Map<String, String?>,
 ) {
 
   val message: String?
@@ -48,16 +48,4 @@ data class SyslogMsg(
   val structuredElementIdPnamePattern: String?
     get() = src[SyslogFieldKeys.STRUCTURED_ELEMENT_ID_PNAME_PATTERN.field]
 
-
-  companion object {
-    fun Map<String, Any?>.toSyslogMsg(): SyslogMsg {
-      val map = entries.mapNotNull { (k, v) ->
-        when (v) {
-          is String -> k to v
-          else      -> null
-        }
-      }.toMap()
-      return SyslogMsg(map)
-    }
-  }
 }

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/socket/SyslogSocketServer.kt

@@ -8,13 +8,13 @@ import com.github.palindromicity.syslog.SyslogParser
 import com.github.palindromicity.syslog.SyslogParserBuilder
 import com.github.palindromicity.syslog.SyslogSpecification
 import dev.adamko.kafkatorio.server.config.ApplicationProperties
-import dev.adamko.kafkatorio.server.socket.SyslogMsg.Companion.toSyslogMsg
 import io.ktor.network.selector.SelectorManager
 import io.ktor.network.sockets.Socket
 import io.ktor.network.sockets.aSocket
 import io.ktor.network.sockets.openReadChannel
 import io.ktor.utils.io.readUTF8Line
 import java.util.EnumSet
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.MutableSharedFlow
@@ -30,10 +30,8 @@ class SyslogSocketServer(
   appProps: ApplicationProperties,
 ) {
 
-  private val selectorManager: SelectorManager = SelectorManager(Dispatchers.IO)
-
-
-  private val serverSocketTcp = aSocket(selectorManager)
+  /** Incoming messages */
+  private val serverSocketTcp = aSocket(SelectorManager(Dispatchers.IO))
     .tcp()
     .bind(
       appProps.socketServerHost,
@@ -50,8 +48,7 @@ class SyslogSocketServer(
           NilPolicy.OMIT,
           EnumSet.of(AllowableDeviations.PRIORITY, AllowableDeviations.VERSION),
         )
-      )
-      .build()
+      ).build()
 
 
   private val _messages: MutableSharedFlow<SyslogMsg> = MutableSharedFlow()
@@ -74,31 +71,30 @@ class SyslogSocketServer(
               channel.readUTF8Line()
             }.fold(
               onSuccess = { line ->
-                when (line) {
-                  null -> {
-                    log("received 'null' from ${socket.description()} - closing connection")
-                    socket.close()
-                  }
-//                  else -> log("received '$line' from ${socket.description()}")
+                if (line == null) {
+                  log("received 'null' from ${socket.description()} - closing connection")
+                  socket.close()
                 }
+                // log("received '$line' from ${socket.description()}")
                 line
               },
-              onFailure = { e ->
-                log("reading line from socket failed ${socket.description()}, cause: $e")
+              onFailure = { ex ->
+                if (ex is CancellationException) throw ex
+
+                log("reading line from socket failed ${socket.description()}, cause: $ex")
                 null
               },
             )
             if (line != null) {
-
-              val syslog = syslogParser.parseLine(line).toSyslogMsg()
-
+              val syslog = SyslogMsg(syslogParser.parseLine(line))
               _messages.emit(syslog)
             }
-
             yield()
           }
         } catch (e: Exception) {
-          log("closing socket ${socket.description()}, cause: $e")
+          if (e is CancellationException) throw e
+          log("closing socket ${socket.description()}, cause: ${e::class.java.name} ${e.message}")
+          e.printStackTrace()
           withContext(Dispatchers.IO) { socket.close() }
         }
       }

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/rest/factorioServersRoutes.kt

@@ -6,6 +6,7 @@ import dev.adamko.kafkatorio.schema.common.ServerMapTileLayer
 import dev.adamko.kafkatorio.schema.common.ServerMapTilePngFilename
 import dev.adamko.kafkatorio.schema.common.SurfaceIndex
 import dev.adamko.kafkatorio.server.config.ApplicationProperties
+import dev.adamko.kafkatorio.server.config.jsonMapper
 import io.ktor.http.HttpStatusCode
 import io.ktor.http.content.EntityTagVersion
 import io.ktor.http.content.versions
@@ -21,6 +22,7 @@ import io.ktor.util.combineSafe
 import io.ktor.util.pipeline.PipelineContext
 import kotlin.io.path.Path
 import kotlinx.serialization.Serializable
+import kotlinx.serialization.json.encodeToJsonElement
 
 
 fun Routing.factorioServersRoutes(
@@ -28,7 +30,9 @@ fun Routing.factorioServersRoutes(
 ) {
 
   get<FactorioServers> {
-    call.respond(HttpStatusCode.NotImplemented)
+    val serverIds = appProps.kafkatorioServers.values.toSet().sortedBy { it.id }
+    val serverIdsJson = jsonMapper.encodeToJsonElement(serverIds)
+    call.respond(serverIdsJson)
   }
 
   get<FactorioServer.MapData.Tile> { tile ->

modules/events-server/src/main/kotlin/dev/adamko/kafkatorio/server/web/websocket/WebmapWebsocketServer.kt

@@ -7,6 +7,7 @@ import io.ktor.websocket.send
 import java.util.concurrent.atomic.AtomicInteger
 import kotlin.time.Duration.Companion.seconds
 import kotlin.time.TimeSource
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.coroutineScope
@@ -57,6 +58,8 @@ class WebmapWebsocketServer {
         if (client.isActive) {
           client.outgoing.send(Frame.Text(msg))
         }
+      }.onFailure { e ->
+        if (e is CancellationException) throw e
       }
     }.launchIn(client)
 
@@ -68,6 +71,8 @@ class WebmapWebsocketServer {
       while (isActive) {
         runCatching {
           client.send("elapsed ${startTime.elapsedNow()}")
+        }.onFailure { e ->
+          if (e is CancellationException) throw e
         }
         delay(10.seconds)
         yield()
@@ -100,7 +105,11 @@ class WebmapWebsocketServer {
       onSuccess = {
 //        log("sent message to client $name: $frame")
       },
-      onFailure = { e -> log("failed to send message to client $name: $e") }
+      onFailure = { e ->
+        if (e is CancellationException) throw e
+
+        log("failed to send message to client $name: $e")
+      }
     )
 
     override fun equals(other: Any?): Boolean = name == other

modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/server/processor/topology/FactorioServerPacketsTest.kt

@@ -0,0 +1,48 @@
+package dev.adamko.kafkatorio.server.processor.topology
+
+import dev.adamko.kafkatorio.library.kxsBinary
+import dev.adamko.kafkatorio.schema.common.ColourHex
+import dev.adamko.kafkatorio.schema.common.FactorioServerId
+import dev.adamko.kafkatorio.schema.common.ServerMapChunkId
+import dev.adamko.kotka.kxs.serde
+import io.kotest.core.spec.style.FunSpec
+import kotlin.time.Duration
+import kotlin.time.toJavaDuration
+import org.apache.kafka.streams.StreamsBuilder
+import org.apache.kafka.streams.TestInputTopic
+import org.apache.kafka.streams.TopologyTestDriver
+
+class FactorioServerPacketsTest : FunSpec({
+
+
+})
+
+private class Scenario(
+  streamsBuilder: StreamsBuilder = StreamsBuilder(),
+
+  groupedMapChunksInputTopicName: String = "kafkatorio.state.map-chunk.terrain.colour",
+
+  val serverId: FactorioServerId = FactorioServerId("test-server-id"),
+) : AutoCloseable {
+
+  init {
+    factorioServerPacketStream(streamsBuilder)
+  }
+
+  private val testDriver: TopologyTestDriver = TopologyTestDriver(streamsBuilder.build())
+
+  val groupedMapChunksInputTopic: TestInputTopic<ServerMapChunkId, ServerMapChunkTiles<ColourHex>> =
+    testDriver.createInputTopic(
+      groupedMapChunksInputTopicName,
+      kxsBinary.serde<ServerMapChunkId>().serializer(),
+      kxsBinary.serde<ServerMapChunkTiles<ColourHex>>().serializer(),
+    )
+
+  operator fun plusAssign(wallClockTime: Duration) {
+    testDriver.advanceWallClockTime(wallClockTime.toJavaDuration())
+  }
+
+  override fun close() {
+    testDriver.close()
+  }
+}

modules/events-server/src/test/kotlin/dev/adamko/kafkatorio/server/security/AuthenticatorTest.kt

@@ -0,0 +1,61 @@
+package dev.adamko.kafkatorio.server.security
+
+import dev.adamko.kafkatorio.schema.common.MaskedValue
+import dev.adamko.kafkatorio.server.config.ApplicationProperties
+import dev.adamko.kafkatorio.server.config.PortVal
+import io.kotest.core.spec.style.FunSpec
+import io.kotest.matchers.nulls.shouldBeNull
+import io.kotest.matchers.nulls.shouldNotBeNull
+import kotlin.io.path.Path
+
+
+class AuthenticatorTest : FunSpec({
+
+  test("verify correct JWT is verified") {
+
+    val authenticator = Authenticator(
+      testApplicationProperties(
+        jwtSecret = "123 456 789"
+      )
+    )
+
+    val result = authenticator.verifyJwt(jwtMessage)
+
+    result.shouldNotBeNull()
+  }
+
+  test("verify invalid JWT returns null") {
+
+    val authenticator = Authenticator(
+      testApplicationProperties(
+        jwtSecret = "wrong"
+      )
+    )
+
+    val result = authenticator.verifyJwt(jwtMessage)
+
+    result.shouldBeNull()
+  }
+
+}) {
+
+  companion object {
+    /** Key: `123 465 789` */
+    private val jwtMessage = """
+        eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.ToFLkPBs_EKF4B2bRQyVPfkKrB-x6WgyAq6KsR9oHwg
+    """.trimIndent()
+
+    private fun testApplicationProperties(
+      jwtSecret: String,
+    ) = ApplicationProperties(
+      webPort = PortVal.DEFAULT,
+      serverDataDir = Path(""),
+      kafkaAdminConfig = emptyMap(),
+      kafkaStreamsConfig = emptyMap(),
+      socketServerHost = "",
+      socketServerPort = 9999,
+      jwtSecret = MaskedValue(jwtSecret),
+      kafkatorioServers = emptyMap(),
+    )
+  }
+}