fix: strip NUL padding from monitor export and shared HBP fields

Centralize fixed-width NUL/space trimming in domain helpers and use them
for dashboard_state peer fields, OPTIONS parsing, proxy self-service, and
remaining CALLSIGN log lines.

src/adn_server/application/report/payloads.py

@@ -32,6 +32,7 @@ from adn_server.application.routing.helpers import (
     peer_rf_mode,
 )
 from adn_server.domain import int_id
+from adn_server.domain.hbp_protocol import normalize_fixed_width_ascii
 from adn_server.domain.ua_timer import normalize_ua_timer_minutes, ua_timer_is_infinite
 
 REPORT_PROTOCOL = 2
@@ -68,11 +69,8 @@ def _parse_options_kv(options: Any) -> dict[str, str]:
     """Parse RPTO OPTIONS string into upper-case keys (legacy normalisation)."""
     if options is None:
         return {}
-    if isinstance(options, bytes):
-        text = options.decode("utf-8", errors="replace")
-    else:
-        text = str(options)
-    text = text.rstrip("\x00").encode("ascii", "ignore").decode()
+    text = normalize_fixed_width_ascii(options)
+    text = text.encode("ascii", "ignore").decode()
     text = re.sub(r"['\"]", "", text).strip()
     if not text:
         return {}
@@ -262,10 +260,7 @@ def _peer_field_json(value: Any) -> str | None:
     """Sanitize a legacy peer field for JSON (no secrets)."""
     if value is None:
         return None
-    if isinstance(value, bytes):
-        text = value.decode("utf-8", errors="replace").strip()
-        return text or None
-    text = str(value).strip()
+    text = normalize_fixed_width_ascii(value)
     return text or None
 
 
@@ -301,11 +296,7 @@ def _sanitized_peer_options_text(options: Any) -> str | None:
     """RPTO OPTIONS for monitor display (omit ``PASS=`` secrets)."""
     if options is None:
         return None
-    if isinstance(options, bytes):
-        text = options.decode("utf-8", errors="replace")
-    else:
-        text = str(options)
-    text = text.rstrip("\x00").strip()
+    text = normalize_fixed_width_ascii(options)
     if not text:
         return None
     parts: list[str] = []

src/adn_server/domain/hbp_protocol.py

