diff --git a/RNS/Interfaces/util/HDLC.py b/RNS/Interfaces/util/HDLC.py
new file mode 100644
index 00000000..ab0f1735
--- /dev/null
+++ b/RNS/Interfaces/util/HDLC.py
@@ -0,0 +1,106 @@
+# Reticulum License
+#
+# Copyright (c) 2026 Mark Qvist
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# - The Software shall not be used in any kind of system which includes amongst
+#   its functions the ability to purposefully do harm to human beings.
+#
+# - The Software shall not be used, directly or indirectly, in the creation of
+#   an artificial intelligence, machine learning or language model training
+#   dataset, including but not limited to any use that contributes to the
+#   training or development of such a model or algorithm.
+#
+# - The above copyright notice and this permission notice shall be included in
+#   all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+class HDLC():
+    FLAG       = 0x7E
+    ESC        = 0x7D
+    ESC_MASK   = 0x20
+
+    FLAG_B     = bytes([FLAG])
+    ESC_B      = bytes([ESC])
+    ESCPD_FLAG = bytes([ESC, FLAG ^ ESC_MASK])
+    ESCPD_ESC  = bytes([ESC, ESC  ^ ESC_MASK])
+
+    @staticmethod
+    def escape(data):
+        data = data.replace(HDLC.ESC_B,  HDLC.ESCPD_ESC)
+        data = data.replace(HDLC.FLAG_B, HDLC.ESCPD_FLAG)
+        return data
+
+    @staticmethod
+    def unescape(frame):
+        frame = frame.replace(HDLC.ESCPD_FLAG, HDLC.FLAG_B)
+        frame = frame.replace(HDLC.ESCPD_ESC,  HDLC.ESC_B)
+        return frame
+
+class ReceiveBuffer():
+    def __init__(self, mtu, min_frame_len, max_frame_len=None, on_frame=None, on_invalid=None):
+        self._mtu           = mtu
+        self._min_frame_len = min_frame_len
+        self._max_frame_len = max_frame_len
+        self._on_frame      = on_frame
+        self._on_invalid    = on_invalid
+        self._buf           = bytearray()
+        self._off           = 0
+
+    def mtu(self):
+        return self._mtu() if callable(self._mtu) else self._mtu
+
+    def max(self):
+        m = self._max_frame_len
+        return m() if callable(m) else m
+
+    def feed(self, data):
+        if not data: return
+        self._buf.extend(data)
+        buf = self._buf
+
+        while True:
+            frame_start = buf.find(HDLC.FLAG, self._off)
+            if frame_start == -1:
+                self.reset()
+                return
+
+            frame_end = buf.find(HDLC.FLAG, frame_start+1)
+            if frame_end == -1:
+                if len(buf)-self._off > self.mtu()*2: self.reset()
+                return
+
+            frame = HDLC.unescape(bytes(buf[frame_start+1:frame_end]))
+            frame_len = len(frame)
+            if frame_len > 0:
+                max_len = self.max()
+                if frame_len > self._min_frame_len and (max_len is None or frame_len <= max_len):
+                    if self._on_frame is not None: self._on_frame(frame)
+                elif self._on_invalid is not None: self._on_invalid(frame_len)
+
+            self._off = frame_end
+            if self._off > 0 and self._off >= len(buf) // 2:
+                del buf[:self._off]
+                self._off = 0
+
+    # Drops any partial frames from the buffer and
+    # resets assembly offset
+    def reset(self):
+        self._buf.clear()
+        self._off = 0
+
+    # Buffered bytes currently unconsumed
+    def __len__(self): return len(self._buf)-self._off
diff --git a/tests/hdlc.py b/tests/hdlc.py
new file mode 100644
index 00000000..6352341d
--- /dev/null
+++ b/tests/hdlc.py
@@ -0,0 +1,266 @@
+import unittest
+
+import os
+import time
+import random
+
+from RNS.Interfaces.util.HDLC import HDLC, ReceiveBuffer
+
+SEED    = 0x52EC0FF
+MIN_LEN = 19 # HEADER_MINSIZE
+MTU     = 1048576
+
+def framed(payload): return HDLC.FLAG_B + HDLC.escape(payload) + HDLC.FLAG_B
+
+def rand_payload(rng, size):
+    p = bytearray(os.urandom(size))
+    for off in range(0, max(size, 1), max(size // 16, 1)):
+        p[off] = 0x7E if (off % 2) else 0x7D
+    return bytes(p)
+
+# Original, pre-refactor HDLC handler algorithm
+def legacy_feed(state, data, mtu, min_len, max_len):
+    if len(data) > 0:
+        state["buf"] += data
+        flags_remaining = True
+        while flags_remaining:
+            frame_start = state["buf"].find(HDLC.FLAG)
+            if frame_start != -1:
+                frame_end = state["buf"].find(HDLC.FLAG, frame_start+1)
+                if frame_end != -1:
+                    frame = state["buf"][frame_start+1:frame_end]
+                    frame = frame.replace(bytes([HDLC.ESC, HDLC.FLAG ^ HDLC.ESC_MASK]), bytes([HDLC.FLAG]))
+                    frame = frame.replace(bytes([HDLC.ESC, HDLC.ESC  ^ HDLC.ESC_MASK]), bytes([HDLC.ESC]))
+                    frame_len = len(frame)
+                    if frame_len != 0:
+                        if check_frame_len(frame_len, min_len, max_len): state["frames"].append(frame)
+                        else:                                            state["invalids"].append(frame_len)
+
+                    state["buf"] = state["buf"][frame_end:]
+
+                else:
+                    if len(state["buf"]) > mtu*2: state["buf"] = b""
+                    flags_remaining = False
+            else:
+                state["buf"] = b""
+                flags_remaining = False
+
+def check_frame_len(frame_len, min_len, max_len):
+    if   frame_len <= min_len: return False
+    elif max_len == None:      return True
+    elif frame_len >  max_len: return False
+    else:                      return True
+
+class Harness:
+    def __init__(self, mtu=MTU, min_len=MIN_LEN, max_len=None, track_invalid=True):
+        self.frames = []
+        self.invalids = []
+        self.rb = ReceiveBuffer(mtu, min_len, max_len, on_frame=self.frames.append,
+                                on_invalid=self.invalids.append if track_invalid else None)
+    def unconsumed(self):
+        return bytes(self.rb._buf[self.rb._off:])
+
+class TestReceiveBuffer(unittest.TestCase):
+    # Legacy parity, also checks that the closing flag of
+    # the last consumed frame is retained as the start
+    # marker for the next unframing.
+    def test_01_single_frame_roundtrip(self):
+        print("")
+        rng = random.Random(SEED)
+        for size in (20, 200, 4096, 262144):
+            p = rand_payload(rng, size)
+            h = Harness()
+            h.rb.feed(framed(p))
+            self.assertEqual(h.frames, [p])
+            self.assertEqual(h.invalids, [])
+            self.assertEqual(h.unconsumed(), HDLC.FLAG_B)
+            self.assertEqual(len(h.rb), 1)
+
+    # Test frames split across arbitrary feed boundaries
+    def test_02_split_chunks_straddle(self):
+        print("")
+        rng = random.Random(SEED)
+        payloads = [rand_payload(rng, rng.choice((20, 200, 1024, 16384))) for _ in range(200)]
+        stream = b"".join(framed(p) for p in payloads)
+
+        h = Harness()
+        # One-byte-per-feed pathological case
+        for byte in stream: h.rb.feed(bytes([byte]))
+
+        self.assertEqual(h.frames, payloads)
+        self.assertEqual(len(h.rb), 1) # trailing flag
+
+    # Test exactness of escape-heavy payloads
+    def test_03_escape_roundtrip(self):
+        print("")
+        rng = random.Random(SEED ^ 0xE5C)
+        payloads = []
+        for _ in range(100):
+            size = rng.choice((64, 512, 8192))
+            p = bytearray(os.urandom(size))
+            # Dense 0x7E/0x7D injection
+            for i in range(0, size, 3): p[i] = 0x7E if (i % 2) else 0x7D
+            payloads.append(bytes(p))
+
+        h = Harness()
+        for p in payloads: h.rb.feed(framed(p))
+        self.assertEqual(h.frames, payloads)
+
+    # Empty frames must skip silently
+    def test_04_consecutive_flags_skipped(self):
+        print("")
+        rng = random.Random(SEED)
+        p1 = rand_payload(rng, 100)
+        p2 = rand_payload(rng, 100)
+        h = Harness()
+        h.rb.feed(framed(p1) + HDLC.FLAG_B + HDLC.FLAG_B + framed(p2))
+        self.assertEqual(h.frames, [p1, p2])
+        self.assertEqual(h.invalids, [])
+
+    # Test that invalid sized frames report correctly
+    def test_05_size_range(self):
+        print("")
+        h = Harness(min_len=19, max_len=100)
+        h.rb.feed(framed(b"x" * 5))   # frame len 5   -> too short
+        h.rb.feed(framed(b"x" * 19))  # frame len 19  -> not > 19, which is bad
+        h.rb.feed(framed(b"x" * 100)) # frame len 100 -> within range
+        h.rb.feed(framed(b"x" * 101)) # frame len 101 -> too long
+        self.assertEqual(h.frames, [b"x" * 100])
+        self.assertEqual(h.invalids, [5, 19, 101])
+
+        # Without an on_invalid hook, out-of-range frames are dropped
+        h2 = Harness(min_len=19, max_len=100, track_invalid=False)
+        h2.rb.feed(framed(b"x" * 5))
+        h2.rb.feed(framed(b"x" * 50))
+        h2.rb.feed(framed(b"x" * 101))
+        self.assertEqual(h2.frames, [b"x" * 50])
+        self.assertEqual(h2.invalids, [])
+
+    # A flag without a closing flag is retained up to the
+    # overflow bound, beyond which the buffer must reset
+    def test_06_partial_and_overflow_reset(self):
+        print("")
+        h = Harness(mtu=1024)
+        # Partial frame within bound must retained
+        h.rb.feed(b"\x7e" + b"Y" * 100)
+        self.assertEqual(len(h.rb), 101)
+        self.assertEqual(h.frames, [])
+        # ... and completing it delivers the frame
+        h.rb.feed(b"Z" * 40 + b"\x7e")
+        self.assertEqual(h.frames, [b"Y" * 100 + b"Z" * 40])
+        self.assertEqual(len(h.rb), 1)
+
+        # Partial frame past the overflow bound must drop and reset
+        h2 = Harness(mtu=1024)
+        h2.rb.feed(b"\x7e" + b"Y" * 5000)
+        self.assertEqual(len(h2.rb), 0)
+        # Buffer must be usable after the reset
+        h2.rb.feed(framed(b"x" * 100))
+        self.assertEqual(h2.frames, [b"x" * 100])
+
+    # Test that input without any flag is discarded immediately
+    def test_07_garbage_reset(self):
+        print("")
+        h = Harness()
+        h.rb.feed(b"\x00" * 100000 + b"\x01" * 100000)
+        self.assertEqual(len(h.rb), 0)
+        self.assertEqual(h.frames, [])
+        h.rb.feed(framed(b"x" * 100))
+        self.assertEqual(h.frames, [b"x" * 100])
+
+    # Random stream parity checks
+    def test_08_parity_random_streams(self):
+        print("")
+        rng = random.Random(SEED)
+        verified = 0
+
+        def inert(size): return bytes(rng.randrange(1, 0x7D) for _ in range(size))
+
+        for round in range(20):
+            stream = bytearray()
+            intended = []
+            for _ in range(rng.randrange(5, 60)):
+                kind = rng.randrange(5)
+                if   kind == 0: stream += inert(rng.randrange(40, 300))           # Garbage
+                elif kind == 1: stream += b"\x7e" + inert(rng.randrange(50, 400)) # Unclosed partial
+                else:
+                    p = rand_payload(rng, rng.randrange(20, 8192))
+                    intended.append(p)
+                    stream += framed(p)
+
+            legacy = {"buf": b"", "frames": [], "invalids": []}
+            mine = Harness()
+            pos = 0
+            while pos < len(stream):
+                step = rng.randrange(1, 4096)
+                chunk = bytes(stream[pos:pos + step])
+                pos += step
+                legacy_feed(legacy, chunk, MTU, MIN_LEN, None)
+                mine.rb.feed(chunk)
+                # Behavioural and internal-state parity after every chunk
+                self.assertEqual(mine.frames, legacy["frames"])
+                self.assertEqual(mine.invalids, legacy["invalids"])
+                self.assertEqual(mine.unconsumed(), legacy["buf"])
+
+            # Full-stream parity and intended-payload presence.
+            self.assertEqual(mine.frames, legacy["frames"])
+            self.assertEqual(mine.unconsumed(), legacy["buf"])
+            for p in intended: self.assertGreaterEqual(mine.frames.count(p), intended.count(p))
+            verified += 1
+
+        print(f"  Random-stream parity: {verified} streams verified")
+
+    # Small-frame flood parity
+    def test_09_parity_flood(self):
+        print("")
+        rng = random.Random(SEED)
+        payloads = [rand_payload(rng, 200) for _ in range(20000)]
+        stream = b"".join(framed(p) for p in payloads)
+
+        legacy = {"buf": b"", "frames": [], "invalids": []}
+        mine = Harness()
+        t0 = time.perf_counter()
+        for i in range(0, len(stream), MTU):
+            chunk = stream[i:i + MTU]
+            legacy_feed(legacy, chunk, MTU, MIN_LEN, None)
+            mine.rb.feed(chunk)
+        t_elapsed = time.perf_counter() - t0
+
+        self.assertEqual(mine.frames, legacy["frames"])
+        self.assertEqual(mine.frames, payloads)
+        self.assertEqual(mine.unconsumed(), legacy["buf"])
+        self.assertLess(t_elapsed, 10.0)
+        print(f"  20k x 200B flood ({len(stream)/1e6:.1f} MB) parity-verified in {t_elapsed*1000:.1f} ms")
+
+    # Large frames in small chunks
+    def test_10_parity_large_frames(self):
+        print("")
+        rng = random.Random(SEED ^ 0x1AF)
+        payloads = [rand_payload(rng, 262144) for _ in range(8)]
+        stream = b"".join(framed(p) for p in payloads)
+
+        legacy = {"buf": b"", "frames": [], "invalids": []}
+        mine = Harness(mtu=262144)
+        pos = 0
+        while pos < len(stream):
+            chunk = stream[pos:pos + 4096]
+            pos += 4096
+            legacy_feed(legacy, chunk, 262144, MIN_LEN, None)
+            mine.rb.feed(chunk)
+            self.assertEqual(mine.frames, legacy["frames"])
+            self.assertEqual(mine.unconsumed(), legacy["buf"])
+        self.assertEqual(mine.frames, payloads)
+
+    # After a burst is fully consumed, the internal buffer must be compacted
+    def test_11_len_and_compaction(self):
+        print("")
+        rng = random.Random(SEED)
+        h = Harness()
+        for _ in range(1000):
+            h.rb.feed(framed(rand_payload(rng, 200)))
+        self.assertEqual(len(h.rb), 1)      # Trailing flag marker
+        self.assertEqual(len(h.rb._buf), 1) # Fully compacted
+        self.assertEqual(h.rb._off, 0)
+        self.assertEqual(len(h.frames), 1000)
+
+if __name__ == "__main__": unittest.main(verbosity=2)
\ No newline at end of file