Require authentication before a radio ID changes address

A repeater's registration was displaced by an unauthenticated RPTL: any
packet carrying a known radio ID, from any source address, tore down the
live registration before anything was proven. Radio IDs are public, so
this was enough to knock any repeater off the network with an 8-byte
packet, repeatable at will.

A login claiming an ID that is already registered from a different
sockaddr is now held in _pending_logins and challenged. The incumbent is
removed only after the claimant proves the passphrase; on any failure the
claimant is discarded and the registration is untouched. Recovery for a
genuine NAT remap is unchanged in speed — login, auth, config, one round
trip later.

Liveness is now gated on the address matching too. _check_repeater_timeouts
decides purely on last_ping, which was refreshed by any packet bearing a
known ID before validation, so a stale or spoofed source could keep a dead
entry from ever aging out.

Report reconnects instead of diagnosing them. When a radio ID logs in
again, log how long the previous session lasted as D:HH:MM:SS. Hours means
a NAT lease expired or the device was power cycled; seconds, repeating,
means it cannot stay online — and two addresses alternating is a duplicate
radio ID. The operator reads that in a glance; no thresholds are guessed
at in code. reconnect_report_threshold (default 3600, 0 disables) filters
out routine long sessions. Only reported once a session actually existed,
since connect_time is meaningless before the connected state.

Fix dashboard uptime resetting for repeaters that never dropped. The
dashboard stamped its own event-arrival time as connected_at, but
repeater_connected is also emitted on missed pings, on recovery, and on
state replay when the dashboard reconnects. The server's connect_time now
rides in the payload, with a fallback for older servers.

Logging: ping and pong lines now both carry radio ID and sockaddr, as do
all inbound dispatch lines, so either field can be grepped for a whole
exchange. Fixes an f-string whose conditional sat inside the string
literal, raising AttributeError and surfacing as a generic datagram error
for exactly the wrong-address pings this change is about.

Adds tests/test_id_takeover.py: 31 tests over the auth gate, the login
races, threshold filtering and connect_time. Verified by mutation — each
reverted behavior fails a specific test.

Also documents three unreachable/misleading branches in the dispatch chain
found along the way, in code and in docs/TODO.md. Behavior unchanged.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

config/config_advanced_sample.json

