upApp

upApp/app/src/main/java/de/upapp/WebSocket.kt
WebSocket.kt
⎇
Raw
1package de.upapp
2
3import android.util.Log
4import org.json.JSONObject
5import org.json.JSONTokener
6import java.io.ByteArrayOutputStream
7import java.io.DataInputStream
8import java.io.EOFException
9import java.io.IOException
10import java.io.InputStream
11import java.io.OutputStream
12import java.security.MessageDigest
13import java.security.SecureRandom
14import java.util.Base64
15
16internal const val WS_TEXT = 1
17internal const val WS_CLOSE = 8
18internal const val WS_PING = 9
19internal const val WS_PONG = 10
20
21internal fun wsAccept(key: String): String = Base64.getEncoder().encodeToString(
22 MessageDigest.getInstance("SHA-1").digest((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").toByteArray())
23)
24
25/** Writes one unfragmented frame. Clients must [mask], servers must not. */
26internal fun writeFrame(out: OutputStream, opcode: Int, payload: ByteArray, mask: Boolean) = synchronized(out) {
27 val frame = ByteArrayOutputStream()
28 frame.write(0x80 or opcode)
29 val m = if (mask) 0x80 else 0
30 val n = payload.size
31 when {
32 n < 126 -> frame.write(m or n)
33 n < 0x10000 -> frame.write(byteArrayOf((m or 126).toByte(), (n shr 8).toByte(), n.toByte()))
34 else -> {
35 frame.write(m or 127)
36 for (i in 7 downTo 0) frame.write((n.toLong() shr (8 * i)).toInt())
37 }
38 }
39 if (mask) {
40 val key = ByteArray(4).also { SecureRandom().nextBytes(it) }
41 frame.write(key)
42 frame.write(ByteArray(n) { (payload[it].toInt() xor key[it % 4].toInt()).toByte() })
43 } else {
44 frame.write(payload)
45 }
46 out.write(frame.toByteArray())
47 out.flush()
48}
49
50/** Reads the next data message. Answers pings. Throws [EOFException] on a close frame. */
51internal fun readMessage(input: InputStream, output: OutputStream, mask: Boolean): Pair<Int, ByteArray> {
52 val inp = DataInputStream(input)
53 var opcode = 0
54 val message = ByteArrayOutputStream()
55 while (true) {
56 val b0 = inp.readUnsignedByte()
57 val b1 = inp.readUnsignedByte()
58 val len = when (val l = b1 and 0x7F) {
59 126 -> inp.readUnsignedShort()
60 127 -> inp.readLong().toInt()
61 else -> l
62 }
63 val key = if (b1 and 0x80 != 0) ByteArray(4).also { inp.readFully(it) } else null
64 val data = ByteArray(len).also { inp.readFully(it) }
65 key?.let { k -> for (i in data.indices) data[i] = (data[i].toInt() xor k[i % 4].toInt()).toByte() }
66 when (val op = b0 and 0x0F) {
67 WS_CLOSE -> {
68 runCatching { writeFrame(output, WS_CLOSE, data, mask) }
69 throw EOFException("WebSocket closed")
70 }
71 WS_PING -> writeFrame(output, WS_PONG, data, mask)
72 WS_PONG -> {}
73 else -> {
74 if (op != 0) opcode = op
75 message.write(data)
76 if (b0 and 0x80 != 0) return opcode to message.toByteArray()
77 }
78 }
79 }
80}
81
82/**
83 * Subscribes to [paths], collections or single elements, and calls [onEvent] with the subscribed path and the event data.
84 * Blocks until the connection ends, then throws.
85 */
86fun Car.listen(paths: List<String>, onEvent: (String, Any?) -> Unit) {
87 // Events can be minutes apart, so this channel has no read timeout.
88 val ch = tunnel.open(443, Long.MAX_VALUE)
89 try {
90 val tls = sessionTls(ch.input, ch.output, psk.first, psk.second)
91 val key = Base64.getEncoder().encodeToString(ByteArray(16).also { SecureRandom().nextBytes(it) })
92 val r = exchange(
93 tls.inputStream, tls.outputStream, "GET", HTTPS_HOST, "/", keepAlive = true,
94 extraHeaders = listOf("Upgrade: websocket", "Connection: Upgrade", "Sec-WebSocket-Key: $key", "Sec-WebSocket-Version: 13"),
95 )
96 if (r.status != 101 || r.headers["sec-websocket-accept"] != wsAccept(key)) throw IOException("WebSocket upgrade failed: HTTP ${r.status}")
97 // Like maps+more: one subscription at a time. The car answers 409 Conflict to overlapping ones.
98 val pending = ArrayDeque(paths)
99 fun subscribeNext() = pending.removeFirstOrNull()?.let {
100 // Inputs need a short limit. Otherwise the car merges "pushed" and "released" into one event.
101 val limit = if (it.startsWith("/mechanicalinput/")) 100 else 1000
102 val subscribe = JSONObject().put("type", "subscribe").put("event", "$it#42").put("updatelimit", limit)
103 writeFrame(tls.outputStream, WS_TEXT, subscribe.toString().toByteArray(), mask = true)
104 }
105 subscribeNext()
106 while (true) {
107 val (op, data) = readMessage(tls.inputStream, tls.outputStream, mask = true)
108 if (op != WS_TEXT) continue
109 val text = data.decodeToString()
110 if (BuildConfig.DEBUG) Log.d("Events", text.take(4000))
111 // The car joins several messages in one frame without a separator.
112 val messages = JSONTokener(text)
113 while (true) {
114 val m = (runCatching { messages.nextValue() }.getOrNull() as? JSONObject) ?: break
115 val line = m.toString()
116 val event = m.optString("event").substringBefore('#').substringBefore('?')
117 when (m.optString("type")) {
118 "data" -> (paths.firstOrNull { event == it } ?: paths.firstOrNull { event.startsWith(it) })?.let { onEvent(it, m.opt("data")) }
119 "subscribe" -> {
120 if (m.optString("status") != "ok") Log.w("Events", "Subscription failed: $line")
121 subscribeNext()
122 }
123 "error" -> {
124 Log.w("Events", "Event error: $line")
125 subscribeNext()
126 }
127 }
128 }
129 }
130 } finally {
131 ch.close()
132 }
133}
134