apps\iot\src\test\java\cn\nfarms\iot\ProtocolTest.kt¶
源码路径:
apps\iot\src\test\java\cn\nfarms\iot\ProtocolTest.kt· 177 行
package cn.nfarms.iot
import cn.nfarms.iot.ota.FirmwareFactory
import cn.nfarms.iot.ota.OtaVerdict
import cn.nfarms.iot.ota.OtaVerifier
import cn.nfarms.iot.protocol.Cmd
import cn.nfarms.iot.protocol.Codec
import cn.nfarms.iot.protocol.Frame
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertNotNull
import org.junit.Assert.assertTrue
import org.junit.Test
/**
* 协议与升级包验证。
*
* 这些是最值得脱离真机测的部分:**一个 CRC 算错,真机上表现是"偶尔没反应"**,
* 那种 bug 靠连真机调试要花掉半天。
*/
class ProtocolTest {
// ---------------------------------------------------------------- 帧
@Test
fun frame_roundtrip() {
val payload = byteArrayOf(1, 2, 3, 4)
val bytes = Frame.encode(Cmd.SetPower, payload)
assertEquals("帧长 = 5 + payload", 9, bytes.size)
assertEquals(0xAA, bytes[0].toInt() and 0xFF)
assertEquals(0x55, bytes[1].toInt() and 0xFF)
assertEquals(Cmd.SetPower.code, bytes[2].toInt() and 0xFF)
val res = Frame.decode(bytes)
assertEquals(1, res.frames.size)
assertTrue(res.frames[0].ok)
assertEquals(Cmd.SetPower, res.frames[0].cmd)
assertTrue(payload.contentEquals(res.frames[0].payload))
assertEquals(0, res.rest.size)
}
@Test
fun bad_crc_is_reported_not_thrown() {
val bytes = Frame.encode(Cmd.ReadStatus)
bytes[bytes.size - 1] = (bytes[bytes.size - 1].toInt() xor 0xFF).toByte()
val res = Frame.decode(bytes)
assertEquals(1, res.frames.size)
assertFalse("CRC 坏了必须标成无效而不是抛异常", res.frames[0].ok)
}
/**
* 半帧 + 粘包:BLE 通知会这样送达。
*
* 半帧要**留着**(rest),不能丢 —— 丢了这一帧就永远拼不回来了。
*/
@Test
fun partial_frame_is_kept_and_two_frames_in_one_chunk_both_decode() {
val a = Frame.encode(Cmd.Hello, byteArrayOf(1))
val b = Frame.encode(Cmd.Status, byteArrayOf(9, 9, 1, 2))
// 半帧
val half = a.copyOfRange(0, 3)
val res1 = Frame.decode(half)
assertEquals(0, res1.frames.size)
assertEquals(3, res1.rest.size)
// 把剩下的接上,并紧跟第二帧
val merged = res1.rest + a.copyOfRange(3, a.size) + b
val res2 = Frame.decode(merged)
assertEquals("两帧都要解出来", 2, res2.frames.size)
assertEquals(Cmd.Hello, res2.frames[0].cmd)
assertEquals(Cmd.Status, res2.frames[1].cmd)
assertEquals(0, res2.rest.size)
}
/** 帧头之前的垃圾字节要被跳过,不能吞掉后面的合法帧。 */
@Test
fun leading_garbage_is_skipped() {
val good = Frame.encode(Cmd.HelloAck, byteArrayOf(7))
val stream = byteArrayOf(0x00, 0x11, 0x22) + good
val res = Frame.decode(stream)
assertEquals(1, res.frames.size)
assertTrue(res.frames[0].ok)
}
/** CRC-8/ATM 的已知值(用独立实现交叉验证过)。 */
@Test
fun crc8_matches_known_vector() {
// poly 0x07, init 0x00,"123456789" → 0xF4
val data = "123456789".toByteArray(Charsets.US_ASCII)
assertEquals(0xF4, Frame.crc8(data, 0, data.size))
}
// ------------------------------------------------------------ 载荷
@Test
fun status_payload_roundtrip_and_truncation() {
val p = Codec.statusPayload(temperatureX10 = 236, power = true, mode = 2, version = "1.4.2")
val st = Codec.parseStatus(p)
assertNotNull(st)
assertEquals(23.6, st!!.temperatureC, 0.001)
assertTrue(st.power)
assertEquals(2, st.mode)
assertEquals("1.4.2", st.version)
// 版本串超长要被截到 15 字节,而不是让整帧超长
val long = Codec.statusPayload(0, false, 0, "0123456789ABCDEFGHIJ")
assertEquals(4 + 15, long.size)
}
@Test
fun short_status_payload_returns_null_instead_of_crashing() {
assertEquals(null, Codec.parseStatus(byteArrayOf(1, 2)))
assertEquals(null, Codec.parseStatus(ByteArray(0)))
}
// ---------------------------------------------------------------- OTA
@Test
fun ota_rejects_in_exact_order_signature_then_digest_then_target_then_version() {
val fw = FirmwareFactory()
val verifier = OtaVerifier(fw.publicKeyDer())
val good = fw.make("1.5.0", target = "NF-Demo-01")
assertEquals(OtaVerdict.Ok, verifier.verify(good, "1.4.2", "NF-Demo-01"))
// 1) 换密钥签 → 签名阶段拒绝
val wrongKey = fw.signedByOtherKey(good)
val v1 = verifier.verify(wrongKey, "1.4.2", "NF-Demo-01") as OtaVerdict.Rejected
assertEquals(OtaVerdict.Stage.Signature, v1.stage)
// 2) 改内容但不动 sha256/sig → 签名覆盖的串没变,签名仍然有效,
// 于是必须由**摘要**这一步抓住它(这正是两段检查都要有的原因)
val tamperedSha = good.copy(
payloadB64 = java.util.Base64.getEncoder().encodeToString(
java.util.Base64.getDecoder().decode(good.payloadB64).also { it[100] = (it[100] + 1).toByte() },
),
)
val v2 = verifier.verify(tamperedSha, "1.4.2", "NF-Demo-01") as OtaVerdict.Rejected
assertEquals(OtaVerdict.Stage.Digest, v2.stage)
// 2b) 改内容**并且**把 sha256 也改成新值(但没有重新签名)→ 签名阶段挂
val shaOfTampered = cn.nfarms.core.secure.Crypto.sha256Hex(
java.util.Base64.getDecoder().decode(tamperedSha.payloadB64),
)
val resignMissing = tamperedSha.copy(sha256 = shaOfTampered)
val v2b = verifier.verify(resignMissing, "1.4.2", "NF-Demo-01") as OtaVerdict.Rejected
assertEquals("签名覆盖了 sha256,改了它签名就必须失效", OtaVerdict.Stage.Signature, v2b.stage)
// 3) 机型不符
val wrongTarget = fw.make("1.5.0", target = "NF-Other")
val v3 = verifier.verify(wrongTarget, "1.4.2", "NF-Demo-01") as OtaVerdict.Rejected
assertEquals(OtaVerdict.Stage.Target, v3.stage)
// 4) 版本倒退
val down = fw.make("1.0.0")
val v4 = verifier.verify(down, "1.4.2", "NF-Demo-01") as OtaVerdict.Rejected
assertEquals(OtaVerdict.Stage.Version, v4.stage)
}
/**
* 版本比较必须按**数字段**,不能按字符串。
*
* `"1.4.10" < "1.4.9"` 在字符串序下成立,而那会让一个合法升级被拒。
*/
@Test
fun version_comparison_is_numeric_not_lexicographic() {
val fw = FirmwareFactory()
val verifier = OtaVerifier(fw.publicKeyDer())
val pkg = fw.make("1.4.10")
assertEquals(
OtaVerdict.Ok,
verifier.verify(pkg, "1.4.9", "NF-Demo-01"),
)
}
}