Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 mrambossekandClaude Fable 5
|
7a94c9a3d7 | ||
|
mrambossekandClaude Fable 5
|
2521d39989 |
@@ -40,3 +40,6 @@ keystore.properties
|
||||
|
||||
# wrangler build/dev artifacts
|
||||
web/.wrangler/
|
||||
|
||||
# Eclipse/JDT output from the VSCodium Java extension — not a build artifact we own
|
||||
echolot-app/*/bin/
|
||||
|
||||
@@ -1,166 +0,0 @@
|
||||
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
// Package engine composes core-protocol probes into core-measurement documents — the run engine
|
||||
// the app drives. This file covers the server-facing vertical (control plane + UDP data plane);
|
||||
// device-tier probes (link snapshot, Shizuku, local discovery) plug in from the Android modules.
|
||||
package app.echo_lot.engine
|
||||
|
||||
import app.echo_lot.measurement.*
|
||||
import app.echo_lot.protocol.ControlClient
|
||||
import app.echo_lot.protocol.ProbeSession
|
||||
import kotlinx.serialization.json.Json
|
||||
import kotlinx.serialization.json.JsonObject
|
||||
import kotlinx.serialization.json.encodeToJsonElement
|
||||
|
||||
/**
|
||||
* Runs the server-facing measurements against one target and assembles a [MeasurementDocument]:
|
||||
* an ECHO train (RTT distribution, loss, and NAT-rebinding detection from the server's observed
|
||||
* source port) as `train.udp_updown`. Everything is real evidence with recomputable metrics, and
|
||||
* findings are derived deterministically. IDs/timestamps are injected so the engine stays pure
|
||||
* (no clocks/UUIDs of its own) and unit-testable.
|
||||
*/
|
||||
class ServerMeasurement(
|
||||
private val ids: IdSource,
|
||||
private val app: AppInfo,
|
||||
private val device: DeviceInfo,
|
||||
) {
|
||||
private val json = Json { encodeDefaults = true; explicitNulls = true }
|
||||
|
||||
data class Config(
|
||||
val controlUrl: String,
|
||||
val pins: Set<String>,
|
||||
val credential: String,
|
||||
val target: String,
|
||||
val udpHost: String,
|
||||
val udpPort: Int,
|
||||
val echoCount: Int = 20,
|
||||
val echoPaddingBytes: Int = 64,
|
||||
)
|
||||
|
||||
fun run(cfg: Config): MeasurementDocument {
|
||||
val runId = ids.uuid()
|
||||
val startWall = ids.nowWall()
|
||||
val startMono = ids.monoNs()
|
||||
|
||||
val control = ControlClient(cfg.controlUrl, cfg.pins)
|
||||
val profile = control.profile(cfg.credential)
|
||||
val session = control.createSession(cfg.credential, cfg.target)
|
||||
|
||||
val serverSession = ServerSession(
|
||||
id = "sess-1",
|
||||
profileName = profile.name,
|
||||
controlUrl = cfg.controlUrl,
|
||||
serverVersion = profile.serverVersion,
|
||||
capabilities = profile.capabilities,
|
||||
sessionId = session.sessionId,
|
||||
target = SessionTarget(ip4 = cfg.udpHost, udpPort = cfg.udpPort),
|
||||
)
|
||||
|
||||
val (test, findings) = echoTrain(cfg, control, session, startMono)
|
||||
|
||||
control.deleteSession(cfg.credential, session.sessionId)
|
||||
|
||||
val summary = Verdicts.derive(listOf(test), findings)
|
||||
return MeasurementDocument(
|
||||
run = Run(
|
||||
id = runId, trigger = Trigger.MANUAL, startedAt = startWall, endedAt = ids.nowWall(),
|
||||
clock = Clock(monoOriginWall = startWall),
|
||||
app = app, device = device,
|
||||
tiers = Tiers(app = true),
|
||||
),
|
||||
serverSessions = listOf(serverSession),
|
||||
tests = listOf(test),
|
||||
findings = findings,
|
||||
summary = summary,
|
||||
)
|
||||
}
|
||||
|
||||
private fun echoTrain(
|
||||
cfg: Config, control: ControlClient, session: app.echo_lot.protocol.SessionResponse, startMono: Long,
|
||||
): Pair<Test, List<Finding>> {
|
||||
val testId = ids.uuid()
|
||||
val seqs = ArrayList<Int>()
|
||||
val tTx = ArrayList<Long?>()
|
||||
val tRx = ArrayList<Long?>()
|
||||
val sizes = ArrayList<Int>()
|
||||
val rtts = ArrayList<Double>()
|
||||
val observedPorts = LinkedHashSet<Int>()
|
||||
|
||||
ProbeSession(cfg.credential, session, cfg.udpHost, cfg.udpPort).use { ps ->
|
||||
for (i in 0 until cfg.echoCount) {
|
||||
val txMono = ids.monoNs() - startMono
|
||||
val r = ps.echo(cfg.echoPaddingBytes)
|
||||
seqs.add(i)
|
||||
tTx.add(txMono)
|
||||
sizes.add(Wire_HEADER + cfg.echoPaddingBytes)
|
||||
if (r != null) {
|
||||
tRx.add(ids.monoNs() - startMono)
|
||||
rtts.add(r.rttMs)
|
||||
r.observation?.observedPort?.let { observedPorts.add(it) }
|
||||
} else {
|
||||
tRx.add(null)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
val sent = cfg.echoCount
|
||||
val received = rtts.size
|
||||
val lossPct = if (sent == 0) 0.0 else (sent - received) * 100.0 / sent
|
||||
val natRebinding = observedPorts.size > 1
|
||||
|
||||
val evidence: JsonObject = TrainEvidence(
|
||||
epochMonoNs = startMono, seq = seqs, tTxNs = tTx, tRxNs = tRx, sizeBytes = sizes,
|
||||
).toEvidence()
|
||||
|
||||
val metrics: JsonObject = json.encodeToJsonElement(
|
||||
EchoMetrics(
|
||||
sent = sent, received = received, lossPct = round1(lossPct),
|
||||
rttMsMin = rtts.minOrNull()?.let(::round1),
|
||||
rttMsAvg = rtts.average().takeIf { received > 0 }?.let(::round1),
|
||||
rttMsMax = rtts.maxOrNull()?.let(::round1),
|
||||
observedPorts = observedPorts.toList(),
|
||||
natRebindingDetected = natRebinding,
|
||||
)
|
||||
) as JsonObject
|
||||
|
||||
val status = when {
|
||||
received == 0 -> TestStatus.FAILED
|
||||
received < sent -> TestStatus.PARTIAL
|
||||
else -> TestStatus.OK
|
||||
}
|
||||
val test = Test(
|
||||
id = testId, type = TestType.TRAIN_UDP_UPDOWN, sessionRef = "sess-1", tier = Tier.APP,
|
||||
startedMonoNs = startMono, endedMonoNs = ids.monoNs(), status = status,
|
||||
evidence = evidence, metrics = metrics,
|
||||
)
|
||||
|
||||
val findings = ArrayList<Finding>()
|
||||
if (received == 0) {
|
||||
findings.add(finding("nat.udp_unreachable", Category.CONNECTIVITY, Severity.HIGH, testId,
|
||||
"No UDP echo replies from the server",
|
||||
"Every ECHO probe to the server's UDP data plane was lost — the path blocks or drops the session's UDP traffic."))
|
||||
} else if (lossPct >= 20.0) {
|
||||
findings.add(finding("connectivity.udp_loss", Category.CONNECTIVITY, Severity.MEDIUM, testId,
|
||||
"High UDP loss to the server (${round1(lossPct)}%)",
|
||||
"A large fraction of ECHO probes were lost, indicating an unreliable UDP path."))
|
||||
}
|
||||
if (natRebinding) {
|
||||
findings.add(finding("nat.udp_rebinding", Category.NAT, Severity.MEDIUM, testId,
|
||||
"NAT remapped the UDP source port mid-flow",
|
||||
"The server observed more than one source port for this session (${observedPorts.joinToString()}), i.e. a NAT with a short UDP mapping or per-packet remapping."))
|
||||
}
|
||||
return test to findings
|
||||
}
|
||||
|
||||
private fun finding(code: String, cat: Category, sev: Severity, testId: String, title: String, desc: String) =
|
||||
Finding(
|
||||
id = ids.uuid(), code = code, category = cat, severity = sev, confidence = Confidence.HIGH,
|
||||
title = title, description = desc, evidenceRefs = listOf(EvidenceRef(testId)),
|
||||
)
|
||||
|
||||
private companion object {
|
||||
const val Wire_HEADER = 32
|
||||
fun round1(v: Double) = Math.round(v * 10.0) / 10.0
|
||||
}
|
||||
}
|
||||
@@ -1,38 +0,0 @@
|
||||
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
package app.echo_lot.engine
|
||||
|
||||
import kotlinx.serialization.SerialName
|
||||
import kotlinx.serialization.Serializable
|
||||
import java.time.Instant
|
||||
import java.util.UUID
|
||||
|
||||
/**
|
||||
* Clock/ID source, injected so the engine has no hidden nondeterminism and stays unit-testable.
|
||||
* The default uses wall + monotonic clocks and random UUIDs; tests supply deterministic ones.
|
||||
*/
|
||||
interface IdSource {
|
||||
fun uuid(): String
|
||||
fun monoNs(): Long
|
||||
fun nowWall(): String
|
||||
}
|
||||
|
||||
class SystemIdSource : IdSource {
|
||||
override fun uuid(): String = UUID.randomUUID().toString()
|
||||
override fun monoNs(): Long = System.nanoTime()
|
||||
override fun nowWall(): String = Instant.now().toString()
|
||||
}
|
||||
|
||||
/** Metrics for train.udp_updown; recomputable from the columnar evidence. */
|
||||
@Serializable
|
||||
data class EchoMetrics(
|
||||
val sent: Int,
|
||||
val received: Int,
|
||||
@SerialName("loss_pct") val lossPct: Double,
|
||||
@SerialName("rtt_ms_min") val rttMsMin: Double? = null,
|
||||
@SerialName("rtt_ms_avg") val rttMsAvg: Double? = null,
|
||||
@SerialName("rtt_ms_max") val rttMsMax: Double? = null,
|
||||
@SerialName("observed_ports") val observedPorts: List<Int> = emptyList(),
|
||||
@SerialName("nat_rebinding_detected") val natRebindingDetected: Boolean = false,
|
||||
)
|
||||
@@ -1,63 +0,0 @@
|
||||
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
package app.echo_lot.engine
|
||||
|
||||
import app.echo_lot.measurement.*
|
||||
import kotlinx.serialization.json.Json
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
/**
|
||||
* Runs the full server-facing engine against a live server and validates the produced
|
||||
* MeasurementDocument. Self-skips without ECHOLOT_LIVE_* (same contract as core-protocol's live
|
||||
* test). This is the whole vertical: protocol client → engine → schema document → verdict.
|
||||
*/
|
||||
class LiveMeasurementTest {
|
||||
|
||||
private val url = System.getenv("ECHOLOT_LIVE_URL")
|
||||
private val pin = System.getenv("ECHOLOT_LIVE_PIN")
|
||||
private val cred = System.getenv("ECHOLOT_LIVE_CRED")
|
||||
private val udp = System.getenv("ECHOLOT_LIVE_UDP")
|
||||
private val target = System.getenv("ECHOLOT_LIVE_TARGET") ?: "fmr"
|
||||
|
||||
@Test
|
||||
fun producesValidDocumentFromLiveServer() {
|
||||
if (url == null || pin == null || cred == null || udp == null) {
|
||||
println("LiveMeasurementTest skipped (no ECHOLOT_LIVE_* env)")
|
||||
return
|
||||
}
|
||||
val (host, port) = udp.split(":").let { it[0] to it[1].toInt() }
|
||||
val engine = ServerMeasurement(
|
||||
ids = SystemIdSource(),
|
||||
app = AppInfo(version = "0.1.0", build = 1, flavor = "test"),
|
||||
device = DeviceInfo("test", "jvm", 0, "n/a"),
|
||||
)
|
||||
val doc = engine.run(
|
||||
ServerMeasurement.Config(
|
||||
controlUrl = url, pins = setOf(pin), credential = cred,
|
||||
target = target, udpHost = host, udpPort = port, echoCount = 20,
|
||||
)
|
||||
)
|
||||
|
||||
// The document must round-trip and carry the expected structure.
|
||||
val encoded = Json { encodeDefaults = true }.encodeToString(MeasurementDocument.serializer(), doc)
|
||||
println("document (${encoded.length} bytes): overall=${doc.summary?.overall}")
|
||||
|
||||
assertEquals(1, doc.serverSessions.size)
|
||||
assertTrue(doc.serverSessions[0].capabilities.contains("udp-probe"))
|
||||
val test = doc.tests.single()
|
||||
assertEquals(TestType.TRAIN_UDP_UPDOWN, test.type)
|
||||
assertTrue(test.status == TestStatus.OK || test.status == TestStatus.PARTIAL,
|
||||
"expected replies from live server, got ${test.status}")
|
||||
|
||||
val metrics = Json.parseToJsonElement(test.metrics.toString())
|
||||
println("metrics: $metrics")
|
||||
assertTrue(metrics.toString().contains("rtt_ms_avg"))
|
||||
|
||||
assertTrue(doc.summary != null)
|
||||
// A healthy local->fmr path should be green (no loss, no rebinding) or yellow.
|
||||
println("summary: ${doc.summary}")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,73 @@
|
||||
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
package app.echo_lot.engine
|
||||
|
||||
import app.echo_lot.protocol.ControlClient
|
||||
import app.echo_lot.protocol.ProbeSession
|
||||
import app.echo_lot.protocol.Wire
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
/**
|
||||
* Exercises the server's §5 granted sends against a LIVE server: downtrain (downstream loss /
|
||||
* ordering) and big_send (downstream MTU). Self-skips without ECHOLOT_LIVE_*.
|
||||
*
|
||||
* This is the direction a client cannot measure alone — only the far end can push large or
|
||||
* numerous packets toward it — so it is also the direction that needs the anti-amplification
|
||||
* grant, and this test is the proof that the grant path works end to end.
|
||||
*/
|
||||
class LiveGrantedTest {
|
||||
|
||||
private val url = System.getenv("ECHOLOT_LIVE_URL")
|
||||
private val pin = System.getenv("ECHOLOT_LIVE_PIN")
|
||||
private val cred = System.getenv("ECHOLOT_LIVE_CRED")
|
||||
private val udp = System.getenv("ECHOLOT_LIVE_UDP")
|
||||
private val target = System.getenv("ECHOLOT_LIVE_TARGET") ?: "fmr"
|
||||
|
||||
@Test
|
||||
fun downstreamTrainAndBigSend() {
|
||||
if (url == null || pin == null || cred == null || udp == null) {
|
||||
println("LiveGrantedTest skipped (no ECHOLOT_LIVE_* env)"); return
|
||||
}
|
||||
val control = ControlClient(url, setOf(pin))
|
||||
val profile = control.profile(cred)
|
||||
println("capabilities: ${profile.capabilities}")
|
||||
val session = control.createSession(cred, target)
|
||||
val (host, port) = udp.split(":").let { it[0] to it[1].toInt() }
|
||||
|
||||
ProbeSession(cred, session, host, port).use { ps ->
|
||||
// The grant is bound to the OBSERVED data-plane source, so we must be seen first.
|
||||
val echo = ps.echo()
|
||||
println("primed with echo rtt=${echo?.rttMs}")
|
||||
|
||||
// --- downtrain: 50 packets of 300 bytes, 5ms apart ---
|
||||
val dtResp = control.action(
|
||||
cred, session.sessionId,
|
||||
"""{"action":"downtrain","count":50,"size_bytes":300,"interval_us":5000}""",
|
||||
)
|
||||
println("downtrain accepted: ${dtResp.take(160)}")
|
||||
val down = ps.collectGranted(windowMs = 4000)
|
||||
.filter { it.type == Wire.TYPE_DOWNTRAIN_DATA }
|
||||
val seqs = down.map { it.seq }.toSet()
|
||||
println("downtrain received ${down.size}/50 packets, distinct seqs=${seqs.size}, " +
|
||||
"sizes=${down.map { it.sizeBytes }.distinct()}")
|
||||
assertTrue(down.isNotEmpty(), "no DOWNTRAIN_DATA arrived — granted send path is broken")
|
||||
|
||||
// --- big_send: which downstream sizes survive? ---
|
||||
val sizes = listOf(600, 1200, 1400, 1472, 1500, 2000, 4000)
|
||||
val bsResp = control.action(
|
||||
cred, session.sessionId,
|
||||
"""{"action":"big_send","sizes_bytes":${sizes}}""",
|
||||
)
|
||||
println("big_send accepted: ${bsResp.take(160)}")
|
||||
val big = ps.collectGranted(windowMs = 4000)
|
||||
.filter { it.type == Wire.TYPE_BIG_SEND }
|
||||
val arrived = big.map { it.sizeBytes }.sorted()
|
||||
println("big_send arrived sizes: $arrived (requested $sizes)")
|
||||
assertTrue(big.isNotEmpty(), "no BIG_SEND packets arrived")
|
||||
println("largest downstream datagram delivered: ${arrived.maxOrNull()}")
|
||||
}
|
||||
control.deleteSession(cred, session.sessionId)
|
||||
}
|
||||
}
|
||||
@@ -1,92 +0,0 @@
|
||||
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
package app.echo_lot.protocol
|
||||
|
||||
import kotlinx.serialization.json.Json
|
||||
import java.net.URL
|
||||
import javax.net.ssl.HttpsURLConnection
|
||||
|
||||
/**
|
||||
* The control-plane client (probe-protocol.md §2): enrollment, profile, sessions — over
|
||||
* SPKI-pinned HTTPS. Uses HttpsURLConnection (available since Android API 1, unlike
|
||||
* java.net.http.HttpClient which needs API 34) with a pin-based SSLSocketFactory and hostname
|
||||
* verification DISABLED: trust is the SPKI pin, never the certificate name (self-signed servers
|
||||
* with no SAN are first-class). Blocking; the Android layer wraps calls in coroutines.
|
||||
*
|
||||
* @param controlUrl e.g. "https://fmr-1.echo-lot.app:8443"
|
||||
* @param pins the `pin-sha256` value(s) from the enrollment QR (base64, no prefix)
|
||||
*/
|
||||
class ControlClient(private val controlUrl: String, pins: Set<String>) {
|
||||
|
||||
private val json = Json { ignoreUnknownKeys = true }
|
||||
private val socketFactory = Pinning.sslContext(pins).socketFactory
|
||||
|
||||
private fun open(path: String, method: String, credential: String?): HttpsURLConnection {
|
||||
val conn = URL(controlUrl.trimEnd('/') + path).openConnection() as HttpsURLConnection
|
||||
conn.sslSocketFactory = socketFactory
|
||||
conn.setHostnameVerifier { _, _ -> true } // pin is the trust, not the name
|
||||
conn.requestMethod = method
|
||||
conn.connectTimeout = 10_000
|
||||
conn.readTimeout = 10_000
|
||||
credential?.let { conn.setRequestProperty("Authorization", "Bearer $it") }
|
||||
return conn
|
||||
}
|
||||
|
||||
private fun body(conn: HttpsURLConnection): String {
|
||||
val stream = if (conn.responseCode in 200..299) conn.inputStream else conn.errorStream
|
||||
return stream?.bufferedReader()?.use { it.readText() } ?: ""
|
||||
}
|
||||
|
||||
private fun writeJson(conn: HttpsURLConnection, payload: String) {
|
||||
conn.doOutput = true
|
||||
conn.setRequestProperty("Content-Type", "application/json")
|
||||
conn.outputStream.use { it.write(payload.toByteArray()) }
|
||||
}
|
||||
|
||||
// Minimal JSON string literal (the only bodies we send are one short field).
|
||||
private fun jstr(s: String): String {
|
||||
val sb = StringBuilder("\"")
|
||||
for (c in s) when (c) {
|
||||
'"' -> sb.append("\\\"")
|
||||
'\\' -> sb.append("\\\\")
|
||||
'\n' -> sb.append("\\n")
|
||||
'\r' -> sb.append("\\r")
|
||||
'\t' -> sb.append("\\t")
|
||||
else -> sb.append(c)
|
||||
}
|
||||
return sb.append('"').toString()
|
||||
}
|
||||
|
||||
/** Redeem a single-use enrollment token for a device credential (§2.1). */
|
||||
fun enroll(token: String, name: String? = null): EnrollResponse {
|
||||
val conn = open("/v1/enroll", "POST", null)
|
||||
conn.setRequestProperty("Authorization", "Bearer $token")
|
||||
writeJson(conn, if (name != null) """{"name":${jstr(name)}}""" else "{}")
|
||||
check(conn.responseCode == 201) { "enroll failed: ${conn.responseCode} ${body(conn)}" }
|
||||
return json.decodeFromString(EnrollResponse.serializer(), body(conn))
|
||||
}
|
||||
|
||||
fun profile(credential: String): Profile {
|
||||
val conn = open("/v1/profile", "GET", credential)
|
||||
check(conn.responseCode == 200) { "profile failed: ${conn.responseCode} ${body(conn)}" }
|
||||
return json.decodeFromString(Profile.serializer(), body(conn))
|
||||
}
|
||||
|
||||
fun createSession(credential: String, target: String): SessionResponse {
|
||||
val conn = open("/v1/sessions", "POST", credential)
|
||||
writeJson(conn, """{"target":${jstr(target)}}""")
|
||||
check(conn.responseCode == 201) { "session failed: ${conn.responseCode} ${body(conn)}" }
|
||||
return json.decodeFromString(SessionResponse.serializer(), body(conn))
|
||||
}
|
||||
|
||||
fun observations(credential: String, sessionId: String): String {
|
||||
val conn = open("/v1/sessions/$sessionId/observations", "GET", credential)
|
||||
check(conn.responseCode == 200) { "observations failed: ${conn.responseCode}" }
|
||||
return body(conn)
|
||||
}
|
||||
|
||||
fun deleteSession(credential: String, sessionId: String) {
|
||||
open("/v1/sessions/$sessionId", "DELETE", credential).responseCode
|
||||
}
|
||||
}
|
||||
@@ -1,53 +0,0 @@
|
||||
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
package app.echo_lot.protocol
|
||||
|
||||
import javax.crypto.Mac
|
||||
import javax.crypto.spec.SecretKeySpec
|
||||
|
||||
/**
|
||||
* The protocol crypto primitives, matching the server exactly (probe-protocol.md §2.4/§3.1):
|
||||
* HMAC-SHA256 for the data-plane gate, and HKDF-SHA256 for the session key
|
||||
* `HKDF(ikm = device_credential, salt = key_salt, info = "echolot-v1/" + session_id)`.
|
||||
* JDK-only (javax.crypto) — no third-party crypto.
|
||||
*/
|
||||
object Crypto {
|
||||
|
||||
fun hmacSha256(key: ByteArray, data: ByteArray): ByteArray =
|
||||
Mac.getInstance("HmacSHA256").run {
|
||||
init(SecretKeySpec(key, "HmacSHA256"))
|
||||
doFinal(data)
|
||||
}
|
||||
|
||||
/** First 4 bytes of HMAC-SHA256 — the wire anti-abuse gate (spec §3.1). */
|
||||
fun hmac32(key: ByteArray, data: ByteArray): ByteArray = hmacSha256(key, data).copyOf(4)
|
||||
|
||||
/**
|
||||
* HKDF-SHA256 (RFC 5869) extract-then-expand. The JDK exposes no HKDF, so it is built from
|
||||
* HMAC — small and standard.
|
||||
*/
|
||||
fun hkdfSha256(ikm: ByteArray, salt: ByteArray, info: ByteArray, length: Int): ByteArray {
|
||||
val prk = hmacSha256(if (salt.isEmpty()) ByteArray(32) else salt, ikm) // extract
|
||||
val out = ByteArray(length)
|
||||
var t = ByteArray(0)
|
||||
var pos = 0
|
||||
var counter = 1
|
||||
while (pos < length) {
|
||||
val mac = Mac.getInstance("HmacSHA256").apply { init(SecretKeySpec(prk, "HmacSHA256")) }
|
||||
mac.update(t)
|
||||
mac.update(info)
|
||||
mac.update(counter.toByte())
|
||||
t = mac.doFinal()
|
||||
val n = minOf(t.size, length - pos)
|
||||
t.copyInto(out, pos, 0, n)
|
||||
pos += n
|
||||
counter++
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
/** Derives the 32-byte session key for a session (spec §2.4). */
|
||||
fun sessionKey(credential: String, keySalt: ByteArray, sessionId: String): ByteArray =
|
||||
hkdfSha256(credential.toByteArray(), keySalt, "echolot-v1/$sessionId".toByteArray(), 32)
|
||||
}
|
||||
@@ -1,64 +0,0 @@
|
||||
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
package app.echo_lot.protocol
|
||||
|
||||
import kotlinx.serialization.SerialName
|
||||
import kotlinx.serialization.Serializable
|
||||
import kotlinx.serialization.json.JsonElement
|
||||
|
||||
/** Control-plane JSON shapes (probe-protocol.md §2). Only fields the client uses are modeled;
|
||||
* unknown fields are ignored by the lenient Json in [ControlClient]. */
|
||||
|
||||
@Serializable
|
||||
data class EnrollResponse(
|
||||
@SerialName("device_id") val deviceId: String,
|
||||
val credential: String,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class Target(
|
||||
val id: String,
|
||||
val ip4: String? = null,
|
||||
val ip6: String? = null,
|
||||
@SerialName("udp_port") val udpPort: Int = 0,
|
||||
@SerialName("tcp_port") val tcpPort: Int = 0,
|
||||
@SerialName("stun_port") val stunPort: Int = 0,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class SelfTest(
|
||||
@SerialName("mtu_ok") val mtuOk: Boolean? = null,
|
||||
@SerialName("sysctl_ok") val sysctlOk: Boolean? = null,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class Profile(
|
||||
@SerialName("profile_version") val profileVersion: Int = 0,
|
||||
val name: String = "",
|
||||
@SerialName("server_version") val serverVersion: String = "",
|
||||
val capabilities: List<String> = emptyList(),
|
||||
val targets: List<Target> = emptyList(),
|
||||
@SerialName("canary_zone") val canaryZone: String = "",
|
||||
@SerialName("server_selftest") val serverSelftest: SelfTest? = null,
|
||||
val pins: List<String> = emptyList(),
|
||||
) {
|
||||
fun supports(capability: String) = capability in capabilities
|
||||
}
|
||||
|
||||
@Serializable
|
||||
data class SessionResponse(
|
||||
@SerialName("session_id") val sessionId: String,
|
||||
@SerialName("key_salt") val keySalt: String, // base64
|
||||
val epoch: String,
|
||||
@SerialName("expires_s") val expiresS: Int,
|
||||
)
|
||||
|
||||
/** Observations bundle (§6). Kept as raw JSON where the shape is still evolving server-side. */
|
||||
@Serializable
|
||||
data class Observations(
|
||||
val udp: JsonElement? = null,
|
||||
val tcp: JsonElement? = null,
|
||||
@SerialName("connect_back") val connectBack: JsonElement? = null,
|
||||
@SerialName("dns_canary") val dnsCanary: JsonElement? = null,
|
||||
)
|
||||
@@ -1,39 +0,0 @@
|
||||
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
package app.echo_lot.protocol
|
||||
|
||||
import java.security.MessageDigest
|
||||
import java.security.cert.X509Certificate
|
||||
import javax.net.ssl.SSLContext
|
||||
import javax.net.ssl.X509TrustManager
|
||||
|
||||
/**
|
||||
* SPKI-pinned trust (probe-protocol.md §1): the client trusts the server ONLY against the
|
||||
* `pin-sha256` from enrollment — CA validation is not required and self-signed is first-class.
|
||||
* The pin is base64(SHA-256(SubjectPublicKeyInfo)), RFC 7469.
|
||||
*/
|
||||
object Pinning {
|
||||
|
||||
fun spkiPin(cert: X509Certificate): String {
|
||||
val spki = cert.publicKey.encoded // DER SubjectPublicKeyInfo
|
||||
val digest = MessageDigest.getInstance("SHA-256").digest(spki)
|
||||
return java.util.Base64.getEncoder().encodeToString(digest)
|
||||
}
|
||||
|
||||
/** An SSLContext that accepts a chain iff its leaf SPKI matches one of the expected pins. */
|
||||
fun sslContext(expectedPins: Set<String>): SSLContext {
|
||||
val tm = object : X509TrustManager {
|
||||
override fun checkServerTrusted(chain: Array<out X509Certificate>, authType: String) {
|
||||
val leaf = chain.firstOrNull() ?: throw java.security.cert.CertificateException("empty chain")
|
||||
val pin = spkiPin(leaf)
|
||||
if (pin !in expectedPins) {
|
||||
throw java.security.cert.CertificateException("SPKI pin mismatch: got $pin")
|
||||
}
|
||||
}
|
||||
override fun checkClientTrusted(chain: Array<out X509Certificate>, authType: String) = Unit
|
||||
override fun getAcceptedIssuers(): Array<X509Certificate> = emptyArray()
|
||||
}
|
||||
return SSLContext.getInstance("TLS").apply { init(null, arrayOf(tm), null) }
|
||||
}
|
||||
}
|
||||
@@ -1,77 +0,0 @@
|
||||
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
package app.echo_lot.protocol
|
||||
|
||||
import java.net.DatagramPacket
|
||||
import java.net.DatagramSocket
|
||||
import java.net.InetSocketAddress
|
||||
import java.util.Base64
|
||||
|
||||
/**
|
||||
* A data-plane session (probe-protocol.md §3): derives the session key, then sends signed ELT1
|
||||
* packets to the server's UDP endpoint and reads back verified responses. One session ↔ one
|
||||
* server target. Blocking; the caller owns threading.
|
||||
*/
|
||||
class ProbeSession(
|
||||
private val credential: String,
|
||||
private val session: SessionResponse,
|
||||
private val serverHost: String,
|
||||
private val serverUdpPort: Int,
|
||||
) : AutoCloseable {
|
||||
|
||||
private val key: ByteArray =
|
||||
Crypto.sessionKey(credential, Base64.getDecoder().decode(session.keySalt), session.sessionId)
|
||||
private val prefix: ByteArray = Wire.wirePrefix(session.sessionId)
|
||||
private val epochNanos = System.nanoTime()
|
||||
private val socket = DatagramSocket().apply { soTimeout = 3000 }
|
||||
private val server = InetSocketAddress(serverHost, serverUdpPort)
|
||||
private var seq = 0
|
||||
|
||||
private fun nowNs() = System.nanoTime() - epochNanos
|
||||
|
||||
/**
|
||||
* One ECHO round trip. Returns RTT in ms and the server's observation, or null on loss.
|
||||
*
|
||||
* The response is capped at the request size (§3.4 anti-amplification) and the observation
|
||||
* block is 40 bytes, so the request must be at least header+40 = 72 bytes for the full
|
||||
* observation to fit — hence the ≥40 default padding. Smaller requests still measure RTT.
|
||||
*/
|
||||
fun echo(paddingBytes: Int = 40): EchoResult? {
|
||||
val t0 = System.nanoTime()
|
||||
val pkt = Wire.build(Wire.TYPE_ECHO_REQ, prefix, ++seq, nowNs(), key, ByteArray(paddingBytes))
|
||||
socket.send(DatagramPacket(pkt, pkt.size, server))
|
||||
val resp = receive(Wire.TYPE_ECHO_RESP) ?: return null
|
||||
val rttMs = (System.nanoTime() - t0) / 1_000_000.0
|
||||
return EchoResult(rttMs, Observation.parse(resp.payload))
|
||||
}
|
||||
|
||||
/** One MTU probe of [totalSize] bytes (DF is set by the OS on the socket where supported).
|
||||
* Returns the size the server acknowledged receiving, or null if the probe was lost. */
|
||||
fun mtuProbe(totalSize: Int): Int? {
|
||||
val payloadLen = (totalSize - Wire.HEADER_SIZE).coerceAtLeast(0)
|
||||
val pkt = Wire.build(Wire.TYPE_MTU_PROBE, prefix, ++seq, nowNs(), key, ByteArray(payloadLen))
|
||||
socket.send(DatagramPacket(pkt, pkt.size, server))
|
||||
val resp = receive(Wire.TYPE_MTU_ACK) ?: return null
|
||||
if (resp.payload.size < 4) return null
|
||||
return ((resp.payload[0].toInt() and 0xFF) shl 24) or
|
||||
((resp.payload[1].toInt() and 0xFF) shl 16) or
|
||||
((resp.payload[2].toInt() and 0xFF) shl 8) or
|
||||
(resp.payload[3].toInt() and 0xFF)
|
||||
}
|
||||
|
||||
private fun receive(wantType: Int): Wire.Packet? {
|
||||
val buf = ByteArray(2048)
|
||||
return try {
|
||||
val dp = DatagramPacket(buf, buf.size)
|
||||
socket.receive(dp)
|
||||
Wire.parseVerified(buf, dp.length, key)?.takeIf { it.type == wantType }
|
||||
} catch (e: java.net.SocketTimeoutException) {
|
||||
null
|
||||
}
|
||||
}
|
||||
|
||||
override fun close() = socket.close()
|
||||
|
||||
data class EchoResult(val rttMs: Double, val observation: Observation?)
|
||||
}
|
||||
@@ -1,115 +0,0 @@
|
||||
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
package app.echo_lot.protocol
|
||||
|
||||
import java.nio.ByteBuffer
|
||||
import java.nio.ByteOrder
|
||||
|
||||
/**
|
||||
* The binary UDP probe protocol wire format (probe-protocol.md §3.1): a fixed 32-byte header
|
||||
* plus payload, HMAC-gated. Mirrors the Go server's dataplane package byte-for-byte.
|
||||
*
|
||||
* ```
|
||||
* 0 4 magic "ELT1" 8 8 session_prefix (first 8 bytes of session id)
|
||||
* 4 1 type 16 4 seq
|
||||
* 5 1 flags 20 8 t_ns (sender clock, ns since session epoch)
|
||||
* 6 2 payload_len 28 4 hmac32(session_key, header[0..28] || payload)
|
||||
* ```
|
||||
*/
|
||||
object Wire {
|
||||
const val HEADER_SIZE = 32
|
||||
val MAGIC = byteArrayOf('E'.code.toByte(), 'L'.code.toByte(), 'T'.code.toByte(), '1'.code.toByte())
|
||||
|
||||
const val TYPE_ECHO_REQ: Int = 0x01
|
||||
const val TYPE_ECHO_RESP: Int = 0x02
|
||||
const val TYPE_TIMESYNC_REQ: Int = 0x07
|
||||
const val TYPE_TIMESYNC_RSP: Int = 0x08
|
||||
const val TYPE_MTU_PROBE: Int = 0x09
|
||||
const val TYPE_MTU_ACK: Int = 0x0A
|
||||
const val TYPE_DELAYED_ECHO: Int = 0x0B
|
||||
|
||||
/** The 8-byte on-the-wire prefix = first 16 hex chars of the session id, decoded. */
|
||||
fun wirePrefix(sessionId: String): ByteArray {
|
||||
require(sessionId.length >= 16) { "session id too short" }
|
||||
val p = ByteArray(8)
|
||||
for (i in 0 until 8) {
|
||||
p[i] = ((hex(sessionId[i * 2]) shl 4) or hex(sessionId[i * 2 + 1])).toByte()
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
private fun hex(c: Char): Int = when (c) {
|
||||
in '0'..'9' -> c - '0'
|
||||
in 'a'..'f' -> c - 'a' + 10
|
||||
in 'A'..'F' -> c - 'A' + 10
|
||||
else -> 0
|
||||
}
|
||||
|
||||
/** Builds a signed packet ready to send. */
|
||||
fun build(
|
||||
type: Int, sessionPrefix: ByteArray, seq: Int, tNs: Long, key: ByteArray,
|
||||
payload: ByteArray = ByteArray(0),
|
||||
): ByteArray {
|
||||
val buf = ByteBuffer.allocate(HEADER_SIZE + payload.size).order(ByteOrder.BIG_ENDIAN)
|
||||
buf.put(MAGIC)
|
||||
buf.put(type.toByte())
|
||||
buf.put(0) // flags
|
||||
buf.putShort(payload.size.toShort())
|
||||
buf.put(sessionPrefix, 0, 8)
|
||||
buf.putInt(seq)
|
||||
buf.putLong(tNs)
|
||||
buf.position(28) // leave hmac slot; fill after
|
||||
buf.putInt(0)
|
||||
buf.put(payload)
|
||||
val bytes = buf.array()
|
||||
// HMAC over header[0..28] || payload (the hmac slot itself excluded).
|
||||
val mac = Crypto.hmacSha256(key, concat(bytes, 0, 28, bytes, HEADER_SIZE, payload.size))
|
||||
mac.copyInto(bytes, 28, 0, 4)
|
||||
return bytes
|
||||
}
|
||||
|
||||
/** A parsed, HMAC-verified inbound packet. */
|
||||
data class Packet(val type: Int, val seq: Int, val tNs: Long, val payload: ByteArray)
|
||||
|
||||
/** Parses and verifies an inbound datagram; null if malformed or the HMAC fails. */
|
||||
fun parseVerified(data: ByteArray, len: Int, key: ByteArray): Packet? {
|
||||
if (len < HEADER_SIZE) return null
|
||||
for (i in MAGIC.indices) if (data[i] != MAGIC[i]) return null
|
||||
val bb = ByteBuffer.wrap(data, 0, len).order(ByteOrder.BIG_ENDIAN)
|
||||
val type = bb.get(4).toInt() and 0xFF
|
||||
val payloadLen = bb.getShort(6).toInt() and 0xFFFF
|
||||
if (HEADER_SIZE + payloadLen > len) return null
|
||||
val expect = Crypto.hmacSha256(key, concat(data, 0, 28, data, HEADER_SIZE, payloadLen))
|
||||
for (i in 0 until 4) if (expect[i] != data[28 + i]) return null
|
||||
val seq = bb.getInt(16)
|
||||
val tNs = bb.getLong(20)
|
||||
val payload = data.copyOfRange(HEADER_SIZE, HEADER_SIZE + payloadLen)
|
||||
return Packet(type, seq, tNs, payload)
|
||||
}
|
||||
|
||||
private fun concat(a: ByteArray, aOff: Int, aLen: Int, b: ByteArray, bOff: Int, bLen: Int): ByteArray {
|
||||
val out = ByteArray(aLen + bLen)
|
||||
a.copyInto(out, 0, aOff, aOff + aLen)
|
||||
b.copyInto(out, aLen, bOff, bOff + bLen)
|
||||
return out
|
||||
}
|
||||
}
|
||||
|
||||
/** Server observation block appended to ECHO_RESP (spec §3.3), fixed 40 bytes. */
|
||||
data class Observation(
|
||||
val tRxNs: Long, val tTxNs: Long, val observedPort: Int, val receivedSize: Int,
|
||||
) {
|
||||
companion object {
|
||||
fun parse(payload: ByteArray): Observation? {
|
||||
if (payload.size < 40) return null
|
||||
val bb = ByteBuffer.wrap(payload).order(ByteOrder.BIG_ENDIAN)
|
||||
return Observation(
|
||||
tRxNs = bb.getLong(0),
|
||||
tTxNs = bb.getLong(8),
|
||||
observedPort = bb.getShort(32).toInt() and 0xFFFF,
|
||||
receivedSize = bb.getInt(36),
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,76 +0,0 @@
|
||||
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
package app.echo_lot.protocol
|
||||
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertNull
|
||||
import kotlin.test.assertNotNull
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
class CryptoWireTest {
|
||||
|
||||
@Test
|
||||
fun hkdfMatchesRfc5869Vector() {
|
||||
// RFC 5869 Appendix A.1 (SHA-256).
|
||||
val ikm = ByteArray(22) { 0x0b }
|
||||
val salt = byteArrayOf(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12)
|
||||
val info = byteArrayOf(
|
||||
0xf0.toByte(), 0xf1.toByte(), 0xf2.toByte(), 0xf3.toByte(), 0xf4.toByte(),
|
||||
0xf5.toByte(), 0xf6.toByte(), 0xf7.toByte(), 0xf8.toByte(), 0xf9.toByte(),
|
||||
)
|
||||
val okm = Crypto.hkdfSha256(ikm, salt, info, 42)
|
||||
val expect = "3cb25f25faacd57a90434f64d0362f2a" +
|
||||
"2d2d0a90cf1a5a4c5db02d56ecc4c5bf" +
|
||||
"34007208d5b887185865"
|
||||
assertEquals(expect, okm.joinToString("") { "%02x".format(it) })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun wirePrefixDecodesHex() {
|
||||
val prefix = Wire.wirePrefix("805a43f8395ae08ace7a14803766cb11")
|
||||
assertEquals("805a43f8395ae08a", prefix.joinToString("") { "%02x".format(it) })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun buildThenParseRoundTripsAndVerifies() {
|
||||
val key = ByteArray(32) { it.toByte() }
|
||||
val prefix = ByteArray(8) { (it + 1).toByte() }
|
||||
val payload = "hello-echolot".toByteArray()
|
||||
val pkt = Wire.build(Wire.TYPE_ECHO_REQ, prefix, 7, 123_456L, key, payload)
|
||||
assertEquals(Wire.HEADER_SIZE + payload.size, pkt.size)
|
||||
|
||||
val parsed = Wire.parseVerified(pkt, pkt.size, key)
|
||||
assertNotNull(parsed)
|
||||
assertEquals(Wire.TYPE_ECHO_REQ, parsed.type)
|
||||
assertEquals(7, parsed.seq)
|
||||
assertEquals(123_456L, parsed.tNs)
|
||||
assertEquals("hello-echolot", String(parsed.payload))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun tamperedHmacIsRejected() {
|
||||
val key = ByteArray(32) { it.toByte() }
|
||||
val pkt = Wire.build(Wire.TYPE_ECHO_REQ, ByteArray(8), 1, 0, key, ByteArray(4))
|
||||
pkt[pkt.size - 1] = (pkt[pkt.size - 1].toInt() xor 0xFF).toByte() // flip a payload byte
|
||||
assertNull(Wire.parseVerified(pkt, pkt.size, key))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun wrongKeyIsRejected() {
|
||||
val pkt = Wire.build(Wire.TYPE_ECHO_REQ, ByteArray(8), 1, 0, ByteArray(32) { 1 }, ByteArray(0))
|
||||
assertNull(Wire.parseVerified(pkt, pkt.size, ByteArray(32) { 2 }))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun observationParses() {
|
||||
// 40-byte block: t_rx, t_tx, 16-byte addr, port, ttl/dscp, size.
|
||||
val b = ByteArray(40)
|
||||
b[33] = 0x1F // port low byte = 8191... set port bytes 32..33
|
||||
b[32] = 0x00
|
||||
val obs = Observation.parse(b)
|
||||
assertNotNull(obs)
|
||||
assertTrue(obs.observedPort in 0..65535)
|
||||
}
|
||||
}
|
||||
@@ -1,66 +0,0 @@
|
||||
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
package app.echo_lot.protocol
|
||||
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertNotNull
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
/**
|
||||
* End-to-end test of the Kotlin client against a REAL running server. It self-skips unless the
|
||||
* environment provides a live target, so it never breaks CI (no network / no server):
|
||||
*
|
||||
* ECHOLOT_LIVE_URL = https://fmr-1.echo-lot.app:8443
|
||||
* ECHOLOT_LIVE_PIN = <base64 pin-sha256>
|
||||
* ECHOLOT_LIVE_CRED = <device credential from an enrollment>
|
||||
* ECHOLOT_LIVE_UDP = fmr-1.echo-lot.app:8442
|
||||
* ECHOLOT_LIVE_TARGET = fmr (profile target id)
|
||||
*
|
||||
* The harness (test-fmr.sh) mints a token over SSH, enrolls via the public control plane, and
|
||||
* exports these — proving the client talks to the deployed server over the wire.
|
||||
*/
|
||||
class LiveServerTest {
|
||||
|
||||
private val url = System.getenv("ECHOLOT_LIVE_URL")
|
||||
private val pin = System.getenv("ECHOLOT_LIVE_PIN")
|
||||
private val cred = System.getenv("ECHOLOT_LIVE_CRED")
|
||||
private val udp = System.getenv("ECHOLOT_LIVE_UDP")
|
||||
private val target = System.getenv("ECHOLOT_LIVE_TARGET") ?: "fmr"
|
||||
|
||||
@Test
|
||||
fun fullFlowAgainstLiveServer() {
|
||||
if (url == null || pin == null || cred == null || udp == null) {
|
||||
println("LiveServerTest skipped (no ECHOLOT_LIVE_* env)")
|
||||
return
|
||||
}
|
||||
val control = ControlClient(url, setOf(pin))
|
||||
|
||||
val profile = control.profile(cred)
|
||||
println("profile: name=${profile.name} v=${profile.serverVersion} caps=${profile.capabilities}")
|
||||
assertTrue(profile.supports("udp-probe"), "server must offer udp-probe")
|
||||
|
||||
val session = control.createSession(cred, target)
|
||||
println("session: ${session.sessionId} expires=${session.expiresS}s")
|
||||
|
||||
val (host, port) = udp.split(":").let { it[0] to it[1].toInt() }
|
||||
ProbeSession(cred, session, host, port).use { ps ->
|
||||
// ECHO: verified response + observation with our observed port.
|
||||
val echo = ps.echo(paddingBytes = 64) // ≥40 so the observation block fits (§3.4)
|
||||
assertNotNull(echo, "no verified ECHO_RESP from live server")
|
||||
println("echo rtt=${"%.1f".format(echo.rttMs)}ms observedPort=${echo.observation?.observedPort} size=${echo.observation?.receivedSize}")
|
||||
assertNotNull(echo.observation, "ECHO_RESP missing observation block")
|
||||
|
||||
// MTU probe: server acks the size it received.
|
||||
val acked = ps.mtuProbe(1400)
|
||||
assertNotNull(acked, "no MTU_ACK from live server")
|
||||
println("mtu probe 1400 -> server received $acked bytes")
|
||||
assertTrue(acked!! in 1300..1500, "acked size implausible: $acked")
|
||||
}
|
||||
|
||||
val obs = control.observations(cred, session.sessionId)
|
||||
println("observations bytes: ${obs.length}")
|
||||
assertTrue(obs.contains("packets_seen"), "observations should report packets_seen")
|
||||
control.deleteSession(cred, session.sessionId)
|
||||
}
|
||||
}
|
||||
@@ -80,6 +80,19 @@ class ControlClient(private val controlUrl: String, pins: Set<String>) {
|
||||
return json.decodeFromString(SessionResponse.serializer(), body(conn))
|
||||
}
|
||||
|
||||
/**
|
||||
* Requests a §5 action. The server creates an asymmetric grant for the granted ones
|
||||
* (downtrain / big_send) and starts sending toward the session's observed data-plane source,
|
||||
* so the caller must already have sent at least one ECHO. Returns the raw JSON reply.
|
||||
*/
|
||||
fun action(credential: String, sessionId: String, bodyJson: String): String {
|
||||
val conn = open("/v1/sessions/$sessionId/actions", "POST", credential)
|
||||
writeJson(conn, bodyJson)
|
||||
val body = body(conn)
|
||||
check(conn.responseCode in 200..299) { "action failed: ${conn.responseCode} $body" }
|
||||
return body
|
||||
}
|
||||
|
||||
fun observations(credential: String, sessionId: String): String {
|
||||
val conn = open("/v1/sessions/$sessionId/observations", "GET", credential)
|
||||
check(conn.responseCode == 200) { "observations failed: ${conn.responseCode}" }
|
||||
|
||||
@@ -60,6 +60,40 @@ class ProbeSession(
|
||||
(resp.payload[3].toInt() and 0xFF)
|
||||
}
|
||||
|
||||
/**
|
||||
* Collects packets the SERVER sends under a grant (downtrain / big_send) for [windowMs].
|
||||
* These arrive unsolicited after a control-plane action, so this just drains the socket and
|
||||
* keeps every HMAC-verified packet — anything that fails verification is not ours and is
|
||||
* silently ignored (an injected packet must not be able to fake a measurement).
|
||||
*/
|
||||
fun collectGranted(windowMs: Long): List<Received> {
|
||||
val out = ArrayList<Received>()
|
||||
val deadline = System.nanoTime() + windowMs * 1_000_000
|
||||
val buf = ByteArray(9200)
|
||||
val prevTimeout = socket.soTimeout
|
||||
try {
|
||||
while (System.nanoTime() < deadline) {
|
||||
val remainMs = ((deadline - System.nanoTime()) / 1_000_000).toInt()
|
||||
if (remainMs <= 0) break
|
||||
socket.soTimeout = remainMs.coerceAtMost(2000)
|
||||
val dp = DatagramPacket(buf, buf.size)
|
||||
try {
|
||||
socket.receive(dp)
|
||||
} catch (e: java.net.SocketTimeoutException) {
|
||||
continue
|
||||
}
|
||||
val pkt = Wire.parseVerified(buf, dp.length, key) ?: continue
|
||||
out.add(Received(pkt.type, pkt.seq, dp.length, (System.nanoTime() - epochNanos)))
|
||||
}
|
||||
} finally {
|
||||
socket.soTimeout = prevTimeout
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
/** One packet received from the server, with the wire size actually delivered. */
|
||||
data class Received(val type: Int, val seq: Int, val sizeBytes: Int, val tRxNs: Long)
|
||||
|
||||
private fun receive(wantType: Int): Wire.Packet? {
|
||||
val buf = ByteArray(2048)
|
||||
return try {
|
||||
|
||||
@@ -28,6 +28,9 @@ object Wire {
|
||||
const val TYPE_MTU_PROBE: Int = 0x09
|
||||
const val TYPE_MTU_ACK: Int = 0x0A
|
||||
const val TYPE_DELAYED_ECHO: Int = 0x0B
|
||||
/** Server->client under an asymmetric grant (spec §3.4/§5). */
|
||||
const val TYPE_DOWNTRAIN_DATA: Int = 0x06
|
||||
const val TYPE_BIG_SEND: Int = 0x0C
|
||||
|
||||
/** The 8-byte on-the-wire prefix = first 16 hex chars of the session id, decoded. */
|
||||
fun wirePrefix(sessionId: String): ByteArray {
|
||||
|
||||
@@ -8,7 +8,8 @@
|
||||
# whole lot to the Gradle test. Proves the Kotlin client talks to the real
|
||||
# server over the wire.
|
||||
#
|
||||
# Usage: JAVA_HOME=... echolot-app/scripts/test-fmr.sh
|
||||
# Usage: JAVA_HOME=... echolot-app/scripts/test-fmr.sh [gradle-task] [test-filter]
|
||||
# e.g. ... test-fmr.sh :core-engine:test '*LiveGrantedTest*'
|
||||
set -euo pipefail
|
||||
|
||||
SSH_HOST="${ECHOLOT_SSH:-claude-echolot}"
|
||||
@@ -33,12 +34,14 @@ PIN=$(echo | openssl s_client -connect "${CTL_HOST}:${CTL_PORT}" 2>/dev/null \
|
||||
| openssl dgst -sha256 -binary | openssl base64)
|
||||
|
||||
echo "· pin=${PIN}"
|
||||
echo "· running LiveServerTest ..."
|
||||
TASK="${1:-:core-protocol:test}"
|
||||
FILTER="${2:-*LiveServerTest*}"
|
||||
echo "· running ${TASK} ${FILTER} ..."
|
||||
cd "$(dirname "$0")/.."
|
||||
ECHOLOT_LIVE_URL="$CTL_URL" \
|
||||
ECHOLOT_LIVE_PIN="$PIN" \
|
||||
ECHOLOT_LIVE_CRED="$CRED" \
|
||||
ECHOLOT_LIVE_UDP="${CTL_HOST}:${UDP_PORT}" \
|
||||
ECHOLOT_LIVE_TARGET="${ECHOLOT_LIVE_TARGET:-fmr}" \
|
||||
./gradlew :core-protocol:test --tests '*LiveServerTest*' --info --rerun-tasks --console=plain \
|
||||
2>&1 | grep -E "profile:|session:|echo |mtu probe|observations bytes|LiveServerTest|BUILD|FAIL|PASS" || true
|
||||
./gradlew "$TASK" --tests "$FILTER" --info --rerun-tasks --console=plain \
|
||||
2>&1 | grep -E "profile:|capabilities:|session:|echo |primed|mtu probe|downtrain|big_send|largest|observations bytes|Live[A-Za-z]*Test|BUILD|FAIL|PASS|^e:" || true
|
||||
|
||||
@@ -39,6 +39,7 @@ import (
|
||||
"echo-lot.app/server/internal/config"
|
||||
"echo-lot.app/server/internal/control"
|
||||
"echo-lot.app/server/internal/dataplane"
|
||||
"echo-lot.app/server/internal/runs"
|
||||
"echo-lot.app/server/internal/selftest"
|
||||
"echo-lot.app/server/internal/selfupdate"
|
||||
"echo-lot.app/server/internal/session"
|
||||
@@ -113,6 +114,23 @@ func serve(cfg *config.Config) error {
|
||||
caps = append(caps, "tcp-echo", "tls-echo")
|
||||
}
|
||||
|
||||
// Uploaded-run storage. A failure here is not fatal: measurement still works, uploads just
|
||||
// stay unavailable and say so in the profile.
|
||||
runStore, err := runs.Open(cfg.StateDir, runs.Policy{
|
||||
Mode: runs.Mode(cfg.UploadsMode),
|
||||
MaxBytes: cfg.UploadMaxBytes,
|
||||
RetentionDays: cfg.UploadRetentionDays,
|
||||
MaxRunsPerDevice: cfg.UploadMaxRuns,
|
||||
MinAnonymization: cfg.UploadMinAnon,
|
||||
})
|
||||
if err != nil {
|
||||
slog.Warn("uploaded-run storage unavailable — uploads disabled", "err", err)
|
||||
runStore = nil
|
||||
} else {
|
||||
slog.Info("uploads", "mode", cfg.UploadsMode, "min_anonymization", cfg.UploadMinAnon,
|
||||
"retention_days", cfg.UploadRetentionDays, "max_runs_per_device", cfg.UploadMaxRuns)
|
||||
}
|
||||
|
||||
ctl := &control.Server{
|
||||
Store: st, Sessions: sessions, Name: cfg.Name,
|
||||
UDPPort: mustPort(firstAddr(cfg.UDPListen)), TCPPort: mustPort(firstAddr(cfg.TCPListen)),
|
||||
@@ -121,6 +139,7 @@ func serve(cfg *config.Config) error {
|
||||
DownTrain: dp.DownTrain,
|
||||
BigSend: dp.BigSend,
|
||||
TCPRecent: func(ip string) any { return tcpSrv.RecentFor(ip) },
|
||||
Runs: runStore,
|
||||
}
|
||||
|
||||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
||||
@@ -172,6 +191,17 @@ func serve(cfg *config.Config) error {
|
||||
r := selftestPtr.Load()
|
||||
return r.MTUOK, r.SysctlOK
|
||||
}
|
||||
// The smallest egress MTU we measured is the ceiling for DF-mode big_send: above it our own
|
||||
// kernel refuses the datagram, which would otherwise look like a downstream path limit.
|
||||
ctl.EgressMTU = func() int {
|
||||
best := 0
|
||||
for _, m := range selftestPtr.Load().EgressMTU {
|
||||
if m.DiscoveredMTU > 0 && (best == 0 || m.DiscoveredMTU < best) {
|
||||
best = m.DiscoveredMTU
|
||||
}
|
||||
}
|
||||
return best
|
||||
}
|
||||
|
||||
// Admin/health (plain HTTP, localhost by default; spec §7)
|
||||
admin := http.NewServeMux()
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"flag"
|
||||
"fmt"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
@@ -49,10 +50,28 @@ type Config struct {
|
||||
// e.g. https://git.example.net/api/v1/repos/owner/repo
|
||||
SelfUpdateAPI string // ECHOLOT_SELF_UPDATE_API / --self-update-api
|
||||
|
||||
// Uploaded-run storage. The default is "anonymous": any enrolled device may upload,
|
||||
// which is what a self-hosted server wants. Operators of shared servers turn it down.
|
||||
UploadsMode string // ECHOLOT_UPLOADS / --uploads (off|anonymous|account)
|
||||
UploadMaxBytes int64 // ECHOLOT_UPLOAD_MAX_BYTES / --upload-max-bytes
|
||||
UploadRetentionDays int // ECHOLOT_UPLOAD_RETENTION_DAYS / --upload-retention-days
|
||||
UploadMaxRuns int // ECHOLOT_UPLOAD_MAX_RUNS / --upload-max-runs (per device)
|
||||
UploadMinAnon string // ECHOLOT_UPLOAD_MIN_ANONYMIZATION / --upload-min-anonymization
|
||||
|
||||
// Mode
|
||||
Docker bool // --docker (or autodetected; env ECHOLOT_DOCKER=1 forces)
|
||||
}
|
||||
|
||||
// envInt reads ECHOLOT_<key> as an integer with a fallback.
|
||||
func envInt(key string, def int) int {
|
||||
if v := envOr(key, ""); v != "" {
|
||||
if n, err := strconv.Atoi(v); err == nil {
|
||||
return n
|
||||
}
|
||||
}
|
||||
return def
|
||||
}
|
||||
|
||||
// envOr reads ECHOLOT_<key> with a fallback.
|
||||
func envOr(key, def string) string {
|
||||
if v, ok := os.LookupEnv("ECHOLOT_" + key); ok {
|
||||
@@ -82,6 +101,11 @@ func Load(args []string) (*Config, *Actions, error) {
|
||||
fs.StringVar(&c.StateDir, "state-dir", envOr("STATE_DIR", defaultStateDir()), "state directory (device store, generated TLS)")
|
||||
fs.StringVar(&c.Name, "name", envOr("NAME", "echolot"), "server profile name")
|
||||
fs.StringVar(&c.SelfUpdateAPI, "self-update-api", envOr("SELF_UPDATE_API", ""), "Gitea repo API base for self-update; empty disables")
|
||||
fs.StringVar(&c.UploadsMode, "uploads", envOr("UPLOADS", "anonymous"), "who may upload measurement runs: off|anonymous|account")
|
||||
fs.Int64Var(&c.UploadMaxBytes, "upload-max-bytes", int64(envInt("UPLOAD_MAX_BYTES", 4<<20)), "largest accepted uploaded run, bytes")
|
||||
fs.IntVar(&c.UploadRetentionDays, "upload-retention-days", envInt("UPLOAD_RETENTION_DAYS", 90), "delete uploaded runs older than this; 0 disables")
|
||||
fs.IntVar(&c.UploadMaxRuns, "upload-max-runs", envInt("UPLOAD_MAX_RUNS", 200), "keep at most this many runs per device; 0 disables")
|
||||
fs.StringVar(&c.UploadMinAnon, "upload-min-anonymization", envOr("UPLOAD_MIN_ANONYMIZATION", "full"), "least anonymization accepted: full|balanced|strict")
|
||||
fs.BoolVar(&c.Docker, "docker", envOr("DOCKER", "") == "1", "force container mode (config from env, no systemd/self-update)")
|
||||
|
||||
fs.BoolVar(&a.InstallSystemd, "install-systemd", false, "install a systemd unit for this binary and exit")
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
package control
|
||||
|
||||
import (
|
||||
"net/netip"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"echo-lot.app/server/internal/session"
|
||||
)
|
||||
|
||||
// The DF ceiling is the difference between "the client's path cannot carry this" and "we could
|
||||
// never have sent it in the first place". Getting the header arithmetic wrong would silently
|
||||
// attribute a server limit to the client's network, so it is pinned here.
|
||||
func TestMaxDFPayload(t *testing.T) {
|
||||
mgr := session.NewManager(time.Minute)
|
||||
|
||||
newSess := func(src string) *session.Session {
|
||||
s, _, err := mgr.New("dev", "cred", netip.MustParseAddr("203.0.113.9"))
|
||||
if err != nil {
|
||||
t.Fatalf("new session: %v", err)
|
||||
}
|
||||
if src != "" {
|
||||
s.NoteDataSource(netip.MustParseAddrPort(src))
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
mtu func() int
|
||||
src string
|
||||
want int
|
||||
}{
|
||||
{"no egress mtu hook means no clamp", nil, "198.51.100.4:5000", 0},
|
||||
{"unknown egress mtu means no clamp", func() int { return 0 }, "198.51.100.4:5000", 0},
|
||||
{"ipv4 subtracts ip+udp", func() int { return 1500 }, "198.51.100.4:5000", 1472},
|
||||
{"ipv6 subtracts the larger header", func() int { return 1500 }, "[2001:db8::4]:5000", 1452},
|
||||
{"pppoe-style 1492 egress", func() int { return 1492 }, "198.51.100.4:5000", 1464},
|
||||
{"no data source yet falls back to ipv4 overhead", func() int { return 1500 }, "", 1472},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
s := &Server{EgressMTU: tc.mtu}
|
||||
if got := s.maxDFPayload(newSess(tc.src)); got != tc.want {
|
||||
t.Fatalf("maxDFPayload = %d, want %d", got, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -15,6 +15,7 @@ import (
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net"
|
||||
"net/http"
|
||||
@@ -23,6 +24,8 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"echo-lot.app/server/internal/dataplane"
|
||||
"echo-lot.app/server/internal/runs"
|
||||
"echo-lot.app/server/internal/session"
|
||||
"echo-lot.app/server/internal/store"
|
||||
)
|
||||
@@ -51,7 +54,13 @@ type Server struct {
|
||||
DelayedEcho func(sess *session.Session, actionID string) error
|
||||
// Granted server->client sends (spec §5). Both consume an asymmetric grant.
|
||||
DownTrain func(sess *session.Session, g *session.Grant, count, sizeBytes, intervalUs int) (int, error)
|
||||
BigSend func(sess *session.Session, g *session.Grant, sizes []int) ([]int, error)
|
||||
BigSend func(sess *session.Session, g *session.Grant, sizes []int, df bool) ([]dataplane.BigSendResult, error)
|
||||
// Runs stores uploaded measurement documents (may be nil: uploads unsupported).
|
||||
Runs *runs.Store
|
||||
// EgressMTU reports the server's own measured egress path MTU (0 = unknown). With DF set
|
||||
// we cannot emit a datagram larger than this, so requested sizes above it are refused up
|
||||
// front and reported as such — the client must not read that as a downstream path limit.
|
||||
EgressMTU func() int
|
||||
// CanaryQueries returns logged canary lookups for a session prefix (may be nil).
|
||||
CanaryQueries func(sessionPrefix string) any
|
||||
// CanaryZone is surfaced in the profile so the app knows what to query.
|
||||
@@ -73,6 +82,10 @@ func (s *Server) Handler() http.Handler {
|
||||
mux.HandleFunc("POST /v1/sessions/{id}/actions", s.actions)
|
||||
mux.HandleFunc("POST /v1/echo", s.httpEcho)
|
||||
mux.HandleFunc("GET /v1/tls-reference", s.tlsReference)
|
||||
mux.HandleFunc("POST /v1/runs", s.uploadRun)
|
||||
mux.HandleFunc("GET /v1/runs", s.listRuns)
|
||||
mux.HandleFunc("GET /v1/runs/{id}", s.getRun)
|
||||
mux.HandleFunc("DELETE /v1/runs/{id}", s.deleteRun)
|
||||
// TODO(spec §5): frag_send, throughput (both build on the same grant machinery)
|
||||
return mux
|
||||
}
|
||||
@@ -156,6 +169,7 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) {
|
||||
SizeBytes int `json:"size_bytes"`
|
||||
IntervalUs int `json:"interval_us"`
|
||||
SizesBytes []int `json:"sizes_bytes"`
|
||||
DF *bool `json:"df"`
|
||||
}
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "bad body"})
|
||||
@@ -241,6 +255,35 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) {
|
||||
if len(sizes) > 32 {
|
||||
sizes = sizes[:32]
|
||||
}
|
||||
// DF on by default: an unfragmented burst is what makes the result a path-MTU
|
||||
// measurement rather than a fragment-delivery one. Callers opt out explicitly.
|
||||
df := true
|
||||
if req.DF != nil {
|
||||
df = *req.DF
|
||||
}
|
||||
// With DF we can only emit up to our own egress MTU minus IP+UDP headers. Drop the
|
||||
// rest here and say so, rather than sending nothing and letting the client blame
|
||||
// the path.
|
||||
maxDF := 0
|
||||
if df {
|
||||
maxDF = s.maxDFPayload(sess)
|
||||
if maxDF > 0 {
|
||||
kept := sizes[:0]
|
||||
for _, x := range sizes {
|
||||
if x <= maxDF {
|
||||
kept = append(kept, x)
|
||||
}
|
||||
}
|
||||
sizes = kept
|
||||
}
|
||||
}
|
||||
if len(sizes) == 0 {
|
||||
writeJSON(w, http.StatusBadRequest, map[string]any{
|
||||
"error": "every requested size exceeds the server's own egress MTU with DF set",
|
||||
"max_df_bytes": maxDF,
|
||||
})
|
||||
return
|
||||
}
|
||||
total := 0
|
||||
for _, x := range sizes {
|
||||
total += clamp(x, dataMinPacket, 9000)
|
||||
@@ -251,11 +294,11 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
go func() {
|
||||
attempted, err := s.BigSend(sess, g, sizes)
|
||||
slog.Info("big_send finished", "action", actionID, "attempted", attempted, "err", err)
|
||||
results, err := s.BigSend(sess, g, sizes, df)
|
||||
slog.Info("big_send finished", "action", actionID, "results", results, "df", df, "err", err)
|
||||
}()
|
||||
writeJSON(w, http.StatusAccepted, map[string]any{
|
||||
"action_id": actionID, "sizes_bytes": sizes,
|
||||
"action_id": actionID, "sizes_bytes": sizes, "df": df, "max_df_bytes": maxDF,
|
||||
"grant": map[string]any{"max_bytes": g.MaxBytes, "max_kbps": g.MaxKbps},
|
||||
})
|
||||
|
||||
@@ -264,6 +307,24 @@ func (s *Server) actions(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
}
|
||||
|
||||
// maxDFPayload is the largest UDP payload the server can emit toward this session without
|
||||
// fragmenting: its own egress MTU less the IP and UDP headers of the session's address family.
|
||||
// Returns 0 when the egress MTU is unknown, meaning "do not clamp".
|
||||
func (s *Server) maxDFPayload(sess *session.Session) int {
|
||||
if s.EgressMTU == nil {
|
||||
return 0
|
||||
}
|
||||
mtu := s.EgressMTU()
|
||||
if mtu <= 0 {
|
||||
return 0
|
||||
}
|
||||
overhead := 28 // IPv4 (20) + UDP (8)
|
||||
if src := sess.DataSource(); src.IsValid() && !src.Addr().Unmap().Is4() {
|
||||
overhead = 48 // IPv6 (40) + UDP (8)
|
||||
}
|
||||
return mtu - overhead
|
||||
}
|
||||
|
||||
// dataMinPacket is the smallest datagram that still carries a header + a little payload.
|
||||
const dataMinPacket = 40
|
||||
|
||||
@@ -360,6 +421,9 @@ func (s *Server) profile(w http.ResponseWriter, r *http.Request) {
|
||||
"canary_zone": s.CanaryZone,
|
||||
"server_selftest": selftestSignal(s.ProvenGood),
|
||||
"limits": map[string]any{"max_kbps": 50000, "max_session_s": 900},
|
||||
// The app needs the upload rules before it offers the switch: whether uploads are
|
||||
// accepted at all, and how much identifying detail it must strip first.
|
||||
"uploads": s.uploadPolicy(),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -411,3 +475,102 @@ func SpkiPinB64(cert tls.Certificate) (string, error) {
|
||||
sum := sha256.Sum256(leaf.RawSubjectPublicKeyInfo)
|
||||
return base64.StdEncoding.EncodeToString(sum[:]), nil
|
||||
}
|
||||
|
||||
// uploadPolicy is the profile's advertisement of the operator's upload rules.
|
||||
func (s *Server) uploadPolicy() map[string]any {
|
||||
if s.Runs == nil {
|
||||
return map[string]any{"mode": string(runs.ModeOff), "reason": "not configured"}
|
||||
}
|
||||
p := s.Runs.Policy()
|
||||
return map[string]any{
|
||||
"mode": string(p.Mode),
|
||||
"max_bytes": p.MaxBytes,
|
||||
"retention_days": p.RetentionDays,
|
||||
"max_runs_per_device": p.MaxRunsPerDevice,
|
||||
"min_anonymization": p.MinAnonymization,
|
||||
}
|
||||
}
|
||||
|
||||
// uploadRun stores one measurement document for the calling device.
|
||||
func (s *Server) uploadRun(w http.ResponseWriter, r *http.Request) {
|
||||
dev := s.Store.DeviceByCredential(bearer(r))
|
||||
if dev == nil {
|
||||
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unknown credential"})
|
||||
return
|
||||
}
|
||||
if s.Runs == nil {
|
||||
writeJSON(w, http.StatusForbidden, map[string]string{"error": runs.ErrDisabled.Error()})
|
||||
return
|
||||
}
|
||||
limit := s.Runs.Policy().MaxBytes
|
||||
if limit <= 0 {
|
||||
limit = 4 << 20
|
||||
}
|
||||
// +1 so a body exactly at the limit is distinguishable from one over it.
|
||||
body, err := io.ReadAll(io.LimitReader(r.Body, limit+1))
|
||||
if err != nil {
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "read failed"})
|
||||
return
|
||||
}
|
||||
meta, err := s.Runs.Put(dev.ID, body)
|
||||
switch {
|
||||
case err == nil:
|
||||
slog.Info("run uploaded", "device", dev.ID, "run", meta.ID,
|
||||
"bytes", meta.SizeBytes, "anon", meta.Anonymization, "findings", meta.FindingCount)
|
||||
writeJSON(w, http.StatusCreated, meta)
|
||||
case errors.Is(err, runs.ErrDisabled), errors.Is(err, runs.ErrNeedAccount),
|
||||
errors.Is(err, runs.ErrNotAnonEnough):
|
||||
writeJSON(w, http.StatusForbidden, map[string]any{
|
||||
"error": err.Error(), "uploads": s.uploadPolicy(),
|
||||
})
|
||||
case errors.Is(err, runs.ErrTooLarge):
|
||||
writeJSON(w, http.StatusRequestEntityTooLarge, map[string]any{
|
||||
"error": err.Error(), "max_bytes": s.Runs.Policy().MaxBytes,
|
||||
})
|
||||
default:
|
||||
writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) listRuns(w http.ResponseWriter, r *http.Request) {
|
||||
dev := s.Store.DeviceByCredential(bearer(r))
|
||||
if dev == nil || s.Runs == nil {
|
||||
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unknown credential"})
|
||||
return
|
||||
}
|
||||
list := s.Runs.List(dev.ID)
|
||||
if list == nil {
|
||||
list = []runs.Meta{}
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]any{"runs": list})
|
||||
}
|
||||
|
||||
func (s *Server) getRun(w http.ResponseWriter, r *http.Request) {
|
||||
dev := s.Store.DeviceByCredential(bearer(r))
|
||||
if dev == nil || s.Runs == nil {
|
||||
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unknown credential"})
|
||||
return
|
||||
}
|
||||
// Scoped to the calling device's own directory: one device cannot read another's runs by
|
||||
// guessing a run id.
|
||||
b, err := s.Runs.Get(dev.ID, r.PathValue("id"))
|
||||
if err != nil {
|
||||
writeJSON(w, http.StatusNotFound, map[string]string{"error": "no such run"})
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write(b)
|
||||
}
|
||||
|
||||
func (s *Server) deleteRun(w http.ResponseWriter, r *http.Request) {
|
||||
dev := s.Store.DeviceByCredential(bearer(r))
|
||||
if dev == nil || s.Runs == nil {
|
||||
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unknown credential"})
|
||||
return
|
||||
}
|
||||
if err := s.Runs.Delete(dev.ID, r.PathValue("id")); err != nil {
|
||||
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
//go:build linux
|
||||
|
||||
package dataplane
|
||||
|
||||
import (
|
||||
"net"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
// PMTUD socket-option values. Go's syscall package exports IP_MTU_DISCOVER and
|
||||
// IPV6_MTU_DISCOVER but not the IP_PMTUDISC_* values, so they are spelled out
|
||||
// here (include/linux/in.h, in6.h — stable ABI, same reasoning as the client's
|
||||
// OsAbi.kt).
|
||||
const (
|
||||
pmtudiscWant = 0 // per-route default: fragment locally when needed
|
||||
pmtudiscDo = 2 // always set DF: oversized sends fail with EMSGSIZE, never fragment
|
||||
)
|
||||
|
||||
// withDF runs fn with the Don't-Fragment bit forced on for conn, then puts the
|
||||
// socket back the way it was found.
|
||||
//
|
||||
// The socket is shared by every session on that address family, so the caller
|
||||
// must hold Server.dfMu: a concurrent big_send must not silently ride along
|
||||
// with someone else's DF window (or, worse, clear it mid-flight).
|
||||
func withDF(conn *net.UDPConn, fn func() error) error {
|
||||
raw, err := conn.SyscallConn()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
v4 := conn.LocalAddr().(*net.UDPAddr).IP.To4() != nil
|
||||
level, opt := syscall.IPPROTO_IPV6, syscall.IPV6_MTU_DISCOVER
|
||||
if v4 {
|
||||
level, opt = syscall.IPPROTO_IP, syscall.IP_MTU_DISCOVER
|
||||
}
|
||||
|
||||
var setErr error
|
||||
prev := pmtudiscWant
|
||||
if err := raw.Control(func(fd uintptr) {
|
||||
if p, e := syscall.GetsockoptInt(int(fd), level, opt); e == nil {
|
||||
prev = p
|
||||
}
|
||||
setErr = syscall.SetsockoptInt(int(fd), level, opt, pmtudiscDo)
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
if setErr != nil {
|
||||
return setErr
|
||||
}
|
||||
defer func() {
|
||||
_ = raw.Control(func(fd uintptr) {
|
||||
_ = syscall.SetsockoptInt(int(fd), level, opt, prev)
|
||||
})
|
||||
}()
|
||||
return fn()
|
||||
}
|
||||
|
||||
// dfSupported reports whether withDF can actually set the DF bit here.
|
||||
const dfSupported = true
|
||||
@@ -0,0 +1,16 @@
|
||||
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
//go:build !linux
|
||||
|
||||
package dataplane
|
||||
|
||||
import "net"
|
||||
|
||||
// Forcing DF per-socket is Linux-specific (IP_MTU_DISCOVER). Off Linux the
|
||||
// send still happens — just without the guarantee that nothing fragmented it,
|
||||
// so the caller must report the result as fragment-delivery evidence rather
|
||||
// than a path-MTU measurement. See dfSupported.
|
||||
func withDF(conn *net.UDPConn, fn func() error) error { return fn() }
|
||||
|
||||
const dfSupported = false
|
||||
@@ -52,10 +52,25 @@ func (s *Server) DownTrain(sess *session.Session, g *session.Grant, count, sizeB
|
||||
return sent, nil
|
||||
}
|
||||
|
||||
// BigSend transmits one datagram per requested size, largest-first metadata intact, so the client
|
||||
// can see which sizes survive the *downstream* path — the mtu.pmtud_down / mtu.blackhole evidence.
|
||||
// The client cannot produce this itself: only the far end can emit a large packet toward it.
|
||||
func (s *Server) BigSend(sess *session.Session, g *session.Grant, sizes []int) ([]int, error) {
|
||||
// BigSendResult records what happened to one requested size. `Sent` false with an EMSGSIZE-ish
|
||||
// Err means *we* could not put it on the wire (the datagram exceeds our own egress MTU with DF
|
||||
// set) — the client must not read its absence as a path limit, so this is reported, not hidden.
|
||||
type BigSendResult struct {
|
||||
SizeBytes int `json:"size_bytes"`
|
||||
Seq int `json:"seq"`
|
||||
Sent bool `json:"sent"`
|
||||
Err string `json:"err,omitempty"`
|
||||
}
|
||||
|
||||
// BigSend transmits one datagram per requested size so the client can see which sizes survive the
|
||||
// *downstream* path — the mtu.pmtud_down / mtu.frag_delivery evidence. The client cannot produce
|
||||
// this itself: only the far end can emit a large packet toward it.
|
||||
//
|
||||
// With df set, the DF bit is forced for the whole burst, so nothing fragments and the largest
|
||||
// size that arrives IS the downstream path MTU. Without it, the kernel fragments freely and the
|
||||
// result only says whether fragments get through — a different (also useful) measurement, and
|
||||
// the reason the two are separate test types.
|
||||
func (s *Server) BigSend(sess *session.Session, g *session.Grant, sizes []int, df bool) ([]BigSendResult, error) {
|
||||
target := sess.DataSource()
|
||||
if !target.IsValid() {
|
||||
return nil, fmt.Errorf("no observed data-plane source")
|
||||
@@ -64,24 +79,46 @@ func (s *Server) BigSend(sess *session.Session, g *session.Grant, sizes []int) (
|
||||
if conn == nil {
|
||||
return nil, fmt.Errorf("no data-plane socket matches target family")
|
||||
}
|
||||
attempted := make([]int, 0, len(sizes))
|
||||
for i, size := range sizes {
|
||||
if size < HeaderSize+8 {
|
||||
size = HeaderSize + 8
|
||||
|
||||
results := make([]BigSendResult, 0, len(sizes))
|
||||
burst := func() error {
|
||||
for i, size := range sizes {
|
||||
if size < HeaderSize+8 {
|
||||
size = HeaderSize + 8
|
||||
}
|
||||
if size > 9000 { // jumbo ceiling; beyond this the kernel will refuse anyway
|
||||
size = 9000
|
||||
}
|
||||
if !g.Allow(size) {
|
||||
break
|
||||
}
|
||||
payload := make([]byte, size-HeaderSize)
|
||||
// Echo the intended size into the payload so a truncated/fragmented arrival is
|
||||
// still attributable to the size we meant to send.
|
||||
binary.BigEndian.PutUint32(payload[0:4], uint32(size))
|
||||
err := s.sendErr(conn, target, sess, TypeBigSend, uint32(i), payload)
|
||||
results = append(results, BigSendResult{
|
||||
SizeBytes: size, Seq: i, Sent: err == nil, Err: errString(err),
|
||||
})
|
||||
time.Sleep(20 * time.Millisecond) // keep bursts from being read as congestion loss
|
||||
}
|
||||
if size > 9000 { // jumbo ceiling; beyond this the kernel will refuse anyway
|
||||
size = 9000
|
||||
}
|
||||
if !g.Allow(size) {
|
||||
break
|
||||
}
|
||||
payload := make([]byte, size-HeaderSize)
|
||||
// Echo the intended size into the payload so a truncated/fragmented arrival is
|
||||
// still attributable to the size we meant to send.
|
||||
binary.BigEndian.PutUint32(payload[0:4], uint32(size))
|
||||
s.send(conn, target, sess, TypeBigSend, uint32(i), payload)
|
||||
attempted = append(attempted, size)
|
||||
time.Sleep(20 * time.Millisecond) // keep bursts from being read as congestion loss
|
||||
return nil
|
||||
}
|
||||
return attempted, nil
|
||||
|
||||
if df && dfSupported {
|
||||
s.dfMu.Lock()
|
||||
defer s.dfMu.Unlock()
|
||||
if err := withDF(conn, burst); err != nil {
|
||||
return results, err
|
||||
}
|
||||
return results, nil
|
||||
}
|
||||
return results, burst()
|
||||
}
|
||||
|
||||
func errString(err error) string {
|
||||
if err == nil {
|
||||
return ""
|
||||
}
|
||||
return err.Error()
|
||||
}
|
||||
|
||||
@@ -45,6 +45,10 @@ type Server struct {
|
||||
|
||||
mu sync.Mutex
|
||||
conns []*net.UDPConn
|
||||
|
||||
// dfMu serialises DF windows: the listening socket is shared by every session on that
|
||||
// family, so two concurrent big_sends must not overlap their DF on/off transitions.
|
||||
dfMu sync.Mutex
|
||||
}
|
||||
|
||||
// Serve runs the read loop for one socket; call once per bound address.
|
||||
@@ -203,6 +207,13 @@ func (s *Server) timesyncResp(conn *net.UDPConn, raddr netip.AddrPort, sess *ses
|
||||
}
|
||||
|
||||
func (s *Server) send(conn *net.UDPConn, raddr netip.AddrPort, sess *session.Session, typ byte, seq uint32, payload []byte) {
|
||||
_ = s.sendErr(conn, raddr, sess, typ, seq, payload)
|
||||
}
|
||||
|
||||
// sendErr is send with the write error surfaced. Only the DF-mode big_send cares: there an
|
||||
// EMSGSIZE means our own egress MTU refused the datagram, which is a different fact from the
|
||||
// client not receiving it.
|
||||
func (s *Server) sendErr(conn *net.UDPConn, raddr netip.AddrPort, sess *session.Session, typ byte, seq uint32, payload []byte) error {
|
||||
pkt := make([]byte, HeaderSize+len(payload))
|
||||
copy(pkt[0:4], Magic)
|
||||
pkt[4] = typ
|
||||
@@ -218,7 +229,8 @@ func (s *Server) send(conn *net.UDPConn, raddr netip.AddrPort, sess *session.Ses
|
||||
mac.Write(pkt[0:28])
|
||||
mac.Write(payload)
|
||||
copy(pkt[28:32], mac.Sum(nil)[:4])
|
||||
_, _ = conn.WriteToUDPAddrPort(pkt, raddr)
|
||||
_, err := conn.WriteToUDPAddrPort(pkt, raddr)
|
||||
return err
|
||||
}
|
||||
|
||||
func hexByte(hi, lo byte) byte {
|
||||
|
||||
@@ -0,0 +1,300 @@
|
||||
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
// Package runs stores uploaded measurement documents.
|
||||
//
|
||||
// The premise (and the reason uploads exist at all) is that an engineer runs their own server:
|
||||
// uploading a run there is how history, diffing and "it was fine last Tuesday" work. That makes
|
||||
// the storage deliberately dumb — one JSON file per run, on disk, greppable, deletable with rm —
|
||||
// and puts the interesting policy in two places instead:
|
||||
//
|
||||
// - who may upload (Policy.Mode), because a public server is a different proposition from a
|
||||
// private one; and
|
||||
// - how much identifying detail the client must strip first (Policy.MinAnonymization), because
|
||||
// someone measuring against a stranger's server should not be shipping their SSIDs there.
|
||||
//
|
||||
// Retention is enforced on every upload, not by a sweeper, so a server left alone does not grow.
|
||||
package runs
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Mode says who may upload.
|
||||
type Mode string
|
||||
|
||||
const (
|
||||
// ModeOff refuses every upload. The endpoint still answers, with 403 and a reason, so the
|
||||
// app can say "this server does not accept uploads" instead of showing a network error.
|
||||
ModeOff Mode = "off"
|
||||
// ModeAnonymous accepts uploads from any enrolled device. The default: enrollment already
|
||||
// required an admin-minted token, so "anyone enrolled" is not "anyone".
|
||||
ModeAnonymous Mode = "anonymous"
|
||||
// ModeAccount accepts uploads only from a device tied to a signed-in account. The account
|
||||
// system (OIDC) is not built yet, so today this refuses everything with a distinct reason —
|
||||
// it exists so operators can pick the strict setting now and have it mean the right thing
|
||||
// when accounts land, rather than silently loosening on upgrade.
|
||||
ModeAccount Mode = "account"
|
||||
)
|
||||
|
||||
// Anonymization levels, mirroring the client's redaction levels (measurement-schema.md §8).
|
||||
// Ordered: full < balanced < strict.
|
||||
const (
|
||||
AnonFull = "full" // nothing removed — for your own server
|
||||
AnonBalanced = "balanced" // network names and device identity pseudonymized, neighbours dropped
|
||||
AnonStrict = "strict" // metrics and findings only
|
||||
)
|
||||
|
||||
func anonRank(level string) int {
|
||||
switch level {
|
||||
case AnonStrict:
|
||||
return 2
|
||||
case AnonBalanced:
|
||||
return 1
|
||||
case AnonFull:
|
||||
return 0
|
||||
}
|
||||
return -1 // unknown
|
||||
}
|
||||
|
||||
// Policy is the operator's upload configuration.
|
||||
type Policy struct {
|
||||
Mode Mode `json:"mode"`
|
||||
MaxBytes int64 `json:"max_bytes"`
|
||||
RetentionDays int `json:"retention_days"`
|
||||
MaxRunsPerDevice int `json:"max_runs_per_device"`
|
||||
MinAnonymization string `json:"min_anonymization"`
|
||||
}
|
||||
|
||||
func DefaultPolicy() Policy {
|
||||
return Policy{
|
||||
Mode: ModeAnonymous,
|
||||
MaxBytes: 4 << 20,
|
||||
RetentionDays: 90,
|
||||
MaxRunsPerDevice: 200,
|
||||
MinAnonymization: AnonFull,
|
||||
}
|
||||
}
|
||||
|
||||
var (
|
||||
ErrDisabled = errors.New("uploads are disabled on this server")
|
||||
ErrNeedAccount = errors.New("this server only accepts uploads from signed-in accounts")
|
||||
ErrTooLarge = errors.New("run exceeds the server's upload size limit")
|
||||
ErrNotAnonEnough = errors.New("run is less anonymized than this server requires")
|
||||
ErrMalformed = errors.New("run is not a measurement document")
|
||||
)
|
||||
|
||||
// Meta is the index entry for one stored run — enough to list history without opening the files.
|
||||
type Meta struct {
|
||||
ID string `json:"id"`
|
||||
DeviceID string `json:"device_id"`
|
||||
UploadedAt time.Time `json:"uploaded_at"`
|
||||
StartedAt string `json:"started_at,omitempty"`
|
||||
Anonymization string `json:"anonymization"`
|
||||
SizeBytes int64 `json:"size_bytes"`
|
||||
Verdict string `json:"verdict,omitempty"`
|
||||
FindingCount int `json:"finding_count"`
|
||||
}
|
||||
|
||||
type Store struct {
|
||||
mu sync.Mutex
|
||||
dir string
|
||||
policy Policy
|
||||
}
|
||||
|
||||
func Open(stateDir string, p Policy) (*Store, error) {
|
||||
dir := filepath.Join(stateDir, "runs")
|
||||
if err := os.MkdirAll(dir, 0o700); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &Store{dir: dir, policy: p}, nil
|
||||
}
|
||||
|
||||
func (s *Store) Policy() Policy { return s.policy }
|
||||
|
||||
// Accepts reports whether an upload would be allowed at all, so callers can answer the
|
||||
// capability question without a body.
|
||||
func (s *Store) Accepts() error {
|
||||
switch s.policy.Mode {
|
||||
case ModeOff:
|
||||
return ErrDisabled
|
||||
case ModeAccount:
|
||||
return ErrNeedAccount
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Put validates and stores one uploaded document. body is the raw JSON as received: it is stored
|
||||
// byte-for-byte so what the device signed off on is what sits on disk.
|
||||
func (s *Store) Put(deviceID string, body []byte) (Meta, error) {
|
||||
if err := s.Accepts(); err != nil {
|
||||
return Meta{}, err
|
||||
}
|
||||
if s.policy.MaxBytes > 0 && int64(len(body)) > s.policy.MaxBytes {
|
||||
return Meta{}, ErrTooLarge
|
||||
}
|
||||
|
||||
// Peek at the parts we index on. Unknown fields are ignored: the server must not become a
|
||||
// second schema authority that rejects documents a newer client legitimately produces.
|
||||
var doc struct {
|
||||
Run struct {
|
||||
ID string `json:"id"`
|
||||
StartedAt string `json:"started_at"`
|
||||
Privacy struct {
|
||||
Anonymization string `json:"anonymization"`
|
||||
} `json:"privacy"`
|
||||
} `json:"run"`
|
||||
Findings []json.RawMessage `json:"findings"`
|
||||
Summary struct {
|
||||
Verdict string `json:"verdict"`
|
||||
} `json:"summary"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &doc); err != nil || doc.Run.ID == "" {
|
||||
return Meta{}, ErrMalformed
|
||||
}
|
||||
|
||||
level := doc.Run.Privacy.Anonymization
|
||||
if level == "" {
|
||||
level = AnonFull // no declaration means nothing was stripped
|
||||
}
|
||||
if anonRank(level) < anonRank(s.policy.MinAnonymization) {
|
||||
return Meta{}, fmt.Errorf("%w: got %q, need at least %q",
|
||||
ErrNotAnonEnough, level, s.policy.MinAnonymization)
|
||||
}
|
||||
|
||||
id := sanitizeID(doc.Run.ID)
|
||||
if id == "" {
|
||||
return Meta{}, ErrMalformed
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
devDir := filepath.Join(s.dir, sanitizeID(deviceID))
|
||||
if err := os.MkdirAll(devDir, 0o700); err != nil {
|
||||
return Meta{}, err
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(devDir, id+".json"), body, 0o600); err != nil {
|
||||
return Meta{}, err
|
||||
}
|
||||
|
||||
meta := Meta{
|
||||
ID: id, DeviceID: deviceID, UploadedAt: time.Now().UTC(),
|
||||
StartedAt: doc.Run.StartedAt, Anonymization: level,
|
||||
SizeBytes: int64(len(body)), Verdict: doc.Summary.Verdict,
|
||||
FindingCount: len(doc.Findings),
|
||||
}
|
||||
if err := os.WriteFile(filepath.Join(devDir, id+".meta.json"), mustJSON(meta), 0o600); err != nil {
|
||||
return Meta{}, err
|
||||
}
|
||||
s.enforceRetentionLocked(devDir)
|
||||
return meta, nil
|
||||
}
|
||||
|
||||
// List returns one device's runs, newest first.
|
||||
func (s *Store) List(deviceID string) []Meta {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.listLocked(filepath.Join(s.dir, sanitizeID(deviceID)))
|
||||
}
|
||||
|
||||
// Get returns the stored document bytes for one run.
|
||||
func (s *Store) Get(deviceID, runID string) ([]byte, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return os.ReadFile(filepath.Join(s.dir, sanitizeID(deviceID), sanitizeID(runID)+".json"))
|
||||
}
|
||||
|
||||
// Delete removes one run. Missing is not an error: delete is idempotent so a client retrying
|
||||
// after a dropped response does not see a spurious failure.
|
||||
func (s *Store) Delete(deviceID, runID string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
dev, run := sanitizeID(deviceID), sanitizeID(runID)
|
||||
for _, suffix := range []string{".json", ".meta.json"} {
|
||||
if err := os.Remove(filepath.Join(s.dir, dev, run+suffix)); err != nil && !errors.Is(err, os.ErrNotExist) {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Store) listLocked(devDir string) []Meta {
|
||||
entries, err := os.ReadDir(devDir)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
out := make([]Meta, 0, len(entries))
|
||||
for _, e := range entries {
|
||||
if !strings.HasSuffix(e.Name(), ".meta.json") {
|
||||
continue
|
||||
}
|
||||
b, err := os.ReadFile(filepath.Join(devDir, e.Name()))
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
var m Meta
|
||||
if json.Unmarshal(b, &m) == nil {
|
||||
out = append(out, m)
|
||||
}
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].UploadedAt.After(out[j].UploadedAt) })
|
||||
return out
|
||||
}
|
||||
|
||||
// enforceRetentionLocked drops runs past the age limit, then past the count limit. Age first, so
|
||||
// a burst of uploads cannot push out runs that are still inside the retention window.
|
||||
func (s *Store) enforceRetentionLocked(devDir string) {
|
||||
metas := s.listLocked(devDir)
|
||||
drop := func(m Meta) {
|
||||
_ = os.Remove(filepath.Join(devDir, m.ID+".json"))
|
||||
_ = os.Remove(filepath.Join(devDir, m.ID+".meta.json"))
|
||||
}
|
||||
kept := metas[:0]
|
||||
if s.policy.RetentionDays > 0 {
|
||||
cutoff := time.Now().Add(-time.Duration(s.policy.RetentionDays) * 24 * time.Hour)
|
||||
for _, m := range metas {
|
||||
if m.UploadedAt.Before(cutoff) {
|
||||
drop(m)
|
||||
continue
|
||||
}
|
||||
kept = append(kept, m)
|
||||
}
|
||||
} else {
|
||||
kept = metas
|
||||
}
|
||||
if s.policy.MaxRunsPerDevice > 0 && len(kept) > s.policy.MaxRunsPerDevice {
|
||||
for _, m := range kept[s.policy.MaxRunsPerDevice:] { // listLocked is newest-first
|
||||
drop(m)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// sanitizeID keeps ids to characters that cannot escape the directory or collide with the
|
||||
// .meta.json suffix convention. Ids are uuids and device ids in practice; anything else is
|
||||
// truncated to nothing and rejected upstream.
|
||||
func sanitizeID(s string) string {
|
||||
var b strings.Builder
|
||||
for _, r := range s {
|
||||
switch {
|
||||
case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '_':
|
||||
b.WriteRune(r)
|
||||
}
|
||||
if b.Len() >= 64 {
|
||||
break
|
||||
}
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
||||
func mustJSON(v any) []byte {
|
||||
b, _ := json.Marshal(v)
|
||||
return b
|
||||
}
|
||||
@@ -0,0 +1,197 @@
|
||||
// SPDX-FileCopyrightText: 2026 Echolot contributors
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
package runs
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func doc(id, anon string) []byte {
|
||||
return []byte(fmt.Sprintf(
|
||||
`{"run":{"id":%q,"started_at":"2026-08-01T10:00:00Z","privacy":{"anonymization":%q}},`+
|
||||
`"findings":[{"id":"f1"},{"id":"f2"}],"summary":{"verdict":"warn"}}`, id, anon))
|
||||
}
|
||||
|
||||
func open(t *testing.T, p Policy) (*Store, string) {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
s, err := Open(dir, p)
|
||||
if err != nil {
|
||||
t.Fatalf("open: %v", err)
|
||||
}
|
||||
return s, dir
|
||||
}
|
||||
|
||||
func TestModeOffRefusesEverything(t *testing.T) {
|
||||
p := DefaultPolicy()
|
||||
p.Mode = ModeOff
|
||||
s, _ := open(t, p)
|
||||
if _, err := s.Put("dev1", doc("run-1", AnonFull)); !errors.Is(err, ErrDisabled) {
|
||||
t.Fatalf("want ErrDisabled, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// ModeAccount must refuse today rather than fall back to anonymous: an operator who selects the
|
||||
// strict setting before accounts exist must not be silently running the permissive one.
|
||||
func TestModeAccountRefusesUntilAccountsExist(t *testing.T) {
|
||||
p := DefaultPolicy()
|
||||
p.Mode = ModeAccount
|
||||
s, _ := open(t, p)
|
||||
if _, err := s.Put("dev1", doc("run-1", AnonFull)); !errors.Is(err, ErrNeedAccount) {
|
||||
t.Fatalf("want ErrNeedAccount, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMinAnonymizationEnforced(t *testing.T) {
|
||||
p := DefaultPolicy()
|
||||
p.MinAnonymization = AnonBalanced
|
||||
s, _ := open(t, p)
|
||||
|
||||
if _, err := s.Put("dev1", doc("run-full", AnonFull)); !errors.Is(err, ErrNotAnonEnough) {
|
||||
t.Fatalf("full should be refused when balanced is required, got %v", err)
|
||||
}
|
||||
// An undeclared level means nothing was stripped, so it must be treated as "full".
|
||||
if _, err := s.Put("dev1", []byte(`{"run":{"id":"run-bare"},"summary":{}}`)); !errors.Is(err, ErrNotAnonEnough) {
|
||||
t.Fatalf("undeclared level should be treated as full, got %v", err)
|
||||
}
|
||||
for _, lvl := range []string{AnonBalanced, AnonStrict} {
|
||||
if _, err := s.Put("dev1", doc("run-"+lvl, lvl)); err != nil {
|
||||
t.Fatalf("%s should be accepted: %v", lvl, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSizeLimit(t *testing.T) {
|
||||
p := DefaultPolicy()
|
||||
p.MaxBytes = 200
|
||||
s, _ := open(t, p)
|
||||
big := append(doc("run-1", AnonFull), make([]byte, 400)...)
|
||||
if _, err := s.Put("dev1", big); !errors.Is(err, ErrTooLarge) {
|
||||
t.Fatalf("want ErrTooLarge, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRetentionByCountKeepsNewest(t *testing.T) {
|
||||
p := DefaultPolicy()
|
||||
p.MaxRunsPerDevice = 3
|
||||
s, _ := open(t, p)
|
||||
for i := 0; i < 6; i++ {
|
||||
if _, err := s.Put("dev1", doc(fmt.Sprintf("run-%d", i), AnonFull)); err != nil {
|
||||
t.Fatalf("put %d: %v", i, err)
|
||||
}
|
||||
time.Sleep(2 * time.Millisecond) // distinct UploadedAt so "newest" is well defined
|
||||
}
|
||||
got := s.List("dev1")
|
||||
if len(got) != 3 {
|
||||
t.Fatalf("kept %d runs, want 3", len(got))
|
||||
}
|
||||
for i, want := range []string{"run-5", "run-4", "run-3"} {
|
||||
if got[i].ID != want {
|
||||
t.Fatalf("kept[%d] = %s, want %s (newest first)", i, got[i].ID, want)
|
||||
}
|
||||
}
|
||||
// The documents themselves must be gone too, not just their index entries.
|
||||
if _, err := s.Get("dev1", "run-0"); err == nil {
|
||||
t.Fatal("purged run is still readable")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRetentionByAge(t *testing.T) {
|
||||
p := DefaultPolicy()
|
||||
p.RetentionDays = 7
|
||||
p.MaxRunsPerDevice = 0
|
||||
s, dir := open(t, p)
|
||||
if _, err := s.Put("dev1", doc("run-old", AnonFull)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Backdate the index entry past the retention window.
|
||||
metaPath := filepath.Join(dir, "runs", "dev1", "run-old.meta.json")
|
||||
b, _ := os.ReadFile(metaPath)
|
||||
var m Meta
|
||||
_ = json.Unmarshal(b, &m)
|
||||
m.UploadedAt = time.Now().Add(-30 * 24 * time.Hour)
|
||||
nb, _ := json.Marshal(m)
|
||||
if err := os.WriteFile(metaPath, nb, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if _, err := s.Put("dev1", doc("run-new", AnonFull)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got := s.List("dev1")
|
||||
if len(got) != 1 || got[0].ID != "run-new" {
|
||||
t.Fatalf("age retention did not drop the old run: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Run and device ids reach the filesystem, so a hostile one must not be able to climb out of the
|
||||
// store directory or overwrite another device's data.
|
||||
func TestIDsCannotEscapeTheStoreDirectory(t *testing.T) {
|
||||
s, dir := open(t, DefaultPolicy())
|
||||
if _, err := s.Put("../../etc", doc("../../../passwd", AnonFull)); err != nil {
|
||||
t.Fatalf("put: %v", err)
|
||||
}
|
||||
var found []string
|
||||
_ = filepath.Walk(dir, func(p string, info os.FileInfo, err error) error {
|
||||
if err == nil && !info.IsDir() {
|
||||
rel, _ := filepath.Rel(dir, p)
|
||||
found = append(found, filepath.ToSlash(rel))
|
||||
}
|
||||
return nil
|
||||
})
|
||||
for _, f := range found {
|
||||
if strings.Contains(f, "..") {
|
||||
t.Fatalf("path escaped the store: %s", f)
|
||||
}
|
||||
}
|
||||
if len(found) == 0 {
|
||||
t.Fatal("nothing written at all")
|
||||
}
|
||||
}
|
||||
|
||||
func TestListIsPerDevice(t *testing.T) {
|
||||
s, _ := open(t, DefaultPolicy())
|
||||
if _, err := s.Put("devA", doc("run-a", AnonFull)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := s.Put("devB", doc("run-b", AnonFull)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := s.List("devA"); len(got) != 1 || got[0].ID != "run-a" {
|
||||
t.Fatalf("devA sees %+v", got)
|
||||
}
|
||||
if _, err := s.Get("devA", "run-b"); err == nil {
|
||||
t.Fatal("devA could read devB's run")
|
||||
}
|
||||
}
|
||||
|
||||
func TestMetaSummarisesTheDocument(t *testing.T) {
|
||||
s, _ := open(t, DefaultPolicy())
|
||||
m, err := s.Put("dev1", doc("run-1", AnonBalanced))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if m.FindingCount != 2 || m.Verdict != "warn" || m.Anonymization != AnonBalanced {
|
||||
t.Fatalf("meta not extracted: %+v", m)
|
||||
}
|
||||
if m.StartedAt != "2026-08-01T10:00:00Z" {
|
||||
t.Fatalf("started_at = %q", m.StartedAt)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMalformedRejected(t *testing.T) {
|
||||
s, _ := open(t, DefaultPolicy())
|
||||
for _, body := range [][]byte{[]byte("not json"), []byte(`{"run":{}}`), []byte(`{}`)} {
|
||||
if _, err := s.Put("dev1", body); !errors.Is(err, ErrMalformed) {
|
||||
t.Fatalf("body %q: want ErrMalformed, got %v", body, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user