EARHORNMUSIC UPnP · Android TV · Phones · Android Auto · GNU/Linux
Release 0.5.0

CastStreamProxy.kt

319 lines · 13.7 KB

Changed in: 0.5.0

1package dev.stave.cast
2
3import okhttp3.OkHttpClient
4import okhttp3.Request
5import java.io.BufferedInputStream
6import java.io.IOException
7import java.io.InputStream
8import java.io.OutputStream
9import java.net.DatagramSocket
10import java.net.Inet6Address
11import java.net.InetAddress
12import java.net.ServerSocket
13import java.net.Socket
14import java.util.concurrent.RejectedExecutionException
15import java.util.concurrent.SynchronousQueue
16import java.util.concurrent.ThreadPoolExecutor
17import java.util.concurrent.TimeUnit
18
19/**
20 * Lightweight HTTP proxy between UPnP media servers (MinimServer, etc.)
21 * and Chromecast speaker groups. Group leaders make HTTP Range requests
22 * to synchronize playback across members; when fetching the raw UPnP
23 * URL directly, the server's responses can cause broken audio on group
24 * members. The proxy normalizes HTTP behavior — proper Content-Length,
25 * Accept-Ranges, Content-Range headers on every response — so the group
26 * leader can seek and sync reliably.
27 *
28 * Single speakers don't need this; they handle server quirks internally.
29 * The proxy is only used when [CastIntegration] detects the target is a
30 * speaker group. Same architecture as BubbleUPnP's internal cast-proxy.
31 *
32 * **Exposure.** The socket is reachable from the network for the whole group
33 * session, so (Team 4 #6, #7):
34 * - it listens only on the address this computer uses to reach the Cast
35 * device — never every interface, never a VPN or container bridge the
36 * group can't use — and hands out no `127.0.0.1` links a speaker can't
37 * open. With no such address, [proxyUrl] returns null and the group
38 * fetches from the server directly;
39 * - only links it issued are served (128-bit token, 403 otherwise —
40 * Android parity, 2026-09-04), and at most [MAX_TOKENS] are kept;
41 * - requests are parsed with bounded lines, a header-count cap and a
42 * deadline, on a bounded pool of threads, so a peer can't exhaust memory
43 * or threads.
44 *
45 * Desktop port note: the Android version also had a "local content://"
46 * code path for proxying on-device MediaStore files through to Cast.
47 * That's been removed here — desktop's "local files" feature uses
48 * regular filesystem paths and will need its own (simpler) proxy or
49 * direct-serve when ported. UPnP-source playback is the path that
50 * actually flows through this proxy in normal use.
51 */
52class CastStreamProxy {
53 private val tag = "EarhornProxy"
54
55 private val client = OkHttpClient.Builder()
56 .connectTimeout(10, TimeUnit.SECONDS)
57 .readTimeout(120, TimeUnit.SECONDS)
58 .build()
59
60 private var server: ServerSocket? = null
61 private var boundAddress: InetAddress? = null
62 private var port: Int = 0
63 @Volatile private var running = false
64
65 /** token → upstream URL. Only links this proxy issued are served
66 * (Android parity, 2026-09-04): anything else is 403.
67 * Least-recently-used entries go first once [MAX_TOKENS] are held — a
68 * group fetches only the current track and the next. */
69 private val tokens: MutableMap<String, String> = java.util.Collections.synchronizedMap(
70 object : LinkedHashMap<String, String>(64, 0.75f, true) {
71 override fun removeEldestEntry(eldest: MutableMap.MutableEntry<String, String>?): Boolean =
72 size > MAX_TOKENS
73 }
74 )
75 private val rng = java.security.SecureRandom()
76 private fun newToken(): String {
77 val b = ByteArray(16); rng.nextBytes(b)
78 return b.joinToString("") { "%02x".format(it) }
79 }
80
81 /** Request handlers. Bounded: when all [MAX_HANDLERS] are busy a new
82 * connection is closed at once instead of getting a thread of its own. */
83 private val pool = ThreadPoolExecutor(
84 0, MAX_HANDLERS, 30, TimeUnit.SECONDS, SynchronousQueue(),
85 { r -> Thread(r, "cast-proxy-req").apply { isDaemon = true } },
86 ThreadPoolExecutor.AbortPolicy(),
87 )
88
89 /** Listen on [bindAddr] (a random port). Restarts if it was listening on
90 * another address — the route to the device changed (DHCP, Wi-Fi
91 * switch). Issued links keep working on the new address's port only. */
92 @Synchronized
93 private fun start(bindAddr: InetAddress) {
94 if (running && boundAddress == bindAddr && server?.isClosed == false) return
95 if (running) closeServer()
96 val ss = ServerSocket(0, 50, bindAddr)
97 port = ss.localPort
98 server = ss
99 boundAddress = bindAddr
100 running = true
101 System.err.println("[$tag] Started on ${bindAddr.hostAddress}:$port")
102 Thread(null, {
103 while (running && !ss.isClosed) {
104 try {
105 val socket = ss.accept()
106 try {
107 pool.execute { handleConnection(socket) }
108 } catch (_: RejectedExecutionException) {
109 runCatching { socket.close() }
110 }
111 } catch (e: Exception) {
112 if (running && !ss.isClosed) System.err.println("[$tag] Accept error: ${e.message}")
113 }
114 }
115 }, "cast-proxy-accept").apply { isDaemon = true }.start()
116 }
117
118 @Synchronized
119 private fun closeServer() {
120 running = false
121 runCatching { server?.close() }
122 server = null
123 boundAddress = null
124 }
125
126 @Synchronized
127 fun stop() {
128 closeServer()
129 tokens.clear()
130 }
131
132 /** Wrap a UPnP resource URL so the Cast device at [castHost] fetches from
133 * this proxy instead. Starts the proxy if not already running. Returns
134 * null when this computer has no usable address toward [castHost]: the
135 * caller then hands the device the original URL. */
136 @Synchronized
137 fun proxyUrl(originalUrl: String, castHost: String?): String? {
138 val bindAddr = castHost?.let(::localAddressToward) ?: run {
139 System.err.println("[$tag] No local address toward the Cast device — not proxying")
140 return null
141 }
142 try {
143 start(bindAddr)
144 } catch (e: IOException) {
145 System.err.println("[$tag] Could not listen on ${bindAddr.hostAddress}: ${e.message}")
146 return null
147 }
148 val token = newToken()
149 tokens[token] = originalUrl
150 val host = bindAddr.hostAddress.substringBefore('%').let {
151 if (bindAddr is Inet6Address) "[$it]" else it
152 }
153 return "http://$host:$port/stream?t=$token"
154 }
155
156 /** The local address the kernel would use to reach [host] — the interface
157 * the Cast device can reach us on. Connecting a UDP socket sends no
158 * packet. Null for loopback / wildcard, or when there's no route. */
159 private fun localAddressToward(host: String): InetAddress? = try {
160 DatagramSocket().use { s ->
161 s.connect(InetAddress.getByName(host), 8009)
162 s.localAddress
163 }?.takeIf { !it.isAnyLocalAddress && !it.isLoopbackAddress }
164 } catch (e: Exception) {
165 // UnknownHost / no route (IOException, or UncheckedIOException from
166 // connect on newer JDKs), a bad address, a security manager.
167 System.err.println("[$tag] No route to the Cast device: ${e.message}")
168 null
169 }
170
171 private fun handleConnection(socket: Socket) {
172 try {
173 // Short timeout and an overall deadline while reading the request,
174 // so a peer trickling bytes can't hold a handler for long.
175 socket.soTimeout = HEADER_READ_TIMEOUT_MS
176 socket.sendBufferSize = 262144 // 256KB — matches the read buffer
177 socket.tcpNoDelay = true // disable Nagle — send chunks immediately
178 val input = BufferedInputStream(socket.getInputStream())
179 val output = socket.getOutputStream()
180 val deadline = System.currentTimeMillis() + HEADER_DEADLINE_MS
181
182 val requestLine = readLineBounded(input, deadline) ?: return
183 val parts = requestLine.split(' ')
184 val method = parts.getOrNull(0) ?: "GET"
185 val path = parts.getOrNull(1) ?: ""
186
187 // Read headers (bounded count; every line bounded in length).
188 val headers = mutableMapOf<String, String>()
189 var count = 0
190 var line = readLineBounded(input, deadline)
191 while (line != null && line.isNotBlank()) {
192 if (++count > MAX_HEADERS) {
193 sendError(output, 431, "Request Header Fields Too Large")
194 return
195 }
196 val colon = line.indexOf(':')
197 if (colon > 0) {
198 headers[line.substring(0, colon).trim().lowercase()] =
199 line.substring(colon + 1).trim()
200 }
201 line = readLineBounded(input, deadline)
202 }
203 socket.soTimeout = BODY_TIMEOUT_MS
204
205 val rangeHeader = headers["range"]
206
207 val token = path.substringAfter("t=", "").substringBefore('&').substringBefore(' ')
208 val originalUrl = tokens[token]
209 if (originalUrl == null) {
210 sendError(output, 403, "Forbidden")
211 return
212 }
213
214 // Not the upstream URL: some servers put credentials in the query
215 // string, and this log goes into bug reports (Team 4 #16).
216 System.err.println("[$tag] $method token=${token.take(6)}… range=$rangeHeader")
217
218 val reqBuilder = Request.Builder().url(originalUrl)
219 if (rangeHeader != null) {
220 reqBuilder.header("Range", rangeHeader)
221 }
222 if (method == "HEAD") {
223 reqBuilder.head()
224 }
225
226 // `use`: the upstream response is closed however this ends —
227 // a group leader dropping a Range probe mid-headers is common.
228 client.newCall(reqBuilder.build()).execute().use { response ->
229 val body = response.body
230 val contentType = response.header("Content-Type") ?: "application/octet-stream"
231 val contentLength = response.header("Content-Length")
232 val contentRange = response.header("Content-Range")
233
234 val statusCode = response.code
235 val statusText = when (statusCode) {
236 200 -> "OK"
237 206 -> "Partial Content"
238 404 -> "Not Found"
239 416 -> "Range Not Satisfiable"
240 else -> response.message.filter { it >= ' ' }.ifBlank { "Error" }
241 }
242
243 val sb = StringBuilder()
244 sb.append("HTTP/1.1 $statusCode $statusText\r\n")
245 sb.append("Content-Type: $contentType\r\n")
246 sb.append("Accept-Ranges: bytes\r\n")
247 sb.append("Access-Control-Allow-Origin: *\r\n")
248 if (contentLength != null) {
249 sb.append("Content-Length: $contentLength\r\n")
250 }
251 if (contentRange != null) {
252 sb.append("Content-Range: $contentRange\r\n")
253 }
254 sb.append("Connection: close\r\n")
255 sb.append("\r\n")
256
257 output.write(sb.toString().toByteArray())
258 output.flush()
259
260 if (method != "HEAD" && body != null) {
261 val buf = ByteArray(32768)
262 try {
263 body.byteStream().use { bodyStream ->
264 var read: Int
265 while (bodyStream.read(buf).also { read = it } != -1) {
266 output.write(buf, 0, read)
267 }
268 }
269 output.flush()
270 } catch (e: IOException) {
271 // Client disconnected — normal during seek
272 }
273 }
274 }
275 } catch (e: Exception) {
276 System.err.println("[$tag] Connection error: ${e.message}")
277 } finally {
278 runCatching { socket.close() }
279 }
280 }
281
282 /** One CRLF/LF-terminated header line, at most [MAX_LINE] bytes, before
283 * [deadline]. Null at end of stream with nothing read. */
284 private fun readLineBounded(input: InputStream, deadline: Long): String? {
285 val sb = StringBuilder()
286 while (true) {
287 if (System.currentTimeMillis() > deadline) throw IOException("request too slow")
288 val c = input.read()
289 if (c < 0) return if (sb.isEmpty()) null else sb.toString()
290 if (c == '\n'.code) return sb.toString().trimEnd('\r')
291 if (sb.length >= MAX_LINE) throw IOException("request line too long")
292 sb.append(c.toChar())
293 }
294 }
295
296 private fun sendError(output: OutputStream, code: Int, message: String) {
297 val body = message.toByteArray()
298 val response = "HTTP/1.1 $code $message\r\n" +
299 "Content-Type: text/plain\r\n" +
300 "Content-Length: ${body.size}\r\n" +
301 "Connection: close\r\n" +
302 "\r\n"
303 output.write(response.toByteArray())
304 output.write(body)
305 output.flush()
306 }
307
308 private companion object {
309 /** Concurrent requests served; a group leader uses a handful. */
310 const val MAX_HANDLERS = 16
311 /** Issued links kept (least recently used dropped first). */
312 const val MAX_TOKENS = 512
313 const val MAX_LINE = 8192
314 const val MAX_HEADERS = 64
315 const val HEADER_READ_TIMEOUT_MS = 10_000
316 const val HEADER_DEADLINE_MS = 15_000L
317 const val BODY_TIMEOUT_MS = 30_000
318 }
319}