@@ -4,6 +4,7 @@
     "global": {
         "max_missed": 3,
         "timeout_duration": 30,
+        "reconnect_report_threshold": 3600,
         "disable_ipv6": false,
         "bind_ipv4": "0.0.0.0",
         "bind_ipv6": "::",

config/config_sample.json

@@ -10,6 +10,8 @@
         "disable_ipv6": true,
         "max_missed": 3,
         "timeout_duration": 30,
+        "_comment_reconnect_report_threshold": "When a repeater logs in again, the log shows how long its previous session lasted. Sessions that ran longer than this many seconds are routine (NAT lease expiring, power cycle) and are not reported; shorter ones are, because a repeater reconnecting every few seconds cannot stay online. Set to 0 to report every reconnect. Connections themselves are always logged regardless of this setting.",
+        "reconnect_report_threshold": 3600,
         "logging": {
             "file": "logs/hblink.log",
             "console_level": "INFO",

dashboard/server.py

@@ -661,7 +661,12 @@ class EventReceiver:
         if event_type == 'repeater_connected':
             state.repeaters[data['repeater_id']] = {
                 **data,
-                'connected_at': event['timestamp'],
+                # Prefer the server's session start. repeater_connected is also
+                # emitted on missed pings, on recovery, and when this dashboard
+                # reconnects and the server replays state — using the event's
+                # arrival time would reset uptime for repeaters that never dropped.
+                # Fall back for servers older than this field.
+                'connected_at': data.get('connect_time', event['timestamp']),
                 'last_activity': event['timestamp'],
                 'last_ping': data.get('last_ping', event['timestamp']),
                 'missed_pings': data.get('missed_pings', 0),

docs/TODO.md

@@ -86,9 +86,35 @@ This document tracks planned features and enhancements for HBlink4. Items are pr
 
 ---
 
+### 3. Unreachable Dispatch Branches and Misreported Malformed Logins 🟢
+**Status**: Not started — identified and commented in code, August 14, 2026
+**Difficulty**: Low
+**Dependencies**: None (all within `HBProtocol.datagram_received` in `hblink4/hblink.py`)
+**Description**: Three related issues in the inbound packet dispatch chain, found while adding radio IDs to the dispatch log lines. None causes incorrect behavior today — the dead branches simply never run — but they mislead anyone reading the code or the logs. Each site is marked with a `DEAD CODE` or `TODO` comment pointing here.
+
+**The root cause**: `repeater_id` is extracted only for known commands, then `if not repeater_id: return` fires before the dispatch chain. So by the time dispatch runs, `_command` is guaranteed to be one of the known commands — every one of which has its own branch.
+
+**1. Raw 4-byte repeater ID login is unreachable**
+- The `elif len(data) == 4` branch intends to accept a bare radio ID as a login.
+- A bare radio ID matches none of the command constants, so `repeater_id` stays `None` and the packet returns at the guard.
+- **Decide first**: should raw-ID login be supported at all? If yes, the check must move *above* the guard. If no, delete the branch. Do not simply "fix" the reachability without answering that — it would newly accept unauthenticated 4-byte packets as logins.
+
+**2. Final `else: Unknown command received` is unreachable**
+- Genuinely unknown commands are already reported by the `UNKNOWN PACKET TYPE` guard.
+- Safe to delete once (1) is resolved, since the two interact.
+
+**3. Truncated packets of known types are reported as "unknown packet type"**
+- A short RPTL (fewer than 8 bytes) makes `data[4:8]` empty, so `repeater_id` is falsy and the packet is logged as an unknown type.
+- The operator sees `⚠️ UNKNOWN PACKET TYPE` for what is really a malformed login, which sends diagnosis in the wrong direction.
+- **Fix**: distinguish "command not recognized" from "known command, wrong length", and log the latter as a malformed packet naming the expected command and length.
+
+**Note**: fixing (3) is independent of (1) and (2) and is the one with real operator-facing value.
+
+---
+
 ## Low Priority
 
-### 3. Web-Based Configuration UI 🟡
+### 4. Web-Based Configuration UI 🟡
 **Status**: Not started  
 **Difficulty**: Medium  
 **Dependencies**: Dashboard  
@@ -107,4 +133,4 @@ This document tracks planned features and enhancements for HBlink4. Items are pr
 
 ---
 
-**Last Updated**: April 21, 2026
+**Last Updated**: August 14, 2026

docs/configuration.md

@@ -20,6 +20,7 @@ The `global` section contains server-wide settings that control the basic operat
     "global": {
         "max_missed": 3,
         "timeout_duration": 30,
+        "reconnect_report_threshold": 3600,
         "bind_ipv4": "0.0.0.0",
         "port_ipv4": 62031,
         "bind_ipv6": "::",
@@ -44,6 +45,7 @@ The `global` section contains server-wide settings that control the basic operat
 |---------|------|-------------|
 | `max_missed` | number | Maximum consecutive missed pings before disconnecting a repeater (default: 3) |
 | `timeout_duration` | number | Seconds between expected pings from repeaters (default: 30) |
+| `reconnect_report_threshold` | number | When a repeater logs in again, the log reports how long its previous session lasted. Sessions longer than this many seconds are routine and are not reported; shorter ones are. Set to `0` to report every reconnect (default: 3600). Connections are always logged regardless of this setting |
 | `disable_ipv6` | boolean | Bind IPv4 only. The shipped sample config sets this to `true`, which is what most networks want; if the setting is absent altogether the built-in default is `false` |
 | `bind_ipv4` | string | IPv4 address to bind ("0.0.0.0" for all IPv4 interfaces) |
 | `bind_ipv6` | string | IPv6 address to bind ("::" for all IPv6 interfaces) |

hblink4/constants.py

@@ -25,6 +25,7 @@ DMR_DATA_PACKET_LENGTH = 55  # Minimum length of valid DMR data packet
 DMR_PORT = 62031  # Default HomeBrew DMR port
 DEFAULT_PING_TIME = 5.0  # Default ping interval in seconds
 MAX_MISSED_PINGS = 3  # Maximum number of missed pings before disconnect
+PENDING_LOGIN_TIMEOUT = 30.0  # Seconds a login challenging an already-registered ID may go unanswered
 
 # DMR Sync Patterns (48 bits / 6 bytes)
 # These patterns appear in the DMR payload at bytes 20-25 to identify frame types

hblink4/hblink.py

@@ -45,13 +45,14 @@ if __package__ in (None, ""):
 
 from .constants import (
     RPTA, RPTL, RPTK, RPTC, RPTCL, MSTCL, DMRD,
-    MSTNAK, MSTPONG, RPTPING, RPTACK, RPTP, RPTO, DMRA
+    MSTNAK, MSTPONG, RPTPING, RPTACK, RPTP, RPTO, DMRA,
+    PENDING_LOGIN_TIMEOUT
 )
 from .access_control import RepeaterMatcher
 from .events import EventEmitter
 from .user_cache import UserCache
 from .utils import (
-    safe_decode_bytes, normalize_addr, rid_to_int, bytes_to_int,
+    safe_decode_bytes, normalize_addr, rid_to_int, bytes_to_int, format_uptime,
     cleanup_old_logs, setup_logging, PeerAddress, detect_connection_type,
     fmt_ts_tg
 )
@@ -109,7 +110,13 @@ class HBProtocol(asyncio.DatagramProtocol):
         super().__init__()
         # All inbound connections (repeaters, hotspots, network links) - see models.py terminology note
         self._repeaters: Dict[bytes, RepeaterState] = {}
-        
+
+        # Logins claiming a radio ID that is already registered from a DIFFERENT
+        # sockaddr. Held aside so an unauthenticated RPTL cannot displace a live
+        # registration; the claimant is promoted (and the incumbent dropped) only
+        # once it proves the passphrase in _handle_auth_response.
+        self._pending_logins: Dict[bytes, RepeaterState] = {}
+
         # Outbound connection state management (Phase 2)
         self._outbounds: Dict[str, 'OutboundState'] = {}  # keyed by connection name
         self._openbridges: Dict[str, 'OpenBridgeState'] = {}  # keyed by OBP name
@@ -231,7 +238,12 @@ class HBProtocol(asyncio.DatagramProtocol):
             'rpto_received': repeater.rpto_received,
             'translations': translations_list,
             'last_ping': repeater.last_ping,
-            'missed_pings': repeater.missed_pings
+            'missed_pings': repeater.missed_pings,
+            # Authoritative session start. The dashboard must use this rather than
+            # the event's arrival time: repeater_connected is also emitted on missed
+            # pings, on recovery, and on dashboard-reconnect state replay, so an
+            # arrival-time stamp resets uptime for repeaters that never dropped.
+            'connect_time': repeater.connect_time
         }
     
     def _load_repeater_tg_config(self, repeater_id: bytes, repeater: RepeaterState) -> None:
@@ -1533,7 +1545,23 @@ class HBProtocol(asyncio.DatagramProtocol):
         self._data_log_recent = {
             k: v for k, v in self._data_log_recent.items() if v > data_log_cutoff
         }
-    
+
+        # Drop unanswered ID-takeover challenges
+        self._expire_pending_logins(current_time)
+
+    def _expire_pending_logins(self, current_time: float) -> None:
+        """
+        Discard takeover challenges that were never answered. A pending login holds
+        no routing state and can never displace anything on its own, so expiring it
+        only prevents unbounded growth from repeated unauthenticated claims.
+        """
+        cutoff = current_time - PENDING_LOGIN_TIMEOUT
+        for repeater_id, pending in list(self._pending_logins.items()):
+            if pending.last_ping < cutoff:
+                del self._pending_logins[repeater_id]
+                LOGGER.debug(f'Expired unanswered login challenge for repeater {rid_to_int(repeater_id)} '
+                             f'from {pending.ip}:{pending.port}')
+
     def _cleanup_user_cache(self):
         """Periodic cleanup of expired user cache entries"""
         if self._user_cache:
@@ -1775,6 +1803,12 @@ class HBProtocol(asyncio.DatagramProtocol):
                     repeater_id = data[4:8]
                 
             if not repeater_id:
+                # TODO (see docs/TODO.md, "Unreachable dispatch branches"): this
+                # also swallows TRUNCATED packets of known types. A short RPTL
+                # (fewer than 8 bytes) makes data[4:8] empty, so a malformed login
+                # is reported as an unknown packet type — misleading when reading
+                # logs. Distinguish "command not recognized" from "known command,
+                # too short" before changing anything here.
                 # Unknown packet type - log full details for investigation
                 try:
                     cmd_str = _command.decode('utf-8', errors='replace')
@@ -1797,8 +1831,12 @@ class HBProtocol(asyncio.DatagramProtocol):
                 self._send_nak(repeater_id, addr, reason="Repeater not registered")
                 return
 
-            # Update ping time for connected repeaters
-            if repeater and repeater.connection_state == 'connected':
+            # Update ping time for connected repeaters. The address must match the
+            # registration: a packet bearing a known ID from any other sockaddr must
+            # not hold the registration alive, or a stale/spoofed source could keep
+            # a dead entry from ever timing out.
+            if (repeater and repeater.connection_state == 'connected'
+                    and self._addr_matches_repeater(repeater, addr)):
                 repeater.last_ping = time()
                 # If missed_pings is being cleared, notify dashboard
                 if repeater.missed_pings > 0:
@@ -1811,35 +1849,50 @@ class HBProtocol(asyncio.DatagramProtocol):
             if _command == DMRD:
                 self._handle_dmr_data(data, addr)
             elif _command == RPTL:
-                LOGGER.debug(f'Received RPTL from {ip}:{port} - Repeater Login Request')
+                LOGGER.debug(f'Received RPTL from repeater {rid_to_int(repeater_id)} at {ip}:{port} - Repeater Login Request')
                 self._handle_repeater_login(repeater_id, addr)
             elif len(data) == 4:  # Special case: raw repeater ID login
-                # Try to interpret as a raw repeater ID
+                # DEAD CODE (see docs/TODO.md, "Unreachable dispatch branches").
+                # Unreachable: reaching here requires repeater_id to be truthy,
+                # but repeater_id is only ever set when _command matched one of
+                # the known commands above — each of which has its own branch in
+                # this chain. A bare 4-byte radio ID matches none of them, so it
+                # returns at the `if not repeater_id` guard and never arrives.
+                # Left in place rather than deleted: decide whether raw-ID login
+                # should be supported at all before removing it.
                 LOGGER.debug(f'Received possible raw repeater ID login from {ip}:{port}')
                 self._handle_repeater_login(data, addr)
             elif _command == RPTK:
-                LOGGER.debug(f'Received RPTK from {ip}:{port} - Authentication Response')
+                LOGGER.debug(f'Received RPTK from repeater {rid_to_int(repeater_id)} at {ip}:{port} - Authentication Response')
                 self._handle_auth_response(repeater_id, data[8:], addr)
             elif _command == RPTC:
                 if data[:5] == RPTCL:
-                    LOGGER.debug(f'Received RPTCL from {ip}:{port} - Disconnect Request')
+                    LOGGER.debug(f'Received RPTCL from repeater {rid_to_int(repeater_id)} at {ip}:{port} - Disconnect Request')
                     self._handle_disconnect(repeater_id, addr)
                 else:
-                    LOGGER.debug(f'Received RPTC from {ip}:{port} - Configuration Data')
+                    LOGGER.debug(f'Received RPTC from repeater {rid_to_int(repeater_id)} at {ip}:{port} - Configuration Data')
                     self._handle_config(data, addr)
             elif _command[:4] == RPTP:  # Check just RPTP prefix since that's enough to identify RPTPING
-                LOGGER.debug(f'Received RPTPING from {ip}:{port} - Repeater Keepalive')
+                LOGGER.debug(f'Received RPTPING from repeater {rid_to_int(repeater_id)} at {ip}:{port}')
                 self._handle_ping(repeater_id, addr)
             elif _command == RPTO:
-                LOGGER.info(f'Received RPTO from {ip}:{port} - Options/TG Configuration')
+                LOGGER.info(f'Received RPTO from repeater {rid_to_int(repeater_id)} at {ip}:{port} - Options/TG Configuration')
                 self._handle_options(repeater_id, data[8:], addr)
             elif _command == DMRA:
-                LOGGER.debug(f'Received DMRA from {ip}:{port} - DMR Talker Alias (packet length: {len(data)})')
+                LOGGER.debug(f'Received DMRA from repeater {rid_to_int(repeater_id)} at {ip}:{port} - DMR Talker Alias (packet length: {len(data)})')
                 if repeater_id:
                     self._handle_talker_alias(repeater_id, data[8:], addr)
                 else:
+                    # DEAD CODE (see docs/TODO.md, "Unreachable dispatch branches").
+                    # Unreachable: a falsy repeater_id already returned at the
+                    # `if not repeater_id` guard above, so this branch cannot run.
                     LOGGER.warning(f'DMRA packet from {ip}:{port} has no repeater_id - packet hex: {data[:20].hex()}')
             else:
+                # DEAD CODE (see docs/TODO.md, "Unreachable dispatch branches").
+                # Unreachable: reaching the else requires repeater_id to be truthy,
+                # which only happens for commands that each have a branch above.
+                # A genuinely unknown command is already reported by the
+                # `if not repeater_id` guard, which is where such packets land.
                 # Try to decode the command as ASCII for better logging
                 try:
                     cmd_str = _command.decode('utf-8', errors='replace')
@@ -2614,16 +2667,67 @@ class HBProtocol(asyncio.DatagramProtocol):
         
         repeater = self._repeaters.get(repeater_id)
         if repeater:
+            # This radio ID is already registered, so this login is a reconnect —
+            # report how long the previous session lasted and leave the judgement
+            # to the operator. Hours means a NAT lease expired or the device was
+            # power cycled; a few seconds, over and over, means it cannot stay
+            # connected. Covers both the same-address and new-address cases.
+            #
+            # Long-lived sessions reconnecting are unremarkable, so they can be
+            # filtered out: report only when the previous session was SHORTER than
+            # reconnect_report_threshold (0 reports every reconnect). Suppressing
+            # the line loses no record of the connection itself — _handle_config
+            # still logs "configured successfully" for every successful connect.
+            #
+            # Only meaningful once a session actually existed: connect_time is
+            # stamped where connection_state becomes 'connected', so in any earlier
+            # state it is just the object's creation time and "since last connect"
+            # would be a false statement. A login-stage retry is already covered by
+            # the "login retry, resending same salt" line below.
+            session_length = time() - repeater.connect_time
+            report_threshold = CONFIG.get('global', {}).get('reconnect_report_threshold', 3600)
+            if (repeater.connection_state == 'connected'
+                    and (report_threshold <= 0 or session_length < report_threshold)):
+                LOGGER.info(
+                    f'Repeater {rid_to_int(repeater_id)} logging in from {ip}:{port} - '
+                    f're-connecting from {repeater.ip}:{repeater.port} '
+                    f'({format_uptime(session_length)} since last connect)'
+                )
+
             if not self._addr_matches_repeater(repeater, addr):
-                LOGGER.warning(f'Repeater {rid_to_int(repeater_id)} attempting to connect from {ip}:{port} but already connected from {repeater.ip}:{repeater.port}')
-                # Remove the old registration first
-                old_addr = repeater.sockaddr
-                self._remove_repeater(repeater_id, "reconnect_different_port")
-                # Then send NAK to the old address to ensure cleanup
-                self._send_nak(repeater_id, old_addr, reason="Repeater reconnecting from new address")
-                # Continue with new connection below
+                # A new sockaddr does NOT inherit the old session, and we never
+                # assume it is the incumbent behind a remapped NAT port. It is held
+                # as a pending login and must prove the passphrase before it may
+                # replace the incumbent — see _handle_auth_response. Until then the
+                # live registration is untouched, so an unauthenticated packet
+                # bearing a known radio ID cannot knock a repeater off the network.
+                pending = self._pending_logins.get(repeater_id)
+                if pending and pending.sockaddr == normalize_addr(addr):
+                    salt = pending.salt  # same claimant retrying, reuse salt
+                else:
+                    salt = None
+                pending = RepeaterState(repeater_id=repeater_id, ip=ip, port=port)
+                if salt is not None:
+                    pending.salt = salt
+                pending.connection_state = 'login'
+                # last_ping is set to now by RepeaterState's default factory and is
+                # what _expire_pending_logins ages the challenge out on.
+                self._pending_logins[repeater_id] = pending
+
+                # INFO, not WARNING: a NAT lease expiring is routine, and warning on
+                # it would train the operator to ignore warnings. The reconnect line
+                # above carries the diagnostic; this line just states the action.
+                LOGGER.info(f'Repeater {rid_to_int(repeater_id)} login from {ip}:{port} claims an ID already '
+                            f'registered from {repeater.ip}:{repeater.port} - challenging; the incumbent is '
+                            f'kept until this address authenticates')
+                salt_bytes = pending.salt.to_bytes(4, 'big')
+                self._send_packet(b''.join([RPTACK, salt_bytes]), addr)
+                return
             else:
-                # Same repeater reconnecting from same IP:port
+                # Same repeater reconnecting from same IP:port. Any outstanding
+                # takeover challenge is now moot — drop it so a later RPTK is
+                # matched against this login's salt, not a stale one.
+                self._pending_logins.pop(repeater_id, None)
                 old_state = repeater.connection_state
                 LOGGER.info(f'Repeater {rid_to_int(repeater_id)} reconnecting while in state {old_state}')
                 # Preserve existing salt on login retry
@@ -2633,14 +2737,18 @@ class HBProtocol(asyncio.DatagramProtocol):
                     repeater.salt = existing_salt  # Reuse same salt
                     repeater.connection_state = 'login'
                     self._repeaters[repeater_id] = repeater
-                    
+
                     # Send login ACK with same salt
                     salt_bytes = repeater.salt.to_bytes(4, 'big')
                     self._send_packet(b''.join([RPTACK, salt_bytes]), addr)
                     LOGGER.info(f'Repeater {rid_to_int(repeater_id)} login retry from {ip}:{port}, resending same salt: {repeater.salt}')
                     return
-                
-        # Create or update repeater state (fresh login)
+
+        # Create or update repeater state (fresh login). Any outstanding takeover
+        # challenge is moot here — there is nothing to take over from — and must be
+        # dropped, or it would intercept this login's RPTK and check it against the
+        # stale salt, NAKing a repeater that answered correctly.
+        self._pending_logins.pop(repeater_id, None)
         repeater = RepeaterState(repeater_id=repeater_id, ip=ip, port=port)
         repeater.connection_state = 'login'
         self._repeaters[repeater_id] = repeater
@@ -2650,8 +2758,65 @@ class HBProtocol(asyncio.DatagramProtocol):
         self._send_packet(b''.join([RPTACK, salt_bytes]), addr)
         LOGGER.info(f'Repeater {rid_to_int(repeater_id)} login request from {ip}:{port}, sent salt: {repeater.salt}')
 
+    def _handle_takeover_auth_response(self, repeater_id: bytes, pending: RepeaterState,
+                                       auth_hash: bytes, addr: PeerAddress) -> None:
+        """
+        Authenticate a login that claims a radio ID already registered from another
+        sockaddr. Only a correct passphrase earns the takeover; on every failure the
+        pending state is discarded and the incumbent registration is left untouched.
+        """
+        del self._pending_logins[repeater_id]
+        ip, port = addr[0], addr[1]
+
+        try:
+            repeater_config = self._matcher.get_repeater_config(
+                rid_to_int(repeater_id), pending.get_callsign_str()
+            )
+            if repeater_config is None:
+                LOGGER.warning(f'Denied takeover of repeater {rid_to_int(repeater_id)} by {ip}:{port} '
+                               f'- no matching configuration; incumbent retained')
+                self._send_nak(repeater_id, addr, reason="No matching configuration")
+                return
+
+            salt_bytes = pending.salt.to_bytes(4, 'big')
+            calc_hash = bytes.fromhex(sha256(b''.join([salt_bytes, repeater_config.passphrase.encode()])).hexdigest())
+            if auth_hash != calc_hash:
+                LOGGER.warning(f'Denied takeover of repeater {rid_to_int(repeater_id)} by {ip}:{port} '
+                               f'- authentication failed; incumbent retained')
+                self._send_nak(repeater_id, addr, reason="Authentication failed")
+                return
+
+            # Passphrase proven. Now — and only now — the incumbent gives way.
+            # Order matters: _remove_repeater deletes by ID, so the old entry must
+            # go before the claimant is installed under the same key.
+            incumbent = self._repeaters.get(repeater_id)
+            if incumbent is not None:
+                old_addr = incumbent.sockaddr
+                LOGGER.info(f'Repeater {rid_to_int(repeater_id)} authenticated from {ip}:{port} - '
+                            f'replacing registration from {old_addr[0]}:{old_addr[1]}')
+                self._remove_repeater(repeater_id, "replaced_by_new_address")
+                self._send_nak(repeater_id, old_addr, reason="Replaced by new registration")
+
+            pending.authenticated = True
+            pending.connection_state = 'config'
+            self._repeaters[repeater_id] = pending
+            self._send_packet(b''.join([RPTACK, repeater_id]), addr)
+            LOGGER.info(f'Repeater {rid_to_int(repeater_id)} authenticated successfully from {ip}:{port}')
+
+        except Exception as e:
+            LOGGER.error(f'Authentication error for repeater {rid_to_int(repeater_id)} from {ip}:{port}: {str(e)}')
+            self._send_nak(repeater_id, addr)
+
     def _handle_auth_response(self, repeater_id: bytes, auth_hash: bytes, addr: PeerAddress) -> None:
         """Handle authentication response from repeater"""
+        # A claimant challenged by _handle_repeater_login for an already-registered
+        # ID answers here. It authenticates against its own pending state so that a
+        # failure cannot touch the incumbent's registration.
+        pending = self._pending_logins.get(repeater_id)
+        if pending is not None and pending.sockaddr == normalize_addr(addr):
+            self._handle_takeover_auth_response(repeater_id, pending, auth_hash, addr)
+            return
+
         repeater = self._validate_repeater(repeater_id, addr)
         if not repeater or repeater.connection_state != 'login':
             LOGGER.warning(f'Auth response from repeater {rid_to_int(repeater_id)} in wrong state')
@@ -2736,6 +2901,9 @@ class HBProtocol(asyncio.DatagramProtocol):
 
             repeater.connected = True
             repeater.connection_state = 'connected'
+            # Mark the start of this session; the next login for this ID reports
+            # the delta so the operator can see how long it actually stayed up.
+            repeater.connect_time = time()
             
             # Load and cache TG sets from config for fast routing checks
             self._load_repeater_tg_config(repeater_id, repeater)
@@ -3276,8 +3444,11 @@ class HBProtocol(asyncio.DatagramProtocol):
     def _handle_ping(self, repeater_id: bytes, addr: PeerAddress) -> None:
         """Handle ping (RPTPING/RPTP) from the repeater as a keepalive."""
         repeater = self._validate_repeater(repeater_id, addr)
-        if not repeater or repeater.connection_state != 'connected':
-            LOGGER.warning(f'Ping from repeater {rid_to_int(repeater_id)} in wrong state (state="{repeater.connection_state}" if repeater else "None")')
+        if not repeater:
+            return  # _validate_repeater already logged and NAKed
+        if repeater.connection_state != 'connected':
+            LOGGER.warning(f'Ping from repeater {rid_to_int(repeater_id)} at {addr[0]}:{addr[1]} '
+                           f'in wrong state (state={repeater.connection_state})')
             self._send_nak(repeater_id, addr, reason="Wrong connection state")
             return
             
@@ -3294,7 +3465,7 @@ class HBProtocol(asyncio.DatagramProtocol):
             self._events.emit('repeater_connected', self._prepare_repeater_event_data(repeater_id, repeater))
         
         # Send MSTPONG in response to RPTPING/RPTP from repeater
-        LOGGER.debug(f'Sending MSTPONG to repeater {rid_to_int(repeater_id)}')
+        LOGGER.debug(f'Sending MSTPONG to repeater {rid_to_int(repeater_id)} at {addr[0]}:{addr[1]}')
         self._send_packet(b''.join([MSTPONG, repeater_id]), addr)
 
     def _handle_disconnect(self, repeater_id: bytes, addr: PeerAddress) -> None:

hblink4/models.py

@@ -297,6 +297,11 @@ class RepeaterState:
     connected: bool = False
     authenticated: bool = False
     last_ping: float = field(default_factory=time)
+    # When this registration last reached the 'connected' state. Stamped at
+    # construction so a login that never completes still has a usable value, then
+    # re-stamped on transition to 'connected'. Used to report how long the
+    # previous session lasted when a radio ID logs in again.
+    connect_time: float = field(default_factory=time)
     ping_count: int = 0
     missed_pings: int = 0
     salt: int = field(default_factory=lambda: randint(0, 0xFFFFFFFF))

hblink4/utils.py

@@ -43,6 +43,26 @@ def normalize_addr(addr: PeerAddress) -> Tuple[str, int]:
     return (addr[0], addr[1])
 
 
+def format_uptime(seconds: float) -> str:
+    """
+    Format an elapsed time as D:HH:MM:SS for log output.
+
+    Zero-padded so the values line up when scanning a log column — the pattern
+    across repeated lines is the diagnostic, not any single value.
+
+    Args:
+        seconds: Elapsed seconds (negative values are clamped to zero)
+
+    Returns:
+        Elapsed time as 'D:HH:MM:SS', e.g. '0:00:00:05' or '3:04:17:09'
+    """
+    total = int(max(0.0, seconds))
+    days, rem = divmod(total, 86400)
+    hours, rem = divmod(rem, 3600)
+    minutes, secs = divmod(rem, 60)
+    return f'{days}:{hours:02d}:{minutes:02d}:{secs:02d}'
+
+
 def rid_to_int(repeater_id: bytes) -> int:
     """
     Convert repeater ID bytes to int.

tests/test_id_takeover.py

@@ -0,0 +1,634 @@
+"""
+Tests for radio-ID takeover: a login claiming a radio ID that is already
+registered from a DIFFERENT socket address.
+
+The rule under test: a registered radio ID is never displaced by an
+unauthenticated packet. A claimant from a new sockaddr is challenged and held in
+_pending_logins; the incumbent is torn down only after the claimant proves the
+passphrase. This matters because radio IDs are public — without the auth gate an
+8-byte RPTL from any source address would evict any repeater on the network.
+
+The invariant that must hold at every step: _repeaters is keyed by radio ID, and
+only entries in _repeaters are routable, so there is never more than one
+forwardable entity per radio ID.
+"""
+
+import unittest
+import sys
+import os
+from hashlib import sha256
+from time import time
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from hblink4 import hblink as hblink_mod
+from hblink4.hblink import HBProtocol
+from hblink4.models import RepeaterState
+from hblink4.constants import RPTACK, MSTNAK, PENDING_LOGIN_TIMEOUT
+from hblink4.utils import format_uptime
+
+
+PASSPHRASE = 'takeover-test-key'
+RADIO_ID = b'\x00\x04\xc2\xc0'          # 312000
+INCUMBENT = ('198.51.100.10', 62031)     # already registered
+CLAIMANT = ('203.0.113.55', 54001)       # same site, new NAT mapping
+OTHER = ('203.0.113.55', 54002)          # a second claimant racing the first, same site
+REMOTE = ('198.51.100.77', 61000)        # a genuinely different host, different IP
+
+_CONFIG = {
+    'global': {'user_cache': {'timeout': 600}},
+    'dashboard': {'enabled': False},
+    'repeater_configurations': {
+        'patterns': [
+            {
+                'name': 'Test Network',
+                'description': 'Fixture pattern for takeover tests',
+                'match': {'id_ranges': [[312000, 312999]]},
+                'config': {
+                    'passphrase': PASSPHRASE,
+                    'slot1_talkgroups': [8],
+                    'slot2_talkgroups': [3120],
+                },
+            }
+        ]
+    },
+}
+
+
+class FakeTransport:
+    """Records outbound datagrams instead of putting them on the wire."""
+
+    def __init__(self):
+        self.sent = []
+
+    def sendto(self, data, addr=None):
+        self.sent.append((data, addr))
+
+    def packets_to(self, addr):
+        return [data for data, dest in self.sent if dest == addr]
+
+    def last_to(self, addr):
+        pkts = self.packets_to(addr)
+        return pkts[-1] if pkts else None
+
+
+class TakeoverTestCase(unittest.TestCase):
+    def setUp(self):
+        # HBProtocol reads the module-level CONFIG at construction time.
+        self._saved_config = hblink_mod.CONFIG
+        hblink_mod.CONFIG = _CONFIG
+        self.hb = HBProtocol()
+        self.hb.transport = FakeTransport()
+        self.tx = self.hb.transport
+
+    def tearDown(self):
+        hblink_mod.CONFIG = self._saved_config
+
+    # -- helpers ------------------------------------------------------------
+
+    def register_incumbent(self, addr=INCUMBENT):
+        """Install a fully connected repeater, as if it had completed login."""
+        state = RepeaterState(repeater_id=RADIO_ID, ip=addr[0], port=addr[1])
+        state.connection_state = 'connected'
+        state.connected = True
+        state.authenticated = True
+        state.callsign = b'W1AW    '
+        self.hb._repeaters[RADIO_ID] = state
+        return state
+
+    def salt_from_ack(self, addr):
+        """Pull the salt out of the RPTACK challenge sent to addr."""
+        pkt = self.tx.last_to(addr)
+        self.assertIsNotNone(pkt, f'no packet was sent to {addr}')
+        self.assertTrue(pkt.startswith(RPTACK), f'expected RPTACK, got {pkt[:8]!r}')
+        self.assertEqual(len(pkt), 10, 'challenge should be RPTACK + 4-byte salt')
+        return int.from_bytes(pkt[6:10], 'big')
+
+    def auth_hash(self, salt, passphrase=PASSPHRASE):
+        return bytes.fromhex(
+            sha256(salt.to_bytes(4, 'big') + passphrase.encode()).hexdigest()
+        )
+
+    def assert_single_routable(self):
+        """The core invariant: at most one routable entity per radio ID."""
+        routable = [rid for rid, r in self.hb._repeaters.items() if rid == RADIO_ID]
+        self.assertLessEqual(len(routable), 1)
+        # A pending claimant must never be routable, i.e. never in _repeaters.
+        pending = self.hb._pending_logins.get(RADIO_ID)
+        if pending is not None:
+            self.assertIsNot(pending, self.hb._repeaters.get(RADIO_ID))
+            self.assertEqual(pending.connection_state, 'login')
+            self.assertFalse(pending.connected)
+            self.assertFalse(pending.authenticated)
+
+    # -- the auth gate ------------------------------------------------------
+
+    def test_login_from_new_address_does_not_evict_incumbent(self):
+        """RPTL alone must never displace a live registration."""
+        incumbent = self.register_incumbent()
+
+        self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)
+
+        self.assertIs(self.hb._repeaters[RADIO_ID], incumbent,
+                      'incumbent was replaced by an unauthenticated login')
+        self.assertEqual(self.hb._repeaters[RADIO_ID].sockaddr, INCUMBENT)
+        self.assertEqual(self.hb._repeaters[RADIO_ID].connection_state, 'connected')
+        self.assertIn(RADIO_ID, self.hb._pending_logins)
+        self.assertEqual(self.hb._pending_logins[RADIO_ID].sockaddr, CLAIMANT)
+        # Claimant was challenged, incumbent was not disturbed.
+        self.salt_from_ack(CLAIMANT)
+        self.assertEqual(self.tx.packets_to(INCUMBENT), [])
+        self.assert_single_routable()
+
+    def test_correct_passphrase_promotes_claimant_and_evicts_incumbent(self):
+        self.register_incumbent()
+        self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)
+        salt = self.salt_from_ack(CLAIMANT)
+
+        self.hb._handle_auth_response(RADIO_ID, self.auth_hash(salt), CLAIMANT)
+
+        promoted = self.hb._repeaters[RADIO_ID]
+        self.assertEqual(promoted.sockaddr, CLAIMANT)
+        self.assertTrue(promoted.authenticated)
+        self.assertEqual(promoted.connection_state, 'config')
+        self.assertNotIn(RADIO_ID, self.hb._pending_logins,
+                         'claimant must not remain pending after promotion')
+        # Old address is told to clean up.
+        self.assertTrue(self.tx.last_to(INCUMBENT).startswith(MSTNAK))
+        self.assert_single_routable()
+
+    def test_wrong_passphrase_leaves_incumbent_untouched(self):
+        incumbent = self.register_incumbent()
+        self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)
+        salt = self.salt_from_ack(CLAIMANT)
+
+        self.hb._handle_auth_response(
+            RADIO_ID, self.auth_hash(salt, 'wrong-key'), CLAIMANT)
+
+        self.assertIs(self.hb._repeaters[RADIO_ID], incumbent,
+                      'failed auth must not disturb the incumbent')
+        self.assertEqual(self.hb._repeaters[RADIO_ID].connection_state, 'connected')
+        self.assertNotIn(RADIO_ID, self.hb._pending_logins,
+                         'failed claimant must be discarded')
+        self.assertTrue(self.tx.last_to(CLAIMANT).startswith(MSTNAK))
+        # The incumbent was never NAKed.
+        self.assertEqual(
+            [p for p in self.tx.packets_to(INCUMBENT) if p.startswith(MSTNAK)], [])
+        self.assert_single_routable()
+
+    def test_unanswered_challenge_expires(self):
+        incumbent = self.register_incumbent()
+        self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)
+        pending = self.hb._pending_logins[RADIO_ID]
+
+        # Not yet stale.
+        self.hb._expire_pending_logins(pending.last_ping + 1.0)
+        self.assertIn(RADIO_ID, self.hb._pending_logins)
+
+        # Past the window.
+        self.hb._expire_pending_logins(pending.last_ping + PENDING_LOGIN_TIMEOUT + 1.0)
+        self.assertNotIn(RADIO_ID, self.hb._pending_logins)
+        self.assertIs(self.hb._repeaters[RADIO_ID], incumbent)
+        self.assert_single_routable()
+
+    def test_stale_salt_cannot_be_replayed_after_expiry(self):
+        """A challenge that aged out must not still authenticate."""
+        self.register_incumbent()
+        self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)
+        salt = self.salt_from_ack(CLAIMANT)
+        pending = self.hb._pending_logins[RADIO_ID]
+        self.hb._expire_pending_logins(pending.last_ping + PENDING_LOGIN_TIMEOUT + 1.0)
+
+        self.hb._handle_auth_response(RADIO_ID, self.auth_hash(salt), CLAIMANT)
+
+        self.assertEqual(self.hb._repeaters[RADIO_ID].sockaddr, INCUMBENT)
+        self.assert_single_routable()
+
+    # -- races --------------------------------------------------------------
+
+    def test_two_claimants_race_only_one_can_promote(self):
+        """
+        Two addresses claim the same ID before either authenticates. Only the
+        holder of the challenge may promote, and only once.
+        """
+        self.register_incumbent()
+
+        self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)
+        salt_a = self.salt_from_ack(CLAIMANT)
+        self.hb._handle_repeater_login(RADIO_ID, OTHER)
+        salt_b = self.salt_from_ack(OTHER)
+
+        # One pending slot per radio ID; the later claim holds it.
+        self.assertEqual(self.hb._pending_logins[RADIO_ID].sockaddr, OTHER)
+        self.assert_single_routable()
+
+        # The displaced claimant answers with a valid hash for its own salt and
+        # must still be refused — it no longer holds the challenge.
+        self.hb._handle_auth_response(RADIO_ID, self.auth_hash(salt_a), CLAIMANT)
+        self.assertEqual(self.hb._repeaters[RADIO_ID].sockaddr, INCUMBENT,
+                         'a stale claimant must not evict the incumbent')
+        self.assertTrue(self.tx.last_to(CLAIMANT).startswith(MSTNAK))
+        self.assert_single_routable()
+
+        # The challenge holder authenticates and takes over.
+        self.hb._handle_auth_response(RADIO_ID, self.auth_hash(salt_b), OTHER)
+        self.assertEqual(self.hb._repeaters[RADIO_ID].sockaddr, OTHER)
+        self.assert_single_routable()
+
+        # A second, replayed RPTK from the winner must not create a duplicate.
+        self.hb._handle_auth_response(RADIO_ID, self.auth_hash(salt_b), OTHER)
+        self.assertEqual(len([r for r in self.hb._repeaters if r == RADIO_ID]), 1)
+        self.assert_single_routable()
+
+    def test_claim_against_incumbent_still_in_login(self):
+        """
+        The incumbent has not finished registering (state='login', nothing
+        de-registered yet) when a second address claims the same ID.
+        """
+        self.hb._handle_repeater_login(RADIO_ID, INCUMBENT)   # fresh login, no incumbent
+        first_salt = self.salt_from_ack(INCUMBENT)
+        self.assertEqual(self.hb._repeaters[RADIO_ID].connection_state, 'login')
+
+        # Second address claims the ID mid-handshake.
+        self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)
+        self.assertEqual(self.hb._repeaters[RADIO_ID].sockaddr, INCUMBENT,
+                         'half-registered entry must not be replaced by a bare login')
+        self.assertEqual(self.hb._pending_logins[RADIO_ID].sockaddr, CLAIMANT)
+        self.assert_single_routable()
+
+        # The original finishes its handshake normally.
+        self.hb._handle_auth_response(RADIO_ID, self.auth_hash(first_salt), INCUMBENT)
+        self.assertEqual(self.hb._repeaters[RADIO_ID].sockaddr, INCUMBENT)
+        self.assertEqual(self.hb._repeaters[RADIO_ID].connection_state, 'config')
+        self.assert_single_routable()
+
+    def test_claimant_retry_reuses_salt(self):
+        """Repeated RPTL from the same claimant must not invalidate its challenge."""
+        self.register_incumbent()
+        self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)
+        salt_first = self.salt_from_ack(CLAIMANT)
+        self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)
+        salt_retry = self.salt_from_ack(CLAIMANT)
+
+        self.assertEqual(salt_first, salt_retry,
+                         'retry from the same address should reuse the salt')
+        self.hb._handle_auth_response(RADIO_ID, self.auth_hash(salt_first), CLAIMANT)
+        self.assertEqual(self.hb._repeaters[RADIO_ID].sockaddr, CLAIMANT)
+        self.assert_single_routable()
+
+    def test_incumbent_relogin_clears_outstanding_challenge(self):
+        """
+        The incumbent re-logs-in from its own address while a claim is pending.
+        The stale challenge must be dropped so a later RPTK is matched against
+        the incumbent's new salt, not the claimant's.
+        """
+        self.register_incumbent()
+        self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)
+        stale_salt = self.salt_from_ack(CLAIMANT)
+
+        self.hb._handle_repeater_login(RADIO_ID, INCUMBENT)
+        self.assertNotIn(RADIO_ID, self.hb._pending_logins)
+
+        # The abandoned claimant's hash is now worthless.
+        self.hb._handle_auth_response(RADIO_ID, self.auth_hash(stale_salt), CLAIMANT)
+        self.assertEqual(self.hb._repeaters[RADIO_ID].sockaddr, INCUMBENT)
+        self.assert_single_routable()
+
+    def test_stale_challenge_does_not_block_a_fresh_login_from_same_address(self):
+        """
+        Regression: a claimant is challenged, the incumbent then times out, and the
+        claimant re-logs-in from the same address. The fresh login issues a new
+        salt; the abandoned challenge must not intercept the RPTK and check it
+        against the old salt, which would NAK a correct answer.
+        """
+        self.register_incumbent()
+        self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)
+        old_salt = self.salt_from_ack(CLAIMANT)
+
+        self.hb._remove_repeater(RADIO_ID, 'timeout')
+        self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)   # fresh login path
+        new_salt = self.salt_from_ack(CLAIMANT)
+
+        self.assertNotIn(RADIO_ID, self.hb._pending_logins,
+                         'fresh login must clear the moot challenge')
+        self.assertNotEqual(old_salt, new_salt)
+
+        self.hb._handle_auth_response(RADIO_ID, self.auth_hash(new_salt), CLAIMANT)
+        self.assertEqual(self.hb._repeaters[RADIO_ID].connection_state, 'config',
+                         'correct answer to the new salt was rejected')
+        self.assertTrue(self.hb._repeaters[RADIO_ID].authenticated)
+        self.assert_single_routable()
+
+    def test_two_fresh_logins_no_incumbent_last_authenticated_wins(self):
+        """
+        Two addresses claim an unregistered ID. Arrival order decides who holds
+        _repeaters, but only authentication decides who keeps it — and neither is
+        routable until it reaches 'connected'.
+        """
+        self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)   # A: fresh login
+        salt_a = self.salt_from_ack(CLAIMANT)
+        self.hb._handle_repeater_login(RADIO_ID, OTHER)      # B: challenged
+        salt_b = self.salt_from_ack(OTHER)
+
+        # A holds the slot purely by arriving first, unauthenticated.
+        self.assertEqual(self.hb._repeaters[RADIO_ID].sockaddr, CLAIMANT)
+        self.assertFalse(self.hb._repeaters[RADIO_ID].authenticated)
+        self.assertEqual(self.hb._pending_logins[RADIO_ID].sockaddr, OTHER)
+        # Neither is routable yet.
+        self.assertNotEqual(self.hb._repeaters[RADIO_ID].connection_state, 'connected')
+        self.assert_single_routable()
+
+        self.hb._handle_auth_response(RADIO_ID, self.auth_hash(salt_a), CLAIMANT)
+        self.assertEqual(self.hb._repeaters[RADIO_ID].sockaddr, CLAIMANT)
+        self.assert_single_routable()
+
+        # B proves the passphrase too and takes it.
+        self.hb._handle_auth_response(RADIO_ID, self.auth_hash(salt_b), OTHER)
+        self.assertEqual(self.hb._repeaters[RADIO_ID].sockaddr, OTHER)
+        self.assert_single_routable()
+
+    def test_takeover_after_incumbent_already_gone(self):
+        """A pending claim must still complete if the incumbent times out first."""
+        self.register_incumbent()
+        self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)
+        salt = self.salt_from_ack(CLAIMANT)
+
+        self.hb._remove_repeater(RADIO_ID, 'timeout')
+        self.assertNotIn(RADIO_ID, self.hb._repeaters)
+
+        self.hb._handle_auth_response(RADIO_ID, self.auth_hash(salt), CLAIMANT)
+        self.assertEqual(self.hb._repeaters[RADIO_ID].sockaddr, CLAIMANT)
+        self.assertEqual(self.hb._repeaters[RADIO_ID].connection_state, 'config')
+        self.assert_single_routable()
+
+    def test_pending_claimant_is_never_a_forwarding_target(self):
+        """
+        The whole point of the pending map: a claimant must not appear anywhere
+        the router can see it while the incumbent is still live.
+        """
+        incumbent = self.register_incumbent()
+        incumbent.slot2_talkgroups = {b'\x00\x0c\x30'}   # make it a real eligible target
+        self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)
+
+        targets = self.hb._calculate_stream_targets(
+            b'\x00\x04\xc2\xc1',           # some other source repeater
+            2, b'\x00\x0c\x30', b'\x11\x22\x33\x44', b'\x2f\xa9\x05')
+
+        # Sanity: the incumbent IS routable, so this test is not vacuous.
+        self.assertIn(RADIO_ID, targets,
+                      'incumbent should be an eligible target; test setup is wrong')
+        # Targets are radio IDs; every one must resolve to a live registration,
+        # and the ID in question must resolve to the incumbent, never the claimant.
+        for t in targets:
+            if isinstance(t, bytes):
+                resolved = self.hb._repeaters[t]
+                self.assertNotEqual(resolved.sockaddr, CLAIMANT,
+                                    'a pending claimant became a forwarding target')
+        self.assertEqual(self.hb._repeaters[RADIO_ID].sockaddr, INCUMBENT)
+        self.assert_single_routable()
+
+
+class ReconnectReportingTestCase(unittest.TestCase):
+    """
+    A login for a radio ID that is already registered reports how long the
+    previous session lasted. We report the fact; the operator judges it.
+    """
+
+    def setUp(self):
+        self._saved_config = hblink_mod.CONFIG
+        hblink_mod.CONFIG = _CONFIG
+        self.hb = HBProtocol()
+        self.hb.transport = FakeTransport()
+
+    def tearDown(self):
+        hblink_mod.CONFIG = self._saved_config
+
+    def install(self, addr, connected_ago):
+        state = RepeaterState(repeater_id=RADIO_ID, ip=addr[0], port=addr[1])
+        state.connection_state = 'connected'
+        state.connected = True
+        state.authenticated = True
+        state.connect_time = time() - connected_ago
+        self.hb._repeaters[RADIO_ID] = state
+        return state
+
+    def reconnect_lines(self, logs):
+        return [m for m in logs.output if 'since last connect' in m]
+
+    def test_first_login_reports_no_reconnect(self):
+        with self.assertLogs('hblink4.hblink', level='INFO') as logs:
+            self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)
+        self.assertEqual(self.reconnect_lines(logs), [],
+                         'a first-ever login is not a reconnect')
+
+    def test_reconnect_from_new_address_reports_both_addresses(self):
+        self.install(INCUMBENT, connected_ago=5)
+        with self.assertLogs('hblink4.hblink', level='INFO') as logs:
+            self.hb._handle_repeater_login(RADIO_ID, REMOTE)
+
+        lines = self.reconnect_lines(logs)
+        self.assertEqual(len(lines), 1)
+        self.assertIn(f'{REMOTE[0]}:{REMOTE[1]}', lines[0])       # where it is now
+        self.assertIn(f'{INCUMBENT[0]}:{INCUMBENT[1]}', lines[0])  # where it was
+        self.assertIn('0:00:00:05', lines[0])
+
+    def test_login_stage_retry_is_not_called_a_reconnect(self):
+        """
+        A repeater retrying RPTL because it never got the ACK has not connected,
+        so there is no session to report. connect_time in a pre-'connected' state
+        is only the object's creation time — reporting it would state something
+        untrue. The 'login retry, resending same salt' line covers this case.
+        """
+        with self.assertLogs('hblink4.hblink', level='INFO') as logs:
+            self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)   # first attempt
+            self.hb._repeaters[RADIO_ID].connect_time -= 5       # 5s passes
+            self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)   # retry
+
+        self.assertEqual(self.hb._repeaters[RADIO_ID].connection_state, 'login')
+        self.assertEqual(self.reconnect_lines(logs), [],
+                         'a device that never connected must not be called a reconnect')
+        self.assertTrue([m for m in logs.output if 'login retry' in m],
+                        'the retry itself should still be logged')
+
+    def test_config_stage_retry_is_not_called_a_reconnect(self):
+        """Authenticated but never configured is still not a completed session."""
+        self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)
+        state = self.hb._repeaters[RADIO_ID]
+        state.connection_state = 'config'
+        state.authenticated = True
+        state.connect_time -= 30
+
+        with self.assertLogs('hblink4.hblink', level='INFO') as logs:
+            self.hb._handle_repeater_login(RADIO_ID, CLAIMANT)
+        self.assertEqual(self.reconnect_lines(logs), [])
+
+    def test_reconnect_from_same_address_is_still_reported(self):
+        """The 'cannot stay logged in' case — same address, over and over."""
+        self.install(INCUMBENT, connected_ago=5)
+        with self.assertLogs('hblink4.hblink', level='INFO') as logs:
+            self.hb._handle_repeater_login(RADIO_ID, INCUMBENT)
+
+        lines = self.reconnect_lines(logs)
+        self.assertEqual(len(lines), 1)
+        self.assertIn('0:00:00:05', lines[0])
+
+    def test_long_session_renders_days_when_reporting_is_unfiltered(self):
+        """
+        With the threshold disabled, a multi-day session renders in full. The
+        default threshold suppresses this case as routine — see
+        ReconnectThresholdTestCase — so reporting must be turned off to see it.
+        """
+        hblink_mod.CONFIG = {**_CONFIG,
+                             'global': {**_CONFIG['global'],
+                                        'reconnect_report_threshold': 0}}
+        self.install(INCUMBENT, connected_ago=(3 * 86400) + (4 * 3600) + (17 * 60) + 9)
+        with self.assertLogs('hblink4.hblink', level='INFO') as logs:
+            self.hb._handle_repeater_login(RADIO_ID, REMOTE)
+        self.assertIn('3:04:17:09', self.reconnect_lines(logs)[0])
+
+    def test_connect_time_is_stamped_on_reaching_connected(self):
+        """Without the re-stamp the delta would measure from object creation."""
+        state = RepeaterState(repeater_id=RADIO_ID, ip=CLAIMANT[0], port=CLAIMANT[1])
+        state.connect_time = time() - 9999
+        state.authenticated = True
+        state.connection_state = 'config'
+        self.hb._repeaters[RADIO_ID] = state
+
+        config_packet = b'RPTC' + RADIO_ID + b'W1AW    ' + (b' ' * 294)
+        self.hb._handle_config(config_packet, CLAIMANT)
+
+        self.assertEqual(state.connection_state, 'connected')
+        self.assertLess(time() - state.connect_time, 5,
+                        'connect_time should be re-stamped when the session starts')
+
+
+class ReconnectThresholdTestCase(unittest.TestCase):
+    """
+    reconnect_report_threshold filters out routine reconnects: only sessions
+    SHORTER than the threshold are reported, because those are the ones that
+    indicate a device which cannot stay online.
+    """
+
+    def setUp(self):
+        self._saved_config = hblink_mod.CONFIG
+
+    def tearDown(self):
+        hblink_mod.CONFIG = self._saved_config
+
+    def build(self, threshold='unset'):
+        config = dict(_CONFIG)
+        config['global'] = dict(_CONFIG['global'])
+        if threshold != 'unset':
+            config['global']['reconnect_report_threshold'] = threshold
+        hblink_mod.CONFIG = config
+        hb = HBProtocol()
+        hb.transport = FakeTransport()
+        return hb
+
+    def attempt(self, hb, session_length):
+        state = RepeaterState(repeater_id=RADIO_ID, ip=INCUMBENT[0], port=INCUMBENT[1])
+        state.connection_state = 'connected'
+        state.connected = True
+        state.connect_time = time() - session_length
+        hb._repeaters[RADIO_ID] = state
+        with self.assertLogs('hblink4.hblink', level='INFO') as logs:
+            hb._handle_repeater_login(RADIO_ID, REMOTE)
+        return [m for m in logs.output if 'since last connect' in m]
+
+    def test_short_session_is_reported(self):
+        hb = self.build(threshold=3600)
+        self.assertEqual(len(self.attempt(hb, session_length=5)), 1)
+
+    def test_long_session_is_suppressed(self):
+        hb = self.build(threshold=3600)
+        self.assertEqual(self.attempt(hb, session_length=3 * 86400), [],
+                         'a routine reconnect after days should not be reported')
+
+    def test_boundary_is_exclusive(self):
+        """Exactly at the threshold counts as routine."""
+        hb = self.build(threshold=3600)
+        self.assertEqual(self.attempt(hb, session_length=3601), [])
+        hb = self.build(threshold=3600)
+        self.assertEqual(len(self.attempt(hb, session_length=3599)), 1)
+
+    def test_zero_reports_everything(self):
+        hb = self.build(threshold=0)
+        self.assertEqual(len(self.attempt(hb, session_length=30 * 86400)), 1)
+
+    def test_missing_config_key_uses_default(self):
+        """An existing config with no such key must keep working."""
+        hb = self.build()  # key absent entirely
+        self.assertEqual(len(self.attempt(hb, session_length=5)), 1)
+        hb = self.build()
+        self.assertEqual(self.attempt(hb, session_length=7200), [],
+                         'default threshold of 3600 should suppress a 2h session')
+
+    def test_takeover_still_proceeds_when_report_is_suppressed(self):
+        """Filtering the log line must not change any protocol behavior."""
+        hb = self.build(threshold=3600)
+        self.attempt(hb, session_length=3 * 86400)
+        self.assertIn(RADIO_ID, hb._pending_logins,
+                      'the claimant should still have been challenged')
+        self.assertEqual(hb._repeaters[RADIO_ID].sockaddr, INCUMBENT)
+
+
+class ConnectTimeEventTestCase(unittest.TestCase):
+    """
+    The dashboard derives uptime from this value. It must come from the server,
+    because repeater_connected is emitted on missed pings and on state replay,
+    where the event's arrival time would wrongly reset uptime.
+    """
+
+    def setUp(self):
+        self._saved_config = hblink_mod.CONFIG
+        hblink_mod.CONFIG = _CONFIG
+        self.hb = HBProtocol()
+        self.hb.transport = FakeTransport()
+
+    def tearDown(self):
+        hblink_mod.CONFIG = self._saved_config
+
+    def test_connect_time_is_included_in_event_payload(self):
+        state = RepeaterState(repeater_id=RADIO_ID, ip=INCUMBENT[0], port=INCUMBENT[1])
+        state.connect_time = time() - 12345
+        payload = self.hb._prepare_repeater_event_data(RADIO_ID, state)
+        self.assertIn('connect_time', payload)
+        self.assertEqual(payload['connect_time'], state.connect_time)
+
+    def test_connect_time_survives_a_missed_ping_and_recovery(self):
+        """The bug this fixes: uptime resetting for a repeater that never dropped."""
+        state = RepeaterState(repeater_id=RADIO_ID, ip=INCUMBENT[0], port=INCUMBENT[1])
+        state.connection_state = 'connected'
+        state.connected = True
+        original = time() - 86400
+        state.connect_time = original
+        self.hb._repeaters[RADIO_ID] = state
+
+        state.missed_pings = 2                       # missed pings, then recovery
+        self.hb._handle_ping(RADIO_ID, INCUMBENT)
+
+        self.assertEqual(state.connect_time, original,
+                         'recovery must not restart the session clock')
+        payload = self.hb._prepare_repeater_event_data(RADIO_ID, state)
+        self.assertEqual(payload['connect_time'], original)
+
+
+class FormatUptimeTestCase(unittest.TestCase):
+    def test_formats(self):
+        self.assertEqual(format_uptime(0), '0:00:00:00')
+        self.assertEqual(format_uptime(5), '0:00:00:05')
+        self.assertEqual(format_uptime(65), '0:00:01:05')
+        self.assertEqual(format_uptime(3671), '0:01:01:11')
+        self.assertEqual(format_uptime(86400), '1:00:00:00')
+        self.assertEqual(format_uptime(273429), '3:03:57:09')
+
+    def test_fractional_seconds_truncate(self):
+        self.assertEqual(format_uptime(5.9), '0:00:00:05')
+
+    def test_negative_clamps_to_zero(self):
+        """Clock adjustments must not produce nonsense in the log."""
+        self.assertEqual(format_uptime(-10), '0:00:00:00')
+
+
+if __name__ == '__main__':
+    unittest.main()