diff --git a/embed/rayforce_q.c b/embed/rayforce_q.c index ec886da..23e825f 100644 --- a/embed/rayforce_q.c +++ b/embed/rayforce_q.c @@ -33,6 +33,7 @@ #include "lang/eval.h" /* ray_fn_*, RAY_FN_NONE */ #include +#include #include #include @@ -66,16 +67,12 @@ 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; @@ -83,6 +80,15 @@ static int64_t q_atom_i64(ray_t *a, int *ok) { } } +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; } diff --git a/q.c b/q.c index 0669269..78cdba7 100644 --- a/q.c +++ b/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; diff --git a/q_server.c b/q_server.c index 4e19cbc..69545f3 100644 --- a/q_server.c +++ b/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) { diff --git a/test/driver.c b/test/driver.c index 8bdf204..c923fea 100644 --- a/test/driver.c +++ b/test/driver.c @@ -33,6 +33,8 @@ #include #include #include +#include +#include #include /* 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"); diff --git a/test/rfl/client/06_errors.rfl b/test/rfl/client/06_errors.rfl index 903a32d..e12a95a 100644 --- a/test/rfl/client/06_errors.rfl +++ b/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 diff --git a/test/rfl/client/07_auth.rfl b/test/rfl/client/07_auth.rfl index c524818..a2eb5c4 100644 --- a/test/rfl/client/07_auth.rfl +++ b/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 diff --git a/test/run.sh b/test/run.sh index 9cd1912..a7f5cf0 100755 --- a/test/run.sh +++ b/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("