From efc5173d93fe311e1ea38f144b2e7ed869ac70d9 Mon Sep 17 00:00:00 2001 From: Mario Fetka Date: Thu, 23 Jul 2026 09:49:21 +0200 Subject: [PATCH] Add queue hold release and listing controls --- include/nmap.h | 1 + src/agents/queue/messages.h | 3 + src/agents/queue/queue.c | 144 ++++++++++++++++-- src/agents/queue/queue.h | 1 + src/agents/queue/queued.c | 1 + src/apps/queuetool/bongo-queuetool | 63 ++++++++ src/apps/queuetool/tests/test_queue_client.py | 32 ++++ src/libs/python/bongo/store/QueueClient.py | 36 ++++- 8 files changed, 270 insertions(+), 11 deletions(-) diff --git a/include/nmap.h b/include/nmap.h index 0906e1a..7d15b85 100644 --- a/include/nmap.h +++ b/include/nmap.h @@ -353,6 +353,7 @@ typedef enum _NMAPMessageFlags { #define Q_OUTGOING 7 #define Q_RTS 8 #define Q_DIGEST 9 +#define Q_HOLD 999 /* Delivery Status Notification and other Control File Flags */ #define DSN_FLAGS (0x0000001F) diff --git a/src/agents/queue/messages.h b/src/agents/queue/messages.h index f3c0dca..26af16e 100644 --- a/src/agents/queue/messages.h +++ b/src/agents/queue/messages.h @@ -49,6 +49,9 @@ #define NMAP_QINFO_COMMAND "QINFO" +#define NMAP_QLIST_COMMAND "QLIST" +#define NMAP_QLIST_HELP "QLIST - Lists committed message queue entries.\r\n" + #define NMAP_QMOD_FROM_COMMAND "QMOD FROM" #define NMAP_QMOD_FLAGS_COMMAND "QMOD FLAGS" diff --git a/src/agents/queue/queue.c b/src/agents/queue/queue.c index 751daab..362c358 100644 --- a/src/agents/queue/queue.c +++ b/src/agents/queue/queue.c @@ -2599,6 +2599,12 @@ CheckQueue(void *queueIn) found++; QueueFormat(path, "%.3s%.7s", dirEntry->d_nameDOS + 9, dirEntry->d_nameDOS + 1); + + /* Held messages remain committed and inspectable, but neither + * the periodic monitor nor QFLUSH may process them. */ + if (atoi(path) == Q_HOLD) { + continue; + } /* skip recently-touched messages that are in the outgoing * queue (queue 7) */ @@ -2795,7 +2801,7 @@ CreateQueueThreads(BOOL failed) count += 2; } else if (failed) { QueueFormat(path, "%s/c%s", Conf.spoolPath, dirEntry->d_nameDOS + 1); - for (i = 0; i < 10; i++) { + for (i = 0; i <= Q_HOLD; i++) { if (strlen(path) >= 3) { size_t extensionOffset = strlen(path) - 3; QueueFormatBuffer(path + extensionOffset, @@ -2809,7 +2815,7 @@ CreateQueueThreads(BOOL failed) } } - if (i == 10) { + if (i > Q_HOLD) { /* Check for matching C file */ fprintf(killFile, "%s/d%s\r\n", Conf.spoolPath, dirEntry->d_nameDOS + 1); @@ -3707,6 +3713,99 @@ CommandQinfo(void *param) (intmax_t)(sb.st_size - count))); /* BodySize */ } +static int +QueueListEntryCompare(const void *left, const void *right) +{ + const char *const *leftEntry = left; + const char *const *rightEntry = right; + + return strcmp(*leftEntry, *rightEntry); +} + +int +CommandQlist(void *param) +{ + QueueClient *client = (QueueClient *)param; + XplDir *dirP; + XplDir *dirEntry; + GPtrArray *entries; + GString *listing; + unsigned int i; + int ccode; + + if (client->buffer[5] != '\0') { + return ConnWrite(client->conn, MSG3010BADARGC, + sizeof(MSG3010BADARGC) - 1); + } + + entries = g_ptr_array_new_with_free_func(g_free); + listing = g_string_new(NULL); + dirP = XplOpenDir(Conf.spoolPath); + if (!dirP) { + g_ptr_array_free(entries, TRUE); + g_string_free(listing, TRUE); + return ConnWrite(client->conn, MSG4224CANTREAD, + sizeof(MSG4224CANTREAD) - 1); + } + + while ((dirEntry = XplReadDir(dirP)) != NULL) { + const char *name = dirEntry->d_nameDOS; + char queueID[12]; + char dataPath[XPL_MAX_PATH + 1]; + struct stat dataStat; + size_t position; + BOOL valid = TRUE; + + if (strlen(name) != 12 || toupper(name[0]) != 'C' || + name[8] != '.') { + continue; + } + for (position = 1; position < 8; position++) { + if (!isxdigit((unsigned char)name[position])) { + valid = FALSE; + break; + } + } + for (position = 9; valid && position < 12; position++) { + if (!isdigit((unsigned char)name[position])) { + valid = FALSE; + } + } + if (!valid) { + continue; + } + + snprintf(queueID, sizeof(queueID), "%.3s-%.7s", name + 9, name + 1); + QueueFormat(dataPath, "%s/d%.7s.msg", Conf.spoolPath, name + 1); + if (stat(dataPath, &dataStat) != 0) { + continue; + } + g_ptr_array_add( + entries, + g_strdup_printf("%s %" PRIdMAX "\r\n", queueID, + (intmax_t)dataStat.st_size)); + } + XplCloseDir(dirP); + + g_ptr_array_sort(entries, QueueListEntryCompare); + for (i = 0; i < entries->len; i++) { + g_string_append(listing, g_ptr_array_index(entries, i)); + } + + ccode = ConnWriteF(client->conn, "2024 %lu Queue list follows\r\n", + (unsigned long)listing->len); + if (ccode != -1 && listing->len) { + ccode = ConnWrite(client->conn, listing->str, listing->len); + } + if (ccode != -1) { + ccode = ConnWrite(client->conn, MSG1000OK, sizeof(MSG1000OK) - 1); + } + + g_ptr_array_free(entries, TRUE); + g_string_free(listing, TRUE); + return ccode; +} + int CommandQmodFrom(void *param) { @@ -3942,6 +4041,8 @@ int CommandQmove(void *param) { int ccode; + unsigned long entryID; + unsigned long target; char *ptr; char *ptr2; char path[XPL_MAX_PATH + 1]; @@ -3966,20 +4067,45 @@ CommandQmove(void *param) return(ConnWrite(client->conn, MSG3010BADARGC, sizeof(MSG3010BADARGC) - 1)); } + if (!*ptr2) { + return(ConnWrite(client->conn, MSG3010BADARGC, + sizeof(MSG3010BADARGC) - 1)); + } + { + char *end = NULL; + target = strtoul(ptr2, &end, 10); + if (!end || *end != '\0') { + return(ConnWrite(client->conn, MSG3010BADARGC, + sizeof(MSG3010BADARGC) - 1)); + } + } + if (target > Q_HOLD) { + return(ConnWrite(client->conn, MSG4000OUTOFRANGE, + sizeof(MSG4000OUTOFRANGE) - 1)); + } + + entryID = strtoul(ptr + 4, NULL, 16); + if (!SpoolEntryIDLock(entryID)) { + return(ConnWrite(client->conn, MSG4000CANTUNLOCKENTRY, + sizeof(MSG4000CANTUNLOCKENTRY) - 1)); + } + QueueFormat(client->path, "%s/c%s.%s", Conf.spoolPath, ptr + 4, ptr); if (stat(client->path, &sb) == 0) { - QueueFormat(path, "%s/c%s.%03d", Conf.spoolPath, ptr + 4, atoi(ptr)); + QueueFormat(path, "%s/c%s.%03lu", Conf.spoolPath, ptr + 4, target); - RENAME_CHECK(client->path, path); - - ccode = ConnWrite(client->conn, MSG1000OK, sizeof(MSG1000OK) - 1); + if (rename(client->path, path) == 0) { + ccode = ConnWriteF(client->conn, "1000 %03lu-%s OK\r\n", + target, ptr + 4); + } else { + ccode = ConnWrite(client->conn, MSG4224CANTREAD, + sizeof(MSG4224CANTREAD) - 1); + } } else { - QueueFormat(client->path, "%s/d%s.msg", Conf.spoolPath, ptr + 4); - UNLINK_CHECK(client->path); - ccode = ConnWrite(client->conn, MSG4224NOENTRY, sizeof(MSG4224NOENTRY) - 1); } + SpoolEntryIDUnlock(entryID); return(ccode); } diff --git a/src/agents/queue/queue.h b/src/agents/queue/queue.h index b3f471b..fcf483e 100644 --- a/src/agents/queue/queue.h +++ b/src/agents/queue/queue.h @@ -122,6 +122,7 @@ int CommandQend(void *param); int CommandQgrep(void *param); int CommandQhead(void *param); int CommandQinfo(void *param); +int CommandQlist(void *param); int CommandQmodFrom(void *param); int CommandQmodFlags(void *param); int CommandQmodLocal(void *param); diff --git a/src/agents/queue/queued.c b/src/agents/queue/queued.c index db9e32f..6b328e5 100644 --- a/src/agents/queue/queued.c +++ b/src/agents/queue/queued.c @@ -74,6 +74,7 @@ static ProtocolCommand commands[] = { { NMAP_QGREP_COMMAND, NMAP_HELP_NOT_DEFINED, sizeof(NMAP_QGREP_COMMAND) - 1, CommandQgrep, NULL, NULL, NULL, BlackCommand }, { NMAP_QHEAD_COMMAND, NMAP_HELP_NOT_DEFINED, sizeof(NMAP_QHEAD_COMMAND) - 1, CommandQhead, NULL, NULL, NULL, BlackCommand }, { NMAP_QINFO_COMMAND, NMAP_HELP_NOT_DEFINED, sizeof(NMAP_QINFO_COMMAND) - 1, CommandQinfo, NULL, NULL, NULL, BlackCommand }, + { NMAP_QLIST_COMMAND, NMAP_QLIST_HELP, sizeof(NMAP_QLIST_COMMAND) - 1, CommandQlist, NULL, NULL, NULL, BlackCommand }, { NMAP_QMOD_FROM_COMMAND, NMAP_HELP_NOT_DEFINED, sizeof(NMAP_QMOD_FROM_COMMAND) - 1, CommandQmodFrom, NULL, NULL, NULL, BlackCommand }, { NMAP_QMOD_FLAGS_COMMAND, NMAP_HELP_NOT_DEFINED, sizeof(NMAP_QMOD_FLAGS_COMMAND) - 1, CommandQmodFlags, NULL, NULL, NULL, BlackCommand }, { NMAP_QMOD_LOCAL_COMMAND, NMAP_HELP_NOT_DEFINED, sizeof(NMAP_QMOD_LOCAL_COMMAND) - 1, CommandQmodLocal, NULL, NULL, NULL, BlackCommand }, diff --git a/src/apps/queuetool/bongo-queuetool b/src/apps/queuetool/bongo-queuetool index 34c99e4..1b46e99 100755 --- a/src/apps/queuetool/bongo-queuetool +++ b/src/apps/queuetool/bongo-queuetool @@ -94,6 +94,24 @@ class ShowCommand(Command): data = queue.RetrieveInfo(args[0]) sys.stdout.buffer.write(data) +class ListCommand(Command): + log = logging.getLogger("Bongo.QueueTool") + + def __init__(self): + Command.__init__(self, "list", aliases=["l"], + summary="List committed queue entries", + usage="%prog %cmd") + + def Run(self, options, args): + if args: + self.error("list does not accept arguments") + queue = QueueClient(host=options.host, port=options.port) + try: + for queue_id, size in queue.List(): + print(f"{queue_id} {size}") + finally: + queue.Quit() + class MessageCommand(Command): log = logging.getLogger("Bongo.QueueTool") @@ -181,13 +199,58 @@ class DeleteCommand(Command): finally: queue.Quit() +class HoldCommand(Command): + log = logging.getLogger("Bongo.QueueTool") + + def __init__(self): + Command.__init__(self, "hold", + summary="Move a committed entry to the hold queue", + usage="%prog %cmd ") + + def Run(self, options, args): + if len(args) != 1: + self.error("exactly one queue ID is required") + queue = QueueClient(host=options.host, port=options.port) + try: + response = queue.Move(args[0], 999) + print(response.message.split(" ", 1)[0]) + finally: + queue.Quit() + +class ReleaseCommand(Command): + log = logging.getLogger("Bongo.QueueTool") + + def __init__(self): + Command.__init__( + self, "release", + summary="Release a held entry and process it immediately", + usage="%prog %cmd []") + + def Run(self, options, args): + if len(args) < 1 or len(args) > 2: + self.error("a queue ID and optional target are required") + if not args[0].startswith("999-"): + self.error("release requires an entry from queue 999") + target = int(args[1]) if len(args) == 2 else 0 + queue = QueueClient(host=options.host, port=options.port) + try: + response = queue.Move(args[0], target) + released_id = response.message.split(" ", 1)[0] + queue.Run(released_id) + print(released_id) + finally: + queue.Quit() + parser.add_command(FlushCommand(), "Queue Management") parser.add_command(SpaceCommand(), "Queue Management") +parser.add_command(ListCommand(), "Queue Management") parser.add_command(ShowCommand(), "Queue Management") parser.add_command(MessageCommand(), "Queue Management") parser.add_command(HoldLocalCommand(), "Queue Recovery and Diagnostics") parser.add_command(DeliverLocalCommand(), "Queue Recovery and Diagnostics") parser.add_command(DeleteCommand(), "Queue Recovery and Diagnostics") +parser.add_command(HoldCommand(), "Queue Recovery and Diagnostics") +parser.add_command(ReleaseCommand(), "Queue Recovery and Diagnostics") if __name__ == '__main__': (command, options, args) = parser.parse_args() diff --git a/src/apps/queuetool/tests/test_queue_client.py b/src/apps/queuetool/tests/test_queue_client.py index ea36949..a371aed 100644 --- a/src/apps/queuetool/tests/test_queue_client.py +++ b/src/apps/queuetool/tests/test_queue_client.py @@ -111,6 +111,38 @@ class QueueClientTests(unittest.TestCase): self.assertEqual(client.stream.commands, ["QDELE 999-deadbee"]) + def test_list_reads_declared_payload(self): + data = b"007-deadbee 42\r\n999-cafebad 81\r\n" + client = QueueClient.__new__(QueueClient) + client.stream = Stream( + [Response(2024, f"{len(data)} Queue list follows"), + Response(1000, "OK")], + data, + ) + + self.assertEqual( + client.List(), + [("007-deadbee", 42), ("999-cafebad", 81)], + ) + self.assertEqual(client.stream.commands, ["QLIST"]) + self.assertEqual(client.stream.lengths, [len(data)]) + + def test_move_uses_qmove(self): + client = QueueClient.__new__(QueueClient) + client.stream = Stream([Response(1000, "999-deadbee OK")], b"") + + client.Move("007-deadbee", 999) + + self.assertEqual(client.stream.commands, ["QMOVE 007-deadbee 999"]) + + def test_run_accepts_existing_queue_id(self): + client = QueueClient.__new__(QueueClient) + client.stream = Stream([Response(1000, "000-deadbee OK")], b"") + + client.Run("000-deadbee") + + self.assertEqual(client.stream.commands, ["QRUN 000-deadbee"]) + def test_retrieve_info_reads_declared_payload(self): client = QueueClient.__new__(QueueClient) client.stream = Stream( diff --git a/src/libs/python/bongo/store/QueueClient.py b/src/libs/python/bongo/store/QueueClient.py index 6d3afd8..4941087 100644 --- a/src/libs/python/bongo/store/QueueClient.py +++ b/src/libs/python/bongo/store/QueueClient.py @@ -69,6 +69,35 @@ class QueueClient: raise bongo.BongoError(r.message) return r + def List(self): + self.stream.Write("QLIST") + r = self.stream.GetResponse() + if r.code != 2024: + raise bongo.BongoError(r.message) + try: + length = int(r.message.split(" ", 1)[0]) + except (ValueError, IndexError): + raise bongo.BongoError("Malformed queue list response") + data = self.stream.Read(length) + r = self.stream.GetResponse() + if r.code != 1000: + raise bongo.BongoError(r.message) + result = [] + for line in data.decode("ascii").splitlines(): + queue_id, size = line.split() + result.append((queue_id, int(size))) + return result + + def Move(self, queue_id, target): + target = int(target) + if target < 0 or target > 999: + raise ValueError("queue target must be between 0 and 999") + self.stream.Write("QMOVE %s %d" % (queue_id, target)) + r = self.stream.GetResponse() + if r.code != 1000: + raise bongo.BongoError(r.message) + return r + def RetrieveInfo(self, queue_id): self.stream.Write("QRETR %s INFO" % queue_id) r = self.stream.GetResponse() @@ -107,8 +136,11 @@ class QueueClient: self.stream.Write("QUIT") self.stream.close() - def Run(self): - self.stream.Write("QRUN") + def Run(self, queue_id=None): + command = "QRUN" + if queue_id is not None: + command += " %s" % queue_id + self.stream.Write(command) r = self.stream.GetResponse() if r.code != 1000: raise bongo.BongoError(r.message)