diff --git a/src/dapps/dapps.core.tests/MeshCoreCompressionVersionTests.cs b/src/dapps/dapps.core.tests/MeshCoreCompressionVersionTests.cs new file mode 100644 index 0000000..b86e871 --- /dev/null +++ b/src/dapps/dapps.core.tests/MeshCoreCompressionVersionTests.cs @@ -0,0 +1,115 @@ +using System.Text; +using AwesomeAssertions; +using dapps.client.Backhaul; +using dapps.meshcore; +using Xunit; + +namespace dapps.core.tests; + +/// +/// Versioned shared dictionary (#23): each compressed MeshCore frame stamps the +/// dictionary version it was produced with, so a receiver decompresses with the +/// matching dictionary and safely DROPS a frame from a dictionary version it doesn't +/// hold (instead of feeding zstd a mismatched dictionary and delivering corruption). +/// +public sealed class MeshCoreCompressionVersionTests +{ + private static BackhaulMessage Sample(string payload) => new( + Id: "ab12cd3", Destination: "GB7ABC-1", Salt: 42, Ttl: 1800, + Payload: Encoding.UTF8.GetBytes(payload), + Originator: "M0LTE-7", LinkSourceCallsign: "M0LTE-7"); + + // A short payload compresses to a single fragment, so there's exactly one frame to + // inspect / mutate. + private const string ShortPayload = "GM all de M0LTE, 73"; + + [Fact] + public void CompressedFrame_StampsCurrentDictionaryVersion() + { + var frames = new MeshCoreChannelTransport().ToFrames(Sample(ShortPayload), DappsCompression.Mode.ZstdDict); + + frames.Should().ContainSingle("a short payload is one fragment"); + var frame = frames[0]; + (frame[0] & 1).Should().Be(1, "the compressed flag must be set"); + frame[1].Should().Be(DappsCompression.CurrentDictionaryVersion, "byte1 carries the dictionary version"); + } + + [Fact] + public void UncompressedFrame_HasNoVersionByte() + { + // Uncompressed frames stay byte-identical to the pre-versioning format: no + // version byte, fragment begins at offset 1. + var frames = new MeshCoreChannelTransport().ToFrames(Sample(ShortPayload), DappsCompression.Mode.None); + + var frame = frames[0]; + (frame[0] & 1).Should().Be(0, "the compressed flag must be clear"); + + var rx = new MeshCoreChannelTransport(); + BackhaulMessage? got = null; + foreach (var f in frames) + { + var r = rx.Ingest(f, DateTime.UtcNow); + if (r.Kind == MeshCoreChannelTransport.Kind.BackhaulComplete) got = r.Message; + } + got.Should().NotBeNull(); + Encoding.UTF8.GetString(got!.Payload).Should().Be(ShortPayload); + } + + [Fact] + public void CurrentVersion_RoundTrips() + { + var original = Sample(ShortPayload); + var frames = new MeshCoreChannelTransport().ToFrames(original, DappsCompression.Mode.ZstdDict); + + var rx = new MeshCoreChannelTransport(); + BackhaulMessage? got = null; + foreach (var f in frames) + { + var r = rx.Ingest(f, DateTime.UtcNow); + if (r.Kind == MeshCoreChannelTransport.Kind.BackhaulComplete) got = r.Message; + } + got.Should().NotBeNull(); + Encoding.UTF8.GetString(got!.Payload).Should().Be(ShortPayload); + } + + [Fact] + public void UnknownDictionaryVersion_IsDroppedAsUnsupported_NotDelivered() + { + var frames = new MeshCoreChannelTransport().ToFrames(Sample(ShortPayload), DappsCompression.Mode.ZstdDict); + frames.Should().ContainSingle(); + + // Simulate a peer on a future dictionary: same compressed body, unknown version. + const byte futureVersion = 99; + DappsCompression.IsKnownVersion(futureVersion).Should().BeFalse("test premise"); + frames[0][1] = futureVersion; + + var rx = new MeshCoreChannelTransport(); + var results = frames.Select(f => rx.Ingest(f, DateTime.UtcNow)).ToList(); + + results.Should().Contain(r => r.Kind == MeshCoreChannelTransport.Kind.Unsupported, + "a frame from an unknown dictionary version must be flagged Unsupported"); + results.Should().NotContain(r => r.Kind == MeshCoreChannelTransport.Kind.BackhaulComplete, + "it must never be decoded/delivered with the wrong dictionary"); + } + + [Fact] + public void Decompress_KnownVersion_Recovers_UnknownVersion_Throws() + { + var payload = Encoding.UTF8.GetBytes("the quick brown fox de M0LTE 73"); + var compressed = DappsCompression.Compress(DappsCompression.Mode.ZstdDict, payload); + + DappsCompression.Decompress(DappsCompression.CurrentDictionaryVersion, compressed) + .Should().Equal(payload); + + var act = () => DappsCompression.Decompress(200, compressed); + act.Should().Throw(); + } + + [Fact] + public void IsKnownVersion_TrueForCurrent_FalseForOthers() + { + DappsCompression.IsKnownVersion(DappsCompression.CurrentDictionaryVersion).Should().BeTrue(); + DappsCompression.IsKnownVersion(0).Should().BeFalse(); + DappsCompression.IsKnownVersion(255).Should().BeFalse(); + } +} diff --git a/src/dapps/dapps.meshcore/DappsCompression.cs b/src/dapps/dapps.meshcore/DappsCompression.cs index 356961e..ae617e8 100644 --- a/src/dapps/dapps.meshcore/DappsCompression.cs +++ b/src/dapps/dapps.meshcore/DappsCompression.cs @@ -11,37 +11,58 @@ namespace dapps.meshcore; /// collapses most messages into a single LoRa packet. /// /// The dictionary is built deterministically from a fixed sample corpus so every -/// node running the same build derives byte-identical dictionary bytes. Version 1. -/// A production build should ship a versioned dictionary blob negotiated by id -/// (#154 / dictionary-versioning) rather than rebuilding from code. +/// node running the same build derives byte-identical dictionary bytes. +/// +/// Versioning (#23): each compressed frame carries the dictionary version it was +/// produced with (see ), and a node keeps a +/// registry of every dictionary version it knows. A sender compresses with +/// ; a receiver decompresses with the version +/// named in the frame, or drops the frame if it doesn't hold that dictionary - so a +/// mixed-version fleet degrades to "can't read newer peers yet" instead of silently +/// corrupting payloads. When the corpus is retrained, add a new entry to +/// and bump , but +/// NEVER mutate an existing version's bytes - old peers and in-flight frames still +/// reference it. /// public static class DappsCompression { public enum Mode { None, ZstdDict } - /// Dictionary version - both ends must agree. Carried implicitly by - /// the build today; negotiate explicitly later. - public const byte DictionaryVersion = 1; + /// The dictionary version this build compresses outbound frames with, and + /// stamps into each compressed frame. The highest version in . + public const byte CurrentDictionaryVersion = 1; + + /// Every dictionary version this build can DECOMPRESS with, keyed by version. + /// Retains superseded versions so a node can still read peers that haven't upgraded. + private static readonly IReadOnlyDictionary Dictionaries = + new Dictionary { [1] = BuildDictV1() }; - private static readonly byte[] Dict = BuildDict(); + /// True if this build holds the dictionary for and + /// can therefore decompress a frame stamped with it. + public static bool IsKnownVersion(byte version) => Dictionaries.ContainsKey(version); public static byte[] Compress(Mode mode, byte[] data) { if (mode == Mode.None) return data; using var c = new ZstdSharp.Compressor(19); - c.LoadDictionary(Dict); + c.LoadDictionary(Dictionaries[CurrentDictionaryVersion]); return c.Wrap(data).ToArray(); } - public static byte[] Decompress(Mode mode, byte[] data) + /// Decompress a frame that was compressed with dictionary . + /// Throws if this build doesn't hold that dictionary - + /// callers must treat that as an undecodable frame, never feed it to a mismatched + /// dictionary (which yields garbage or throws deeper). + public static byte[] Decompress(byte version, byte[] data) { - if (mode == Mode.None) return data; + if (!Dictionaries.TryGetValue(version, out var dict)) + throw new NotSupportedException($"unknown compression dictionary version {version}"); using var d = new ZstdSharp.Decompressor(); - d.LoadDictionary(Dict); + d.LoadDictionary(dict); return d.Unwrap(data).ToArray(); } - private static byte[] BuildDict() + private static byte[] BuildDictV1() { using var ms = new MemoryStream(); foreach (var m in SampleCorpus()) diff --git a/src/dapps/dapps.meshcore/MeshCoreChannelTransport.cs b/src/dapps/dapps.meshcore/MeshCoreChannelTransport.cs index 30add83..9e6e768 100644 --- a/src/dapps/dapps.meshcore/MeshCoreChannelTransport.cs +++ b/src/dapps/dapps.meshcore/MeshCoreChannelTransport.cs @@ -8,21 +8,27 @@ namespace dapps.meshcore; /// using the binary channel-data path. Reuses the real DAPPS codec + packetiser. /// /// send: Encode → (optional) compress → Packetiser.Split → per fragment prepend a -/// 1-byte header and emit as one channel-data datagram. +/// header and emit as one channel-data datagram. /// recv: strip header → Reassembler → (decompress) → Decode. /// -/// Frame header byte: bit0 = compressed, bits1-7 = rolling nonce. The nonce is -/// essential - the binary path has no on-air timestamp, so byte-identical frames -/// share a packet hash and the mesh hasSeen table would drop the duplicate. +/// Frame header: byte0 = flags (bit0 = compressed, bits1-7 = rolling nonce). The nonce +/// is essential - the binary path has no on-air timestamp, so byte-identical frames +/// share a packet hash and the mesh hasSeen table would drop the duplicate. When bit0 +/// is set, byte1 is the compression dictionary VERSION (#23) and the fragment follows +/// at offset 2; when clear, the fragment follows at offset 1 (uncompressed frames are +/// byte-identical to the pre-versioning format). The version lets a receiver pick the +/// matching dictionary and drop frames from a dictionary it doesn't hold rather than +/// decompress with the wrong one. /// public sealed class MeshCoreChannelTransport { /// Fragment size incl. the 13-byte Packetiser header. The channel-data - /// payload = 1 (our header) + fragment, capped at the firmware's 165 B limit. + /// payload = our header (1 B, or 2 B when compressed) + fragment; worst case + /// 2 + 160 = 162, inside the firmware's 165 B limit. public const int Mtu = 160; private readonly Reassembler _reassembler = new(); - private readonly Dictionary _compressed = new(); + private readonly Dictionary _compressed = new(); private readonly object _nonceLock = new(); private byte _nonce; @@ -44,38 +50,62 @@ public IReadOnlyList ToFrames(BackhaulMessage message, DappsCompression. hdr = (byte)((_nonce << 1) | (comp ? 1 : 0)); _nonce = (byte)((_nonce + 1) & 0x7F); } - var frame = new byte[1 + f.Length]; - frame[0] = hdr; - f.CopyTo(frame, 1); + // Compressed frames carry the dictionary version in byte1 so the receiver + // decompresses with the matching dictionary; uncompressed frames omit it. + byte[] frame; + if (comp) + { + frame = new byte[2 + f.Length]; + frame[0] = hdr; + frame[1] = DappsCompression.CurrentDictionaryVersion; + f.CopyTo(frame, 2); + } + else + { + frame = new byte[1 + f.Length]; + frame[0] = hdr; + f.CopyTo(frame, 1); + } frames.Add(frame); } return frames; } - public enum Kind { FragmentPartial, BackhaulComplete, Bad } + /// = a fully-reassembled compressed message whose + /// dictionary version this build doesn't hold; safe to drop, distinct from malformed + /// () so the caller can log "peer on a newer dictionary". + public enum Kind { FragmentPartial, BackhaulComplete, Bad, Unsupported } public readonly record struct Result(Kind Kind, BackhaulMessage? Message, FragmentHeader? Header); /// Feed one received channel-data payload. public Result Ingest(byte[] dataPayload, DateTime now) { - if (dataPayload.Length < 1 + Packetiser.HeaderLength) return new Result(Kind.Bad, null, null); + if (dataPayload.Length < 1) return new Result(Kind.Bad, null, null); bool comp = (dataPayload[0] & 1) != 0; - var fragment = dataPayload[1..]; + + // Compressed frames carry a version byte before the fragment (see ToFrames). + int fragOffset = comp ? 2 : 1; + byte version = comp && dataPayload.Length >= 2 ? dataPayload[1] : (byte)0; + if (dataPayload.Length < fragOffset + Packetiser.HeaderLength) return new Result(Kind.Bad, null, null); + var fragment = dataPayload[fragOffset..]; FragmentHeader header; try { header = Packetiser.ParseHeader(fragment); } catch (InvalidDataException) { return new Result(Kind.Bad, null, null); } - _compressed[header.Id] = (comp, now); + _compressed[header.Id] = (comp, version, now); var assembled = _reassembler.Accept(fragment, now); if (assembled is null) return new Result(Kind.FragmentPartial, null, header); - var compressed = _compressed.TryGetValue(header.Id, out var c) && c.comp; + var meta = _compressed.TryGetValue(header.Id, out var c) ? c : (comp, version, now); _compressed.Remove(header.Id); + + if (meta.comp && !DappsCompression.IsKnownVersion(meta.version)) + return new Result(Kind.Unsupported, null, header); try { - var body = compressed ? DappsCompression.Decompress(DappsCompression.Mode.ZstdDict, assembled) : assembled; + var body = meta.comp ? DappsCompression.Decompress(meta.version, assembled) : assembled; return new Result(Kind.BackhaulComplete, BackhaulMessageCodec.Decode(body), header); } catch (Exception) diff --git a/src/dapps/dapps.meshcore/MeshCoreInbound.cs b/src/dapps/dapps.meshcore/MeshCoreInbound.cs index ccf2783..ba75d7f 100644 --- a/src/dapps/dapps.meshcore/MeshCoreInbound.cs +++ b/src/dapps/dapps.meshcore/MeshCoreInbound.cs @@ -84,6 +84,16 @@ public async Task RunAsync(CancellationToken ct) { var now = DateTime.UtcNow; var r = _rx.Ingest(d.Payload, now); + if (r.Kind == MeshCoreChannelTransport.Kind.Unsupported) + { + // A peer compressed with a dictionary version we don't hold (#23). + // We can't read it and won't ACK it; surface the version gap so an + // operator knows this node needs upgrading. + _log.LogWarning( + "MeshCore: dropped {0} - compressed with a dictionary version this build doesn't have (upgrade needed?)", + r.Header?.Id); + continue; + } if (r.Kind != MeshCoreChannelTransport.Kind.BackhaulComplete) continue; var msg = r.Message!;