@@ -28,6 +28,10 @@ belong in the domain layer so application use cases can reference them
 without importing infrastructure.
 """
 
+from __future__ import annotations
+
+from typing import Any
+
 # Frame types (bits)
 HBPF_VOICE = 0x0
 HBPF_VOICE_SYNC = 0x1
@@ -41,3 +45,23 @@ PROTO_VER = 5
 
 # Stream timeout (seconds) for contention (legacy const.py)
 STREAM_TO = 0.36
+
+
+def normalize_fixed_width_ascii(value: Any) -> str:
+    """Decode fixed-width HBP ASCII (RPTC field, RPTO body) and strip NUL/space pad.
+
+    MMDVMHost pads with spaces; some bridges (e.g. ipsc2hbp) NUL-pad to width.
+    ``str.strip()`` / ``str.rstrip()`` without ``\\x00`` leave false suffix bytes.
+    """
+    if value is None:
+        return ""
+    if isinstance(value, (bytes, bytearray)):
+        text = bytes(value).decode("utf-8", errors="replace")
+    else:
+        text = str(value)
+    return text.rstrip("\x00 ")
+
+
+def normalize_fixed_width_bytes(value: bytes) -> bytes:
+    """Strip NUL/space padding from a fixed-width RPTO/RPTC byte slice."""
+    return value.rstrip(b"\x00 ")

src/adn_server/infrastructure/options_redaction.py

@@ -5,18 +5,14 @@ from __future__ import annotations
 import re
 from typing import Any
 
+from adn_server.domain.hbp_protocol import normalize_fixed_width_ascii
+
 _PASS_RE = re.compile(r"(?i)(PASS=)[^;]*")
 
 
 def normalize_options_text(options: Any) -> str:
     """Decode OPTIONS and strip fixed-width NUL/space padding (e.g. ipsc2hbp RPTO)."""
-    if options is None:
-        return ""
-    if isinstance(options, (bytes, bytearray)):
-        text = bytes(options).decode("utf-8", errors="replace")
-    else:
-        text = str(options)
-    return text.rstrip("\x00 ")
+    return normalize_fixed_width_ascii(options)
 
 
 def redact_pass_in_options(options: Any) -> str:

src/adn_server/infrastructure/proxy/self_service_bridge.py

@@ -38,6 +38,7 @@ from adn_server.application.ports import (
     ProxySelfServiceStore,
 )
 from adn_server.application.proxy import ProxyUseCases
+from adn_server.domain.hbp_protocol import normalize_fixed_width_ascii, normalize_fixed_width_bytes
 from adn_server.domain.proxy import ClientEndpoint
 from adn_server.domain.value_objects import int_id
 from adn_server.infrastructure.hbp_constants import RPTACK, RPTC, RPTCL, RPTO
@@ -149,7 +150,7 @@ class ProxySelfServiceBridge:
             )
             return
         mode = data[97:98].decode("utf-8", errors="replace") if len(data) >= 98 else "4"
-        callsign = data[8:16].rstrip(b"\x00 ").decode("utf-8", errors="replace")
+        callsign = normalize_fixed_width_ascii(data[8:16])
         self._store.ins_conf(int_id(peer_id), peer_id, callsign, host, mode)
         if peer_id in self._mysql_option_peers:
             self._fetch_options_now(peer_id)
@@ -173,7 +174,7 @@ class ProxySelfServiceBridge:
                 host, port, int_id(peer_id),
             )
             return False
-        payload = data[8:].rstrip(b"\x00")
+        payload = normalize_fixed_width_bytes(data[8:]) if len(data) > 8 else b""
         if payload.upper().startswith(b"PASS="):
             psswd_raw = payload[5:]
             if len(psswd_raw) < 6:

src/adn_server/infrastructure/twisted_adapters/udp_hbp.py

@@ -77,6 +77,7 @@ from ...application.server_voice import all_server_voice_ids
 from ...domain import bytes_3, bytes_4, int_id
 from ...domain.dmr import decode
 from ...domain.dmr.const import LC_OPT
+from ...domain.hbp_protocol import normalize_fixed_width_ascii, normalize_fixed_width_bytes
 from ...domain.mesh_routing import MeshEgress, MeshIngress, PeerMeshConfig
 from ...domain.talker_alias import (
     DMRA_PACKET_LEN,
@@ -149,12 +150,8 @@ def _get_passphrase_bytes(sys_cfg: dict) -> bytes:
 
 
 def _rptc_field_str(raw: bytes) -> str:
-    """Decode a fixed-width RPTC ASCII field (space- or NUL-padded).
-
-    MMDVMHost typically pads with spaces; some bridges (e.g. ipsc2hbp) use NUL.
-    ``str.rstrip()`` alone leaves trailing ``\\x00``, so callsign checks falsely fail.
-    """
-    return raw.decode("utf8", errors="replace").rstrip("\x00 ")
+    """Decode a fixed-width RPTC ASCII field (space- or NUL-padded)."""
+    return normalize_fixed_width_ascii(raw)
 
 
 def _calc_hash(salt_str: bytes, password: bytes) -> bytes:
@@ -979,7 +976,7 @@ class HBPProtocol(DatagramProtocol):
         if mode == "MASTER":
             for _peer in self._peers:
                 self.send_peer(_peer, b"".join([MSTCL, _peer]))
-                logger.info("(%s) De-Registration sent to Peer: %s (%s)", self._system, self._peers[_peer].get("CALLSIGN", b""), self._peers[_peer].get("RADIO_ID", b""))
+                logger.info("(%s) De-Registration sent to Peer: %s (%s)", self._system, _rptc_field_str(self._peers[_peer].get("CALLSIGN", b"")), self._peers[_peer].get("RADIO_ID", b""))
         elif mode == "PEER":
             self.send_master(b"".join([RPTCL, self._config.get("RADIO_ID", b"\x00\x00\x00\x00")]))
             logger.info("(%s) De-Registration sent to Master: %s:%s", self._system, self._config.get("MASTER_SOCKADDR", ("?", "?"))[0], self._config.get("MASTER_SOCKADDR", ("?", "?"))[1])
@@ -1082,7 +1079,7 @@ class HBPProtocol(DatagramProtocol):
         for peer in remove_list:
             logger.info(
                 "(%s) Peer %s (%s) has timed out and is being removed",
-                self._system, self._peers[peer].get("CALLSIGN", b""), self._peers[peer].get("RADIO_ID", b""),
+                self._system, _rptc_field_str(self._peers[peer].get("CALLSIGN", b"")), self._peers[peer].get("RADIO_ID", b""),
             )
             self.transport.write(b"".join([MSTCL, peer]), self._peers[peer]["SOCKADDR"])
             self._remove_peer(peer)
@@ -1505,7 +1502,7 @@ class HBPProtocol(DatagramProtocol):
             if _data[:5] == RPTCL:
                 _peer_id = _data[5:9]
                 if _peer_id in self._peers and self._peers[_peer_id]["CONNECTION"] == "YES" and self._peers[_peer_id]["SOCKADDR"] == _sockaddr:
-                    logger.info("(%s) Peer is closing down: %s (%s)", self._system, self._peers[_peer_id]["CALLSIGN"], int_id(_peer_id))
+                    logger.info("(%s) Peer is closing down: %s (%s)", self._system, _rptc_field_str(self._peers[_peer_id]["CALLSIGN"]), int_id(_peer_id))
                     self.transport.write(b"".join([MSTNAK, _peer_id]), _sockaddr)
                     self._remove_peer(_peer_id)
                     sys_cfg = self._CONFIG.get("SYSTEMS", {}).get(self._system, {})
@@ -1578,7 +1575,7 @@ class HBPProtocol(DatagramProtocol):
             if _peer_id in self._peers and self._peers[_peer_id]["SOCKADDR"] == _sockaddr:
                 _this_peer = self._peers[_peer_id]
                 # RPTO body is fixed-width; bridges may NUL-pad (same as RPTC fields).
-                _this_peer["OPTIONS"] = _data[8:].rstrip(b"\x00 ")
+                _this_peer["OPTIONS"] = normalize_fixed_width_bytes(_data[8:])
                 invalidate_peer_options_cache(_this_peer)
                 self._mark_downlink_index_dirty()
                 self.send_peer(_peer_id, b"".join([RPTACK, _peer_id]))
@@ -1626,7 +1623,7 @@ class HBPProtocol(DatagramProtocol):
                 self._peers[_peer_id]["PINGS_RECEIVED"] += 1
                 self._peers[_peer_id]["LAST_PING"] = time.time()
                 self.send_peer(_peer_id, b"".join([MSTPONG, _peer_id]))
-                logger.log(logging.TRACE if hasattr(logging, "TRACE") else logging.DEBUG, "(%s) Received and answered RPTPING from peer %s (%s)", self._system, self._peers[_peer_id]["CALLSIGN"], int_id(_peer_id))
+                logger.log(logging.TRACE if hasattr(logging, "TRACE") else logging.DEBUG, "(%s) Received and answered RPTPING from peer %s (%s)", self._system, _rptc_field_str(self._peers[_peer_id]["CALLSIGN"]), int_id(_peer_id))
             else:
                 self.transport.write(b"".join([MSTNAK, _peer_id]), _sockaddr)
                 logger.info("(%s) Ping from Radio ID that is not logged in: %s", self._system, int_id(_peer_id))

tests/application/test_report_payloads.py

@@ -323,6 +323,30 @@ def test_build_topology_includes_peer_display_fields() -> None:
     assert peer["slots"] == "2"
 
 
+def test_topology_strips_nul_padded_callsign() -> None:
+    systems = {
+        "MASTER-A": {
+            "MODE": "MASTER",
+            "ENABLED": True,
+            "PEERS": {
+                bytes_3(3120001): {
+                    "CONNECTION": "YES",
+                    "CALLSIGN": b"Bridge\x00\x00",
+                }
+            },
+        }
+    }
+    doc = build_topology(systems, seq=1, ts=1.0)
+    assert doc["systems"][0]["peers"][0]["callsign"] == "Bridge"
+
+
+def test_parse_peer_options_static_strips_nul_padding() -> None:
+    raw = b"TS1=73080;TS2=73081;SINGLE=1;" + b"\x00" * 12
+    ts1, ts2 = parse_peer_options_static(raw)
+    assert ts1 == ["73080"]
+    assert ts2 == ["73081"]
+
+
 def test_hello_connected_system_names_only_live_systems() -> None:
     systems = {
         "SYSTEM-0": {

tests/domain/test_hbp_protocol.py

@@ -0,0 +1,37 @@
+# ADN DMR Peer Server - domain HBP protocol helpers
+#
+# Copyright (C) 2026  Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
+#
+###############################################################################
+#   This program is free software; you can redistribute it and/or modify
+#   it under the terms of the GNU General Public License as published by
+#   the Free Software Foundation; either version 3 of the License, or
+#   (at your option) any later version.
+#
+#   This program is distributed in the hope that it will be useful,
+#   but WITHOUT ANY WARRANTY; without even the implied warranty of
+#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#   GNU General Public License for more details.
+#
+#   You should have received a copy of the GNU General Public License
+#   along with this program; if not, write to the Free Software Foundation,
+#   Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
+###############################################################################
+
+from __future__ import annotations
+
+from adn_server.domain.hbp_protocol import (
+    normalize_fixed_width_ascii,
+    normalize_fixed_width_bytes,
+)
+
+
+def test_normalize_fixed_width_ascii_strips_nul_and_space() -> None:
+    assert normalize_fixed_width_ascii(b"CE5RPY\x00\x00") == "CE5RPY"
+    assert normalize_fixed_width_ascii(b"CE5RPY  ") == "CE5RPY"
+    assert normalize_fixed_width_ascii("Bridge\x00 ") == "Bridge"
+
+
+def test_normalize_fixed_width_bytes_strips_nul_and_space() -> None:
+    raw = b"TS1=730;TS2=73081" + b"\x00" * 8 + b"  "
+    assert normalize_fixed_width_bytes(raw) == b"TS1=730;TS2=73081"