fix(q): validate handles and wire protocol frames

embed/rayforce_q.c

@@ -33,6 +33,7 @@
 #include "lang/eval.h"    /* ray_fn_*, RAY_FN_NONE               */
 
 #include <errno.h>
+#include <limits.h>
 #include <stdlib.h>
 #include <string.h>
 
@@ -66,23 +67,28 @@ static int64_t q_atom_i64(ray_t *a, int *ok) {
   }
   switch (a->type) {
   case -RAY_I64:
-  case -RAY_TIMESTAMP:
     return a->i64;
   case -RAY_I32:
-  case -RAY_DATE:
-  case -RAY_TIME:
     return a->i32;
   case -RAY_I16:
     return a->i16;
   case -RAY_U8:
-  case -RAY_BOOL:
     return a->u8;
   default:
     *ok = 0;
     return 0;
   }
 }
 
+static int64_t q_handle_i64(ray_t *a, int *ok) {
+  if (a != NULL && a->type == -RAY_I64 && a->i64 >= 0) {
+    *ok = 1;
+    return a->i64;
+  }
+  *ok = 0;
+  return 0;
+}
+
 static void q_str_arg(ray_t *a, char *buf, size_t cap) {
   buf[0] = '\0';
   if (a == NULL || a->type != -RAY_STR)
@@ -115,15 +121,24 @@ static ray_t *qb_connect(ray_t **args, int64_t n) {
   memcpy(host, ray_str_ptr(args[0]), hn);
   host[hn] = '\0';
 
+  if (n >= 3 && (args[2] == NULL || args[2]->type != -RAY_STR))
+    return ray_error("type", ".q.connect: user must be a string");
+  if (n >= 4 && (args[3] == NULL || args[3]->type != -RAY_STR))
+    return ray_error("type", ".q.connect: password must be a string");
+
   char user[128], password[128];
   q_str_arg(n >= 3 ? args[2] : NULL, user, sizeof user);
   q_str_arg(n >= 4 ? args[3] : NULL, password, sizeof password);
   int timeout_ms = 0;
   if (n >= 5) {
     int tok;
-    timeout_ms = (int)q_atom_i64(args[4], &tok);
+    int64_t timeout_value = q_atom_i64(args[4], &tok);
     if (!tok)
-      timeout_ms = 0;
+      return ray_error("type", ".q.connect: timeout must be an integer");
+    if (timeout_value < 0 || timeout_value > INT_MAX)
+      return ray_error("range",
+                       ".q.connect: timeout must be in 0..INT_MAX ms");
+    timeout_ms = (int)timeout_value;
   }
 
   int fd = q_connect(host, (int)port, user, password, timeout_ms);
@@ -153,9 +168,9 @@ static ray_t *qb_connect(ray_t **args, int64_t n) {
 /* (.q.send handle msg) -> decoded response (may itself be a Q server error). */
 static ray_t *qb_send(ray_t *handle, ray_t *msg) {
   int ok;
-  int64_t fd = q_atom_i64(handle, &ok);
+  int64_t fd = q_handle_i64(handle, &ok);
   if (!ok)
-    return ray_error("type", ".q.send: handle must be an integer");
+    return ray_error("type", ".q.send: handle must be a non-negative i64");
 
   ray_poll_t *poll = q_poll();
   if (poll != NULL)
@@ -174,15 +189,18 @@ static ray_t *qb_send(ray_t *handle, ray_t *msg) {
 /* (.q.close handle) -> null. */
 static ray_t *qb_close(ray_t *handle) {
   int ok;
-  int64_t fd = q_atom_i64(handle, &ok);
+  int64_t fd = q_handle_i64(handle, &ok);
   if (!ok)
-    return ray_error("type", ".q.close: handle must be an integer");
+    return ray_error("type", ".q.close: handle must be a non-negative i64");
 
   ray_poll_t *poll = q_poll();
-  if (poll != NULL)
+  if (poll != NULL) {
+    if (ray_poll_get(poll, fd) == NULL)
+      return ray_error("handle", ".q.close: not an open connection");
     q_conn_close(poll, fd);
-  else
-    q_close((int)fd);
+  } else if (q_close((int)fd) < 0) {
+    return ray_error("handle", ".q.close: not an open connection");
+  }
   return RAY_NULL_OBJ;
 }
 

q.c

@@ -103,6 +103,7 @@ static inline size_t ray_scalar_elem_size(int8_t type) {
 
 #define Q_MSG_SYNC 1
 #define Q_MSG_RESPONSE 2
+#define Q_CAP_MAX 3
 #define Q_MAX_BODY ((int64_t)256 << 20)
 
 typedef struct {
@@ -1058,6 +1059,8 @@ static ray_t *q_des_obj(uint8_t **buf, int64_t *len) {
 
   case Q_XT: { /* table = attrs(0) + dict_marker(99) + keys + values */
     Q_NEED(2);
+    if ((*buf)[0] != 0 || (*buf)[1] != Q_XD)
+      return ray_error("q: malformed table marker", NULL);
     (*buf) += 2;
     *len -= 2;
     ray_t *keys = q_des_obj(buf, len);
@@ -1138,7 +1141,7 @@ static int q_decompress(const uint8_t *src, int64_t src_len, uint8_t **out_buf,
   uint32_t header_size;
   memcpy(&header_size, src, 4);
   int64_t out_size = (int64_t)header_size - (int64_t)sizeof(q_header_t);
-  if (out_size <= 0)
+  if (out_size <= 0 || out_size > Q_MAX_BODY)
     return -1;
 
   uint32_t buffer[256] = {0};
@@ -1253,6 +1256,10 @@ int q_connect(const char *host, int port, const char *user,
     close(fd);
     return Q_ERR_HANDSHAKE;
   }
+  if (cap > Q_CAP_MAX) {
+    close(fd);
+    return Q_ERR_HANDSHAKE;
+  }
   return fd;
 }
 
@@ -1373,7 +1380,9 @@ ray_t *q_decode(uint8_t *resp, int64_t resp_len, int compressed, char *err,
     free(decompressed);
   if (result == NULL)
     q_set_err(err, errlen, "q: deserialization returned null");
-  else if (remaining != 0) {
+  else if (RAY_IS_ERR(result)) {
+    return result;
+  } else if (remaining != 0) {
     q_release_any(result);
     q_set_err(err, errlen, "q: trailing bytes after object");
     return NULL;

q_server.c

@@ -62,6 +62,7 @@ typedef struct {
 
 #define Q_LITTLE_ENDIAN 1
 #define Q_MSG_ASYNC 0
+#define Q_MSG_SYNC 1
 #define Q_MSG_RESPONSE 2
 #define Q_CAP_MAX 3                     /* matches q.c client capability  */
 #define Q_MAX_BODY ((int64_t)256 << 20) /* reject absurd frames (256 MiB) */
@@ -269,6 +270,13 @@ static ray_t *q_read_body(ray_poll_t *poll, ray_selector_t *sel) {
   sel->rx.read_fn = q_read_header;
   ray_poll_rx_request(poll, sel, (int64_t)sizeof(q_header_t));
 
+  if (hdr.msgtype != Q_MSG_ASYNC && hdr.msgtype != Q_MSG_SYNC &&
+      hdr.msgtype != Q_MSG_RESPONSE) {
+    q_release_any(req);
+    ray_poll_deregister(poll, id);
+    return NULL;
+  }
+
   /* A RESPONSE belongs to the q_conn_send parked on this connection — it is
    * data, not something to evaluate. */
   if (hdr.msgtype == Q_MSG_RESPONSE) {

test/driver.c

@@ -33,6 +33,8 @@
 #include <stdlib.h>
 #include <string.h>
 #include <sys/socket.h>
+#include <netinet/in.h>
+#include <sys/wait.h>
 #include <unistd.h>
 
 /* Registers `.q.connect` / `.q.send` / `.q.close` */
@@ -333,11 +335,94 @@ static int run_codec_selftest(void) {
   }
   release_any(r);
 
+  err[0] = '\0';
+  uint8_t bad_table_marker[] = {98, 0, 0, 11, 0, 0, 0, 0, 0, 0, 0, 0,
+                                0,  0, 0, 0,  0, 0, 0, 0};
+  r = q_decode(bad_table_marker, (int64_t)sizeof bad_table_marker, 0, err,
+               sizeof err);
+  ray_t *rs = r ? ray_fmt(r, 0) : NULL;
+  const char *rp = rs ? ray_str_ptr(rs) : err;
+  if (r == NULL || !RAY_IS_ERR(r) || strstr(rp, "q: malf") == NULL) {
+    fprintf(stderr, "codec selftest: malformed table marker was accepted\n");
+    failures++;
+  }
+  if (rs)
+    ray_release(rs);
+  release_any(r);
+
+  uint8_t oversized_compressed[4];
+  uint32_t oversized_size =
+      (uint32_t)((256u << 20) + sizeof(test_q_header_t) + 1u);
+  memcpy(oversized_compressed, &oversized_size, sizeof oversized_size);
+  err[0] = '\0';
+  r = q_decode(oversized_compressed, (int64_t)sizeof oversized_compressed, 1,
+               err, sizeof err);
+  if (r != NULL || strstr(err, "decompression failed") == NULL) {
+    fprintf(stderr, "codec selftest: oversized compressed body was accepted\n");
+    failures++;
+  }
+  release_any(r);
+
   if (q_connect("127.0.0.1", 70000, "", "", 1) != Q_ERR_SOCKET) {
     fprintf(stderr, "codec selftest: client accepted out-of-range port\n");
     failures++;
   }
 
+  int listener = socket(AF_INET, SOCK_STREAM, 0);
+  if (listener < 0) {
+    perror("codec selftest: handshake socket");
+    failures++;
+  } else {
+    struct sockaddr_in addr = {0};
+    addr.sin_family = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    addr.sin_port = 0;
+    if (bind(listener, (struct sockaddr *)&addr, sizeof addr) < 0 ||
+        listen(listener, 1) < 0) {
+      perror("codec selftest: handshake listener");
+      failures++;
+      close(listener);
+    } else {
+      socklen_t addr_len = sizeof addr;
+      if (getsockname(listener, (struct sockaddr *)&addr, &addr_len) < 0) {
+        perror("codec selftest: handshake address");
+        close(listener);
+        failures++;
+        goto handshake_done;
+      }
+      pid_t child = fork();
+      if (child < 0) {
+        perror("codec selftest: handshake fork");
+        close(listener);
+        failures++;
+        goto handshake_done;
+      } else if (child == 0) {
+        int peer = accept(listener, NULL, NULL);
+        if (peer >= 0) {
+          uint8_t b;
+          while (recv(peer, &b, 1, 0) == 1 && b != 0)
+            ;
+          b = 0xff;
+          send(peer, &b, 1, 0);
+          close(peer);
+        }
+        close(listener);
+        _exit(0);
+      }
+      close(listener);
+      int bad_cap =
+          q_connect("127.0.0.1", ntohs(addr.sin_port), "", "", 1000);
+      if (bad_cap != Q_ERR_HANDSHAKE) {
+        fprintf(stderr, "codec selftest: invalid handshake capability accepted\n");
+        if (bad_cap >= 0)
+          q_close(bad_cap);
+        failures++;
+      }
+      waitpid(child, NULL, 0);
+    }
+  }
+handshake_done:
+
   ray_poll_t *poll = ray_poll_create();
   if (poll == NULL) {
     fprintf(stderr, "codec selftest: failed to create poll\n");

test/rfl/client/06_errors.rfl

@@ -7,6 +7,11 @@
 
 (set h (.q.connect qhost qport))
 
+;; Handles are the exact non-negative i64 values returned by .q.connect.
+(.q.send true "1+1") !- type
+(.q.close true) !- type
+(.q.close 999999) !- handle
+
 ;; explicit q signal: 'myerr  -> error text "myerr"
 (.q.send h "'myerr") !- myerr
 

test/rfl/client/07_auth.rfl

@@ -13,3 +13,12 @@
 
 ;; no credentials at all against an auth server -> rejected
 (.q.connect qhost qauthport) !- auth
+
+;; credentials must be strings; invalid values must fail before any socket I/O.
+(.q.connect qhost qauthport 123 quser) !- type
+(.q.connect qhost qauthport quser 456) !- type
+
+;; timeout is a non-negative C int number of milliseconds.
+(.q.connect qhost qport "" "" -1) !- range
+(.q.connect qhost qport "" "" 2147483648) !- range
+(.q.connect qhost qport "" "" "1000") !- type

test/run.sh

@@ -121,6 +121,23 @@ if resp[8] != 0x80:
     raise SystemExit(f"malformed-frame test: expected Q error, got {resp.hex()}")
 PY
 
+echo "checking unknown Q message type handling..."
+python3 - "$HOST" "$SERVERPORT" <<'PY'
+import socket
+import struct
+import sys
+
+host, port = sys.argv[1], int(sys.argv[2])
+with socket.create_connection((host, port), 1.0) as s:
+    s.sendall(bytes([3, 0]))
+    if s.recv(1) != bytes([3]):
+        raise SystemExit("bad Q handshake response")
+    # Q identity is a valid body; only the message type is invalid.
+    s.sendall(struct.pack("<BBBBI", 1, 9, 0, 0, 10) + bytes([101, 0]))
+    if s.recv(1) != b'':
+        raise SystemExit("unknown message type was executed or answered")
+PY
+
 # ---- Leg 2: real-q interop against Rayforce server
 find_q
 if [[ -n "$QBIN" ]]; then