Fix tech-debt issues #7, #10; investigate/document #8, #14

docs/PROTOCOL.md

@@ -32,6 +32,23 @@ parsing paths; their encode-side correctness is instead covered by
 byte-exact vectors (`tests/vectors.h`, built by hand from the spec) in
 `tests/test_codec.c`.
 
+## Decode strictness
+
+`mqtt_decode()` is a tolerant receiver by default (accepts anything the
+spec doesn't explicitly forbid), with a small, deliberate set of
+exceptions where a violation is rejected outright as `-MQTT_ERR_MALFORMED`
+rather than surfaced to the caller:
+
+- **MQTT-4.7.3-1**: a PUBLISH topic name must not be empty.
+- **MQTT-3.3.1-2**: DUP must be 0 on a QoS 0 PUBLISH. A malformed/
+  misbehaving broker (or a MITM'd packet) setting it anyway is rejected
+  rather than passed through with a DUP flag the QoS 0 delivery path has
+  no defined meaning for.
+
+Other spec requirements not on this list are intentionally left
+unenforced for now (a tolerant-receiver posture, not an oversight) -
+see issue tracker for anything still under discussion.
+
 ## QoS scope
 
 - **QoS 0** is fully supported both directions.
@@ -79,6 +96,16 @@ PINGREQ do. In practice this means a PINGREQ may go out slightly earlier
 than the theoretical minimum, never later, so it doesn't threaten the
 timeout guarantee above.
 
+All of the above assumes `now_ms` is monotonic. On the host it is
+(`CLOCK_MONOTONIC`); on the Amiga it is not (`DateStamp()` - see
+`src/tools/tool_clock.h`'s own comment and issue #8) - a clock change
+mid-session there can skew or spuriously trip any of these timeouts.
+Deliberately deferred: a real fix needs a genuinely monotonic source
+(`timer.device`), which would need per-connection state threaded through
+every caller of what is otherwise a stateless clock function, for a
+failure mode (something actively changing the clock mid-session) that's
+rare in practice.
+
 ## Session state
 
 `mqtt_client_connect()` always sets Clean Session (bit 1 of the CONNECT

src/amiga/transport_bsdsocket.c

@@ -12,14 +12,25 @@
 #include <netdb.h>
 #include <netinet/in.h>
 #include <sys/socket.h>
+#include <sys/ioctl.h> /* FIONBIO */
+#include <errno.h> /* EINPROGRESS, via bsdsocket.library's Errno() below */
 
 #include <string.h>
 
+#include "tool_clock.h"
+
 /* WaitSelect() poll interval: short enough that mqtt_client_process()'s
  * caller gets a timely wakeup to check keepalive scheduling, matching the
  * host transport's SO_RCVTIMEO of 1s. */
 #define MQTT_BSDSOCKET_POLL_SECS 1
 
+/* Overall connect-phase budget (issue #7) - same value and rationale as
+ * transport_amissl.c's/transport_openssl.c's handshake timeouts: bounds a
+ * host that accepts nothing (down, firewalled, wrong port) instead of
+ * relying on the TCP stack's own connect timeout (often 75s+) with no way
+ * to abort it in between. */
+#define MQTT_BSDSOCKET_CONNECT_TIMEOUT_MS 30000u
+
 /* proto/bsdsocket.h's macro-based inline stubs (inline/bsdsocket.h) call
  * through whatever C identifier named `SocketBase` is in scope at the call
  * site - they expand as plain text, e.g. `(struct Library *) (SocketBase)`,
@@ -114,6 +125,7 @@ int transport_bsdsocket_connect(mqtt_transport *out, bsdsocket_ctx *ctx,
     struct sockaddr_in addr;
     unsigned long ip;
     int fd;
+    long one = 1, zero = 0;
 
     ctx->fd = -1;
     ctx->ctrl_c = 0;
@@ -129,36 +141,94 @@ int transport_bsdsocket_connect(mqtt_transport *out, bsdsocket_ctx *ctx,
     addr.sin_family = AF_INET;
     addr.sin_port = htons(port);
 
+    /* gethostbyname() itself has no WaitSelect()/CTRL_C abortability of
+     * its own on this API - a DNS query against an unresponsive resolver
+     * blocks for whatever the stack's own resolver timeout is (issue #7's
+     * remaining gap; fixing it would need an async-resolve API this
+     * classic BSD sockets surface doesn't offer). The connect() below is
+     * the part that's actually fixed here. */
     ip = inet_addr((STRPTR)host);
     if (ip != (unsigned long)-1) {
         addr.sin_addr.s_addr = ip;
     } else {
         struct hostent *he = gethostbyname((STRPTR)host);
-        if (!he || !he->h_addr_list[0]) {
-            CloseLibrary(SocketBase);
-            ctx->socket_base = NULL;
-            return -1;
-        }
+        if (!he || !he->h_addr_list[0])
+            goto fail_socket_base;
         memcpy(&addr.sin_addr, he->h_addr_list[0], (size_t)he->h_length);
     }
 
     fd = socket(AF_INET, SOCK_STREAM, 0);
-    if (fd < 0) {
-        CloseLibrary(SocketBase);
-        ctx->socket_base = NULL;
-        return -1;
-    }
+    if (fd < 0)
+        goto fail_socket_base;
+
+    /* Non-blocking for the connect attempt only (issue #7) - a host that
+     * never responds (down, firewalled, wrong port) would otherwise block
+     * here for the TCP stack's own connect timeout (often 75s+) with no
+     * way to abort: unlike bsdsocket_recv()'s WaitSelect()-gated poll,
+     * connect() on a blocking socket has no signal mask of its own to
+     * pass SIGBREAKF_CTRL_C through. Reset to blocking again below once
+     * connected - bsdsocket_send() still assumes that mode (a plain
+     * blocking send(), no WaitSelect gating of its own). */
+    if (IoctlSocket(fd, FIONBIO, (char *)&one) < 0)
+        goto fail_fd;
+
     if (connect(fd, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
-        CloseSocket(fd);
-        CloseLibrary(SocketBase);
-        ctx->socket_base = NULL;
-        return -1;
+        uint32_t connect_start;
+
+        if (Errno() != EINPROGRESS)
+            goto fail_fd;
+
+        connect_start = tool_now_ms();
+        for (;;) {
+            struct timeval tv;
+            fd_set wfds;
+            ULONG sigmask = SIGBREAKF_CTRL_C;
+            long n;
+
+            FD_ZERO(&wfds);
+            FD_SET(fd, &wfds);
+            tv.tv_sec = MQTT_BSDSOCKET_POLL_SECS;
+            tv.tv_usec = 0;
+
+            n = WaitSelect(fd + 1, NULL, &wfds, NULL, &tv, &sigmask);
+            if (n > 0)
+                break; /* fd is writable - the attempt finished, either way */
+            if (n < 0) {
+                if (sigmask & SIGBREAKF_CTRL_C)
+                    ctx->ctrl_c = 1;
+                goto fail_fd;
+            }
+            if (tool_now_ms() - connect_start > MQTT_BSDSOCKET_CONNECT_TIMEOUT_MS)
+                goto fail_fd;
+        }
+
+        /* This bsdsocket.library NDK has no SO_ERROR/SOL_SOCKET to ask
+         * "did that connect actually succeed?" - getpeername() is the
+         * portable substitute: it only succeeds on an established
+         * connection, failing (typically ENOTCONN) if the peer refused or
+         * the attempt otherwise failed. */
+        {
+            struct sockaddr_in peer;
+            socklen_t peerlen = sizeof(peer);
+            if (getpeername(fd, (struct sockaddr *)&peer, &peerlen) < 0)
+                goto fail_fd;
+        }
     }
 
+    if (IoctlSocket(fd, FIONBIO, (char *)&zero) < 0)
+        goto fail_fd;
+
     ctx->fd = fd;
     out->ctx = ctx;
     out->send = bsdsocket_send;
     out->recv = bsdsocket_recv;
     out->close = bsdsocket_close;
     return 0;
+
+fail_fd:
+    CloseSocket(fd);
+fail_socket_base:
+    CloseLibrary(SocketBase);
+    ctx->socket_base = NULL;
+    return -1;
 }

src/core/mqtt_packet.c

@@ -335,6 +335,8 @@ int mqtt_decode(const uint8_t *buf, size_t avail, mqtt_packet *out)
             return -MQTT_ERR_MALFORMED;
         if (qos > 1)
             return -MQTT_ERR_PROTOCOL; /* QoS 2 out of scope */
+        if (qos == 0 && (flags & 0x08)) /* MQTT-3.3.1-2: DUP must be 0 for QoS 0 */
+            return -MQTT_ERR_MALFORMED;
         if (remlen < 2)
             return -MQTT_ERR_MALFORMED;
         tlen = (uint16_t)((content[0] << 8) | content[1]);

src/tools/tool_clock.h

@@ -3,19 +3,43 @@
 
 #include <stdint.h>
 
-/* A monotonic-ish millisecond counter, used as the `now_ms` the CLI tools
- * pump mqtt_client with (see mqtt_client.h) for keepalive/timeout
- * scheduling. One implementation per platform:
+/* A millisecond counter, used as the `now_ms` the CLI tools pump
+ * mqtt_client with (see mqtt_client.h) for keepalive/timeout scheduling.
+ * One implementation per platform:
  *
- *   src/host/clock.c   time() - fine on every host libc.
+ *   src/host/clock.c   clock_gettime(CLOCK_MONOTONIC, ...) - immune to
+ *                       wall-clock adjustments (NTP, DST, a manual clock
+ *                       set), which matters here: this value only ever
+ *                       feeds elapsed-time subtraction, never anything
+ *                       date-like.
  *   src/amiga/clock.c  dos.library DateStamp() - NOT time(): under some
  *                       AROS/Copperline HostSocket configurations, libnix's
  *                       time() was found to hang indefinitely before ever
  *                       returning (see docs/ARCHITECTURE.md's testing
  *                       notes), apparently entangled with its ANSI-locale
  *                       auto-load at first call. DateStamp() is dos.library's
  *                       own native clock, always available, no such
- *                       dependency.
+ *                       dependency - BUT, unlike the host side, it is
+ *                       wall-clock, not monotonic (classic AmigaOS has no
+ *                       cheap equivalent of CLOCK_MONOTONIC - see issue
+ *                       #8). A clock change mid-session (SetClock, an
+ *                       NTP-driven utility, a timezone/DST correction)
+ *                       skews every keepalive/PUBACK/SUBACK/backoff
+ *                       deadline computed from it: stepping the clock
+ *                       either direction makes the next elapsed-time
+ *                       subtraction wrap to a huge unsigned value,
+ *                       triggering an immediate spurious timeout - there
+ *                       is no way to distinguish that from 49 days of
+ *                       genuine `uint32_t` wraparound from the delta
+ *                       alone. A real fix needs a genuinely monotonic
+ *                       source (timer.device's TR_GETSYSTIME is the
+ *                       candidate), which needs a per-task IORequest -
+ *                       i.e. state threaded through every caller of this
+ *                       otherwise-stateless function, not a drop-in
+ *                       replacement. Deliberately deferred (issue #8):
+ *                       accepted as a documented caveat rather than an
+ *                       invasive API change, since a clock actively
+ *                       changing mid-session is rare in practice.
  *
  * Absolute value and epoch don't matter - callers only ever compare two
  * readings' difference. */

tests/test_codec.c

@@ -303,6 +303,17 @@ static void test_decode_publish_rejects_empty_topic(void)
     TEST_CHECK(mqtt_decode(bad, sizeof(bad) - 1, &pkt) == -MQTT_ERR_MALFORMED);
 }
 
+/* MQTT-3.3.1-2: DUP must be 0 for a QoS 0 PUBLISH (issue #10). */
+static void test_decode_publish_rejects_dup_with_qos0(void)
+{
+    mqtt_packet pkt;
+    uint8_t bad[sizeof(V_PUBLISH_QOS0)];
+
+    memcpy(bad, V_PUBLISH_QOS0, sizeof(bad));
+    bad[0] |= 0x08; /* set DUP; QoS bits (2-1) stay 0 */
+    TEST_CHECK(mqtt_decode(bad, sizeof(bad), &pkt) == -MQTT_ERR_MALFORMED);
+}
+
 static void test_truncation_sweeps(void)
 {
     check_truncation_sweep(V_CONNACK_OK, sizeof(V_CONNACK_OK));
@@ -325,5 +336,6 @@ void run_codec_tests(void)
     test_decode_rejects_client_to_broker_types();
     test_decode_malformed_flags();
     test_decode_publish_rejects_empty_topic();
+    test_decode_publish_rejects_dup_with_qos0();
     test_truncation_sweeps();
 }