WiP: sample is able to read packets without crashing

example-android/src/main/java/com/jasonernst/example_android/PacketDumperVPNService.kt

@@ -6,6 +6,8 @@ import android.os.ParcelFileDescriptor
 import android.os.ParcelFileDescriptor.AutoCloseInputStream
 import android.os.ParcelFileDescriptor.AutoCloseOutputStream
 import com.jasonernst.knet.Packet
+import com.jasonernst.knet.network.nextheader.ICMPNextHeaderWrapper
+import com.jasonernst.knet.transport.TransportHeader
 import com.jasonernst.packetdumper.ethernet.EtherType
 import com.jasonernst.packetdumper.stringdumper.StringPacketDumper
 import kotlinx.coroutines.CoroutineScope
@@ -17,6 +19,7 @@ import org.slf4j.LoggerFactory
 import java.nio.ByteBuffer
 import java.util.concurrent.LinkedBlockingDeque
 import java.util.concurrent.atomic.AtomicBoolean
+import kotlin.math.min
 
 class PacketDumperVPNService: VpnService() {
     private val logger = LoggerFactory.getLogger(javaClass)
@@ -46,6 +49,7 @@ class PacketDumperVPNService: VpnService() {
             .addAddress(VPN6_ADDRESS, VPN_SUBNET6_MASK)
             .addDnsServer(DNS_SERVER)
             .addDnsServer(DNS6_SERVER)
+            .setMtu(MAX_RECEIVE_BUFFER_SIZE)
             .addRoute("0.0.0.0", 0)
             .addRoute("2000::", 3) // https://wiki.strongswan.org/issues/1261
 
@@ -68,25 +72,30 @@ class PacketDumperVPNService: VpnService() {
         withContext(Dispatchers.IO) {
             val stream = ByteBuffer.allocate(MAX_STREAM_BUFFER_SIZE)
             while (running.get()) {
-                val bytesRead: Int
-                try {
-                    bytesRead = inputStream.read(readBuffer)
+                // fill up the buffer with data from the OS over multiple reads, or until
+                // there is no more data to read
+                var totalBytesRead = 0
+                do {
+                    // make sure we don't overlfow the buffer
+                    var bytesToRead = min(MAX_RECEIVE_BUFFER_SIZE, stream.remaining())
+                    val bytesRead: Int = inputStream.read(readBuffer, 0, bytesToRead)
                     if (bytesRead == -1) {
-                        logger.debug("End of stream")
+                        logger.warn("End of OS stream")
                         break
                     }
-                    if (bytesRead == 0) {
-                        continue
+                    if (bytesRead > 0) {
+                        logger.debug("About to write {} bytes to buffer at position: {}", bytesRead, stream.position())
+                        stream.put(readBuffer, 0, bytesRead)
+                        totalBytesRead += bytesRead
                     }
-                    logger.debug("Read {} bytes from OS", bytesRead)
-                } catch (e: Exception) {
-                    logger.error("Error reading from OS", e)
-                    break
+                } while (bytesRead > 0 && stream.hasRemaining())
+                if (totalBytesRead > 0) {
+                    logger.debug("Read {} bytes from OS", totalBytesRead)
+                    stream.flip()
+                    handlePackets(parseStream(stream))
+                } else {
+                    Thread.sleep(100) // wait for data to arrive
                 }
-
-                stream.put(readBuffer, 0, bytesRead)
-                stream.flip()
-                val packets = parseStream(stream)
             }
         }
     }
@@ -107,26 +116,43 @@ class PacketDumperVPNService: VpnService() {
      * may be unreachable, time exceeded, etc, or just a successful ping response.
      */
     private fun parseStream(stream: ByteBuffer): List<Packet> {
-        logger.debug("GOT STREAM: \n{}", StringPacketDumper().dumpBufferToString(buffer = stream, addresses = true, etherType = EtherType.DETECT))
+        logger.debug("GOT STREAM: \n{}", StringPacketDumper().dumpBufferToString(buffer = stream, addresses = true, etherType = null))
         val packets = mutableListOf<Packet>()
         while (stream.hasRemaining()) {
             val position = stream.position()
             try {
                 val packet = Packet.fromStream(stream)
                 logger.debug("Parsed packet: {}", packet)
+                logger.debug("Stream position after parsing: {} limit: {}", stream.position(), stream.limit())
                 packets.add(packet)
             } catch (e: IllegalArgumentException) {
                 // don't bother to rewind the stream, just log and continue at position + 1
                 logger.error("Error parsing stream: ", e)
                 stream.position(position + 1)
-            } catch (e: PacketTooShortException) {
+            } catch (e: com.jasonernst.knet.PacketTooShortException) {
+                logger.warn("Packet too short to parse, trying again when more data arrives: {}", e.message)
+                logger.debug("POSITION: {} LIMIT: {}, RESETTING TO START: {}", stream.position(), stream.limit(), position)
                 // rewind the stream to before we tried parsing so we can try again later
                 stream.position(position)
                 break
             }
         }
+        logger.debug("Stream position before compact: {} limit: {}", stream.position(), stream.limit())
+        stream.compact()
+        logger.debug("Stream position after compact: {} limit: {}", stream.position(), stream.limit())
         return packets
+    }
 
+    private fun handlePackets(packets: List<Packet>) {
+        packets.forEach { packet ->
+            if (packet.nextHeaders is TransportHeader) {
+                // todo: use https://github.com/compscidr/kanonproxy
+            } else if (packet.nextHeaders is ICMPNextHeaderWrapper) {
+                // todo: use https://github.com/compscidr/kanonproxy
+            } else {
+                logger.error("Unsupported packet type: {}", packet.javaClass)
+            }
+        }
     }
 
     private suspend fun readFromInternetWriteToOS(outputStream: AutoCloseOutputStream) {
@@ -157,6 +183,6 @@ class PacketDumperVPNService: VpnService() {
         private const val DNS_SERVER = "8.8.8.8"
         private const val DNS6_SERVER = "2001:4860:4860::8888"
         private const val MAX_STREAM_BUFFER_SIZE = 1048576 // max we can write into the stream without parsing
-        private const val MAX_RECEIVE_BUFFER_SIZE = 1024 // max amount we can recv in one read
+        private const val MAX_RECEIVE_BUFFER_SIZE = 1500   // max amount we can recv in one read (should be the MTU or bigger probably)
     }
 }

gradle/libs.versions.toml

@@ -6,7 +6,7 @@ compose-activity = "1.9.1"
 espresso-core = "3.0.2"
 junit = "4.13.2"
 jupiter = "5.11.2"
-knet = "0.0.2"
+knet = "0.0.4"
 kotlin = "2.0.20"
 kotlinter = "4.4.1"
 logback-android = "3.0.0"

packetdumper/src/test/kotlin/com/jasonernst/packetdumper/filedumper/TestPcapNgFilePacketDumper.kt

@@ -10,7 +10,6 @@ import org.junit.jupiter.api.AfterEach
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Test
 import org.slf4j.LoggerFactory
-import java.io.File
 import java.nio.ByteBuffer
 
 class TestPcapNgFilePacketDumper {
@@ -24,7 +23,7 @@ class TestPcapNgFilePacketDumper {
 
         // delete the file created for the test to cleanup
         try {
-            //File(dumper.filename).delete()
+            // File(dumper.filename).delete()
             logger.debug("Deleted file ${dumper.filename}")
         } catch (e: Exception) {
             // ignore