diff --git a/contrib/testing/README.md b/contrib/testing/README.md index 10566d1..5039cf5 100644 --- a/contrib/testing/README.md +++ b/contrib/testing/README.md @@ -106,6 +106,23 @@ reserve was subtracted without 32-bit truncation: ./contrib/testing/queue-disk-space-check.sh ``` +The bounded collector flood check seeds separate POP3 and IMAP Cyrus source +accounts with 128-KiB messages, delivers them against a small Store quota, and +then temporarily lowers the Collector per-user pending limit to one. It +verifies unique delivery across Store and Queue, exact quota enforcement, and +that the first over-limit source message remains at Cyrus: + +```sh +export BONGO_ALLOW_LIVE_USER_TEST=1 +export BONGO_TEST_USER=stqquota +export BONGO_TEST_PASSWORD='a-disposable-password-of-at-least-12-characters' +./contrib/testing/collector-flood-quota-check.py +``` + +The script restores quota and Collector configuration and removes its source, +Queue, Store, and external-account fixtures. Use +`BONGO_TEST_CLEANUP_ONLY=1` to clean up a previously interrupted run. + The Queue failure-safety check retains one isolated message in an 8 MiB tmpfs fixture while it first fills that filesystem below the operative reserve and then overlays the spool with an unreadable directory. It verifies diff --git a/contrib/testing/collector-flood-quota-check.py b/contrib/testing/collector-flood-quota-check.py new file mode 100755 index 0000000..6e65c0a --- /dev/null +++ b/contrib/testing/collector-flood-quota-check.py @@ -0,0 +1,641 @@ +#!/usr/bin/env python3 +# This program is free software, licensed under the terms of the GNU GPL. +# See the Bongo COPYING file for full details. +# Copyright (c) 2026 Bongo Project contributors + +"""Exercise bounded POP3/IMAP collection at quota and Queue boundaries.""" + +from __future__ import annotations + +import http.cookiejar +import imaplib +import json +import os +from pathlib import Path +import re +import socket +import ssl +import subprocess +import sys +import time +import urllib.error +import urllib.parse +import urllib.request + +from bongo.store.StoreClient import StoreClient + + +USER = os.environ.get("BONGO_TEST_USER", "stqquota") +PASSWORD = os.environ.get("BONGO_TEST_PASSWORD", "") +DOMAIN = os.environ.get("BONGO_TEST_DOMAIN", "bongo.test") +WEB_URL = os.environ.get( + "BONGO_TEST_WEB_URL", "http://172.16.11.190:8080" +).rstrip("/") +STORE_HOST = os.environ.get("BONGO_TEST_STORE_HOST", "127.0.0.1") +STORE_PORT = int(os.environ.get("BONGO_TEST_STORE_PORT", "689")) +QUEUE_TOOL = os.environ.get( + "BONGO_TEST_QUEUE_TOOL", "/usr/bin/bongo-queuetool" +) +ADMIN = os.environ.get("BONGO_TEST_ADMIN", "/usr/bin/bongo-admin") +SYSTEMCTL = os.environ.get("BONGO_TEST_SYSTEMCTL", "/usr/bin/systemctl") +CYRUS_HOST = os.environ.get("BONGO_TEST_CYRUS_HOST", "localhost") +CYRUS_IMAP_PORT = int(os.environ.get("BONGO_TEST_CYRUS_IMAP_PORT", "18143")) +CYRUS_POP3_PORT = int(os.environ.get("BONGO_TEST_CYRUS_POP3_PORT", "18110")) +CYRUS_PASSWORD = os.environ.get( + "BONGO_TEST_CYRUS_PASSWORD", "BongoCollector-Test-2026" +) +CYRUS_CA = os.environ.get( + "BONGO_TEST_CYRUS_CA", "/tmp/bongo-cyrus-fixture/tls/ca.crt" +) +CYRUS_FIXTURE = Path(os.environ.get( + "BONGO_TEST_CYRUS_FIXTURE", + str(Path(__file__).with_name("cyrus-fixture.py")), +)) +FLOOD_COUNT = int(os.environ.get("BONGO_TEST_FLOOD_COUNT", "8")) +FLOOD_SIZE = int(os.environ.get("BONGO_TEST_FLOOD_SIZE", str(128 * 1024))) +QUOTA_LIMIT = int(os.environ.get("BONGO_TEST_FLOOD_QUOTA", "400000")) +TIMEOUT = int(os.environ.get("BONGO_TEST_TIMEOUT", "120")) +ALLOW_LIVE = os.environ.get("BONGO_ALLOW_LIVE_USER_TEST") == "1" +TRACE = os.environ.get("BONGO_TEST_TRACE") == "1" +CLEANUP_ONLY = os.environ.get("BONGO_TEST_CLEANUP_ONLY") == "1" + +FLOOD_SOURCES = ( + ("test1-pop", "pop3", CYRUS_POP3_PORT, "INBOX"), + ("test1-imap", "imap", CYRUS_IMAP_PORT, "Cyrus IMAP"), +) +HOLD_SOURCE = ("test2-pop", "pop3", CYRUS_POP3_PORT, "INBOX") +HOLD_MARKER = b"normal-test2-pop-1" +TEST_MARKER = re.compile( + rb"(?:flood-test1-(?:pop|imap)-[0-9]+|normal-test2-pop-1)" +) + + +class FloodCheckError(RuntimeError): + """Raised when STQ-12 loses, duplicates, or over-delivers mail.""" + + +def trace(message: str) -> None: + if TRACE: + print(f"STQ-12 trace: {message}", file=sys.stderr, flush=True) + + +def run(arguments: list[str], *, check: bool = True, + binary: bool = False, input_data=None) -> subprocess.CompletedProcess: + completed = subprocess.run( + arguments, + input=input_data, + capture_output=True, + text=not binary, + check=False, + ) + if check and completed.returncode: + stdout = completed.stdout + stderr = completed.stderr + if binary: + stdout = stdout.decode("utf-8", "replace") + stderr = stderr.decode("utf-8", "replace") + raise FloodCheckError( + f"{' '.join(arguments)} failed: " + f"{stderr.strip() or stdout.strip()}" + ) + return completed + + +def run_admin(*arguments: str, input_text: str | None = None) -> str: + return run( + ["sudo", "-n", ADMIN, *arguments], + input_data=input_text, + ).stdout + + +def run_queue(*arguments: str, check: bool = True, + binary: bool = False) -> subprocess.CompletedProcess: + return run( + ["sudo", "-n", "-u", "bongo", QUEUE_TOOL, *arguments], + check=check, + binary=binary, + ) + + +def restart_bongo() -> None: + run(["sudo", "-n", SYSTEMCTL, "daemon-reload"]) + run(["sudo", "-n", SYSTEMCTL, "restart", "bongo.service"]) + wait_port(STORE_HOST, STORE_PORT) + wait_port("127.0.0.1", 8670) + deadline = time.monotonic() + 30 + while time.monotonic() < deadline: + try: + with urllib.request.urlopen(WEB_URL + "/health", timeout=2) as response: + if response.status == 200: + return + except (OSError, urllib.error.URLError): + pass + time.sleep(0.25) + raise FloodCheckError("Bongo Web did not become ready after restart") + + +def wait_port(host: str, port: int) -> None: + deadline = time.monotonic() + 30 + while time.monotonic() < deadline: + try: + with socket.create_connection((host, port), timeout=0.5): + return + except OSError: + time.sleep(0.1) + raise FloodCheckError(f"{host}:{port} did not become ready") + + +class WebApi: + def __init__(self) -> None: + cookies = http.cookiejar.CookieJar() + self.opener = urllib.request.build_opener( + urllib.request.HTTPCookieProcessor(cookies) + ) + response = self.request( + "POST", "/api/session", + {"user": USER, "password": PASSWORD}, + csrf=False, + ) + self.csrf = str(response["csrf_token"]) + + def request(self, method: str, path: str, payload=None, + *, csrf: bool = True): + headers = {} + data = None + if payload is not None: + data = json.dumps(payload).encode("utf-8") + headers["Content-Type"] = "application/json" + if csrf and hasattr(self, "csrf"): + headers["X-Bongo-CSRF"] = self.csrf + request = urllib.request.Request( + WEB_URL + path, data=data, headers=headers, method=method + ) + try: + with self.opener.open(request, timeout=10) as response: + body = response.read() + return json.loads(body) if body else None + except urllib.error.HTTPError as error: + detail = error.read().decode("utf-8", "replace") + raise FloodCheckError( + f"{method} {path} returned HTTP {error.code}: {detail}" + ) from error + + def accounts(self) -> list[dict]: + return self.request("GET", "/api/external-accounts")["accounts"] + + def create_account(self, source: tuple[str, str, int, str]) -> int: + username, protocol, port, destination = source + result = self.request( + "POST", + "/api/external-accounts", + { + "provider": "manual", + "label": f"STQ-12 {protocol.upper()} {username}", + "protocol": protocol, + "email_address": f"{username}@cyrus.bongo.test", + "username": username, + "secret": CYRUS_PASSWORD, + "inbound_host": "127.0.0.1", + "inbound_port": port, + "inbound_tls": "starttls", + "inbound_mailbox": "INBOX", + "destination_folder": destination, + "poll_interval": 60, + "delete_policy": "after_import", + }, + ) + return int(result["id"]) + + def delete_account(self, account_id: int) -> None: + self.request( + "DELETE", + "/api/external-accounts?" + + urllib.parse.urlencode({"id": account_id}), + ) + + +def open_store() -> StoreClient: + return StoreClient( + USER, USER, authPassword=PASSWORD, + host=STORE_HOST, port=STORE_PORT, + ) + + +def read_quota() -> tuple[int, int]: + output = run_admin("user", "quota", USER) + used = re.search(r"^Used: ([0-9]+) bytes$", output, re.MULTILINE) + limit = re.search( + r"^Quota: (?:(unlimited)|([0-9]+) bytes)$", + output, + re.MULTILINE, + ) + if not used or not limit: + raise FloodCheckError(f"cannot parse quota output: {output!r}") + return int(used.group(1)), 0 if limit.group(1) else int(limit.group(2)) + + +def set_quota(limit: int) -> None: + run_admin( + "user", "quota", USER, + "unlimited" if limit == 0 else str(limit), + ) + + +def seed(profile: str, count: int, username: str) -> None: + run([ + sys.executable, + str(CYRUS_FIXTURE), + "seed", + "--profile", profile, + "--count", str(count), + "--message-size", str(FLOOD_SIZE), + "--user", username, + ]) + + +def cyrus_client(username: str) -> imaplib.IMAP4: + context = ssl.create_default_context(cafile=CYRUS_CA) + client = imaplib.IMAP4(CYRUS_HOST, CYRUS_IMAP_PORT) + client.starttls(context) + client.login(username, CYRUS_PASSWORD) + return client + + +def remote_count(username: str) -> int: + client = cyrus_client(username) + try: + status, values = client.status("INBOX", "(MESSAGES)") + if status != "OK" or not values or values[0] is None: + raise FloodCheckError(f"cannot inspect Cyrus mailbox {username}") + return int(values[0].decode("ascii").rsplit(" ", 1)[1].rstrip(")")) + finally: + client.logout() + + +def clear_remote(username: str) -> None: + client = cyrus_client(username) + try: + status, _ = client.select("INBOX") + if status != "OK": + raise FloodCheckError(f"cannot select Cyrus mailbox {username}") + status, values = client.search(None, "ALL") + if status == "OK" and values and values[0]: + for identifier in values[0].split(): + client.store(identifier, "+FLAGS.SILENT", r"(\Deleted)") + client.expunge() + finally: + client.logout() + + +def queue_items() -> dict[str, bytes]: + result: dict[str, bytes] = {} + for line in run_queue("list").stdout.splitlines(): + fields = line.split() + if not fields or not fields[0].startswith(("006-", "007-")): + continue + queue_id = fields[0] + message = run_queue( + "message", queue_id, check=False, binary=True + ) + if message.returncode == 0: + result[queue_id] = message.stdout + return result + + +def store_items(store: StoreClient) -> dict[str, bytes]: + result: dict[str, bytes] = {} + for folder in ("INBOX", "Cyrus IMAP"): + try: + documents = list(store.List(f"/mail/{folder}")) + except Exception: + continue + for document in documents: + try: + result[document.uid] = store.Read(document.uid) + except Exception: + continue + return result + + +def marker_counts(store: StoreClient) -> tuple[dict[bytes, list[str]], + set[str], set[str]]: + matches: dict[bytes, list[str]] = {} + store_ids: set[str] = set() + queue_ids: set[str] = set() + for uid, payload in store_items(store).items(): + markers = set(TEST_MARKER.findall(payload)) + for marker in markers: + matches.setdefault(marker, []).append(f"store:{uid}") + store_ids.add(uid) + for queue_id, payload in queue_items().items(): + markers = set(TEST_MARKER.findall(payload)) + for marker in markers: + matches.setdefault(marker, []).append(f"queue:{queue_id}") + queue_ids.add(queue_id) + return matches, store_ids, queue_ids + + +def expected_markers() -> set[bytes]: + return { + f"flood-{username}-{number}".encode("ascii") + for username, _protocol, _port, _folder in FLOOD_SOURCES + for number in range(1, FLOOD_COUNT + 1) + } + + +def wait_flood(store: StoreClient, api: WebApi) -> tuple[set[str], set[str]]: + expected = expected_markers() + deadline = time.monotonic() + TIMEOUT + last_detail = "" + while time.monotonic() < deadline: + matches, store_ids, queue_ids = marker_counts(store) + used, limit = read_quota() + if limit != QUOTA_LIMIT or used > limit: + raise FloodCheckError( + f"quota boundary violated: used={used}, limit={limit}" + ) + remote = { + username: remote_count(username) + for username, _protocol, _port, _folder in FLOOD_SOURCES + } + duplicates = { + marker: locations for marker, locations in matches.items() + if len(locations) != 1 + } + checked = [ + account for account in api.accounts() + if str(account.get("label", "")).startswith("STQ-12 ") + ] + if ( + set(matches) == expected + and not duplicates + and all(value == 0 for value in remote.values()) + and len(checked) == len(FLOOD_SOURCES) + and all(int(item.get("last_success_at", 0)) > 0 for item in checked) + and store_ids + and queue_ids + ): + return store_ids, queue_ids + last_detail = ( + f"markers={len(matches)}/{len(expected)} " + f"store={len(store_ids)} queue={len(queue_ids)} " + f"remote={remote} duplicates={len(duplicates)} " + f"used={used}/{limit}" + ) + trace(last_detail) + time.sleep(1) + raise FloodCheckError(f"collector flood did not settle: {last_detail}") + + +def delete_test_content(store: StoreClient) -> None: + # Stop every durable retry before removing Store documents. Doing this in + # the opposite order permits a live Queue worker to redeliver between the + # two cleanup operations. + for _attempt in range(3): + _matches, store_ids, queue_ids = marker_counts(store) + for queue_id in queue_ids: + run_queue("delete", queue_id, check=False) + if queue_ids: + time.sleep(0.25) + for uid in store_ids: + try: + store.Delete(uid) + except Exception: + pass + if not marker_counts(store)[0]: + return + time.sleep(0.25) + + +def cleanup_only() -> int: + _used, original_quota = read_quota() + api = WebApi() + for account in api.accounts(): + if str(account.get("label", "")).startswith("STQ-12 "): + api.delete_account(int(account["id"])) + store = open_store() + try: + # Prevent a final delivery worker already in flight from growing the + # mailbox while its retained Queue entry is being removed. + set_quota(max(read_quota()[0], 1)) + delete_test_content(store) + finally: + store.Quit() + set_quota(original_quota) + for source in (*FLOOD_SOURCES, HOLD_SOURCE): + clear_remote(source[0]) + print("STQ-12 cleanup complete") + return 0 + + +def replace_configuration(name: str, configuration: dict) -> None: + run_admin( + "__config-replace", name, + input_text=json.dumps(configuration, separators=(",", ":")), + ) + + +def wait_local_configuration(name: str, key: str, expected) -> None: + path = Path("/etc/bongo/config.d") / name + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + try: + current = json.loads(path.read_text(encoding="utf-8")) + if current.get(key) == expected: + return + except (OSError, ValueError): + pass + time.sleep(0.1) + raise FloodCheckError( + f"{path} did not mirror {key}={expected!r} from the Store" + ) + + +def wait_for_failed_account(api: WebApi, account_id: int) -> str: + # A newly created account can arrive just after Collector entered its + # bounded 60-second sleep. Keep the live test deterministic without + # reaching into the process to wake that worker. + deadline = time.monotonic() + 75 + while time.monotonic() < deadline: + account = next( + (item for item in api.accounts() + if int(item["id"]) == account_id), + None, + ) + if account is not None and int(account.get("last_checked_at", 0)) > 0: + error = str(account.get("last_error") or "") + if error: + return error + time.sleep(0.5) + raise FloodCheckError("Collector did not record the forced Queue rejection") + + +def require_safe_environment() -> None: + if not ALLOW_LIVE: + raise FloodCheckError( + "set BONGO_ALLOW_LIVE_USER_TEST=1 for the disposable live account" + ) + if not PASSWORD: + raise FloodCheckError("BONGO_TEST_PASSWORD must be set") + if not 1 <= FLOOD_COUNT <= 50: + raise FloodCheckError("BONGO_TEST_FLOOD_COUNT must be between 1 and 50") + if not 4096 <= FLOOD_SIZE <= 4 * 1024 * 1024: + raise FloodCheckError( + "BONGO_TEST_FLOOD_SIZE must be between 4 KiB and 4 MiB" + ) + + +def main() -> int: + require_safe_environment() + if CLEANUP_ONLY: + return cleanup_only() + baseline_used, original_quota = read_quota() + if baseline_used != 0: + raise FloodCheckError( + f"{USER} must be empty before STQ-12 (used={baseline_used})" + ) + original_collector = json.loads( + run_admin("__config-read", "collector") + ) + account_ids: list[int] = [] + store: StoreClient | None = None + collector_changed = False + + for source in (*FLOOD_SOURCES, HOLD_SOURCE): + clear_remote(source[0]) + + try: + api = WebApi() + existing = [ + item for item in api.accounts() + if str(item.get("label", "")).startswith("STQ-12 ") + ] + for item in existing: + api.delete_account(int(item["id"])) + + set_quota(QUOTA_LIMIT) + for source in FLOOD_SOURCES: + seed("flood", FLOOD_COUNT, source[0]) + account_ids.append(api.create_account(source)) + restart_bongo() + api = WebApi() + store = open_store() + flood_store_ids, flood_queue_ids = wait_flood(store, api) + used_at_limit, _ = read_quota() + + for account_id in list(account_ids): + api.delete_account(account_id) + account_ids.remove(account_id) + delete_test_content(store) + set_quota(0) + + blocked_collector = dict(original_collector) + blocked_collector["maximum_pending_messages"] = max( + 2, int(blocked_collector.get("maximum_pending_messages", 10000)) + ) + blocked_collector["maximum_pending_messages_per_user"] = 1 + replace_configuration("collector", blocked_collector) + wait_local_configuration( + "collector", "maximum_pending_messages_per_user", 1 + ) + collector_changed = True + restart_bongo() + api = WebApi() + seed("normal", 2, HOLD_SOURCE[0]) + hold_id = api.create_account(HOLD_SOURCE) + account_ids.append(hold_id) + error = wait_for_failed_account(api, hold_id) + normalized_error = error.casefold() + if "pending-message limit" not in normalized_error: + raise FloodCheckError( + f"Collector limit reported an unexpected error: {error}" + ) + if remote_count(HOLD_SOURCE[0]) != 1: + raise FloodCheckError( + "Collector did not leave the first over-limit message remote" + ) + hold_locations, _store_ids, _queue_ids = marker_counts(store) + if len(hold_locations.get(HOLD_MARKER, ())) != 1: + raise FloodCheckError( + "Collector did not durably retain exactly one pre-limit message" + ) + + api.delete_account(hold_id) + account_ids.remove(hold_id) + replace_configuration("collector", original_collector) + wait_local_configuration( + "collector", "maximum_pending_messages_per_user", + original_collector.get("maximum_pending_messages_per_user"), + ) + collector_changed = False + restart_bongo() + clear_remote(HOLD_SOURCE[0]) + finally: + if collector_changed: + try: + replace_configuration("collector", original_collector) + wait_local_configuration( + "collector", "maximum_pending_messages_per_user", + original_collector.get( + "maximum_pending_messages_per_user" + ), + ) + restart_bongo() + except Exception as cleanup_error: + print( + f"STQ-12 cleanup warning (Collector): {cleanup_error}", + file=sys.stderr, + ) + try: + cleanup_api = WebApi() + for account_id in account_ids: + try: + cleanup_api.delete_account(account_id) + except Exception: + pass + except Exception: + pass + if store is not None: + try: + delete_test_content(store) + except Exception: + pass + try: + store.Quit() + except Exception: + pass + for source in (*FLOOD_SOURCES, HOLD_SOURCE): + try: + clear_remote(source[0]) + except Exception: + pass + try: + set_quota(original_quota) + except Exception as cleanup_error: + print( + f"STQ-12 cleanup warning (quota): {cleanup_error}", + file=sys.stderr, + ) + + final_used, final_limit = read_quota() + if (final_used, final_limit) != (baseline_used, original_quota): + raise FloodCheckError( + "final cleanup did not restore the original quota state" + ) + print( + "STQ-12 PASS " + f"pop3={FLOOD_COUNT} imap={FLOOD_COUNT} size={FLOOD_SIZE} " + f"quota-used={used_at_limit}/{QUOTA_LIMIT} " + f"store={len(flood_store_ids)} queue={len(flood_queue_ids)} " + "duplicates=no overflow=no source-retained-at-limit=yes " + "quota-restored=yes" + ) + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except (FloodCheckError, OSError, ValueError) as error: + print(f"STQ-12 FAIL: {error}", file=sys.stderr) + raise SystemExit(1) diff --git a/contrib/testing/cyrus-fixture.py b/contrib/testing/cyrus-fixture.py index b327080..f5d655b 100755 --- a/contrib/testing/cyrus-fixture.py +++ b/contrib/testing/cyrus-fixture.py @@ -376,17 +376,28 @@ def deliver_message(message: EmailMessage) -> None: raise RuntimeError(f"LMTP refused recipient: {refused}") -def seed_profile(profile: str, count: int, message_size: int) -> None: +def seed_profile(profile: str, count: int, message_size: int, + users: tuple[str, ...] = SOURCE_USERS) -> None: maximum = 256 if profile == "flood" else 100 if count < 1 or count > maximum: raise ValueError(f"{profile} count must be between 1 and {maximum}") if profile == "flood" and not 4096 <= message_size <= 4 * 1024 * 1024: raise ValueError("flood message size must be between 4 KiB and 4 MiB") - for username in SOURCE_USERS: + invalid_users = sorted(set(users) - set(SOURCE_USERS)) + if invalid_users: + raise ValueError( + "unknown Cyrus source user(s): " + ", ".join(invalid_users) + ) + if not users: + raise ValueError("at least one Cyrus source user is required") + for username in users: for number in range(1, count + 1): deliver_message(fixture_message( username, profile, number, message_size)) - print(f"seeded {count} {profile} message(s) in each Cyrus source mailbox") + print( + f"seeded {count} {profile} message(s) in " + f"{', '.join(users)}" + ) def seed_suite(normal_count: int, flood_count: int, flood_size: int) -> None: @@ -485,6 +496,9 @@ def parser() -> argparse.ArgumentParser: seeding.add_argument( "--message-size", type=int, default=128 * 1024, help="approximate flood body size in bytes (default: 131072)") + seeding.add_argument( + "--user", action="append", choices=SOURCE_USERS, + help="seed only this source user (repeat for multiple users)") suite = commands.add_parser("seed-suite") suite.add_argument("--normal-count", type=int, default=3) suite.add_argument("--flood-count", type=int, default=32) @@ -510,7 +524,12 @@ def main() -> int: elif args.command == "reset": reset(args.root) elif args.command == "seed": - seed_profile(args.profile, args.count, args.message_size) + seed_profile( + args.profile, + args.count, + args.message_size, + tuple(args.user) if args.user else SOURCE_USERS, + ) elif args.command == "seed-suite": seed_suite(args.normal_count, args.flood_count, args.flood_size) elif args.command == "verify-collector": diff --git a/contrib/testing/tests/cyrus-fixture-message-test.py b/contrib/testing/tests/cyrus-fixture-message-test.py index 5d19e41..3cdd713 100644 --- a/contrib/testing/tests/cyrus-fixture-message-test.py +++ b/contrib/testing/tests/cyrus-fixture-message-test.py @@ -79,6 +79,12 @@ class CyrusFixtureMessageTest(unittest.TestCase): with self.assertRaises(ValueError): MODULE.seed_profile("flood", 1, 4 * 1024 * 1024 + 1) + def test_targeted_seed_rejects_empty_or_unknown_source_sets(self): + with self.assertRaises(ValueError): + MODULE.seed_profile("normal", 1, 4096, ()) + with self.assertRaises(ValueError): + MODULE.seed_profile("normal", 1, 4096, ("not-a-source",)) + if __name__ == "__main__": unittest.main() diff --git a/docs/external-accounts.md b/docs/external-accounts.md index 0d771fa..8ac80fe 100644 --- a/docs/external-accounts.md +++ b/docs/external-accounts.md @@ -28,3 +28,23 @@ commits the complete message. The normal antivirus, antispam, Sieve/rules, quota, and Store delivery pipeline owns it from that point onward. A failed remote deletion does not cause the message to be delivered again; the collector retries the deletion during a later poll. + +`maximum_pending_messages` and +`maximum_pending_messages_per_user` in the suffixless `collector` +configuration bound the number of externally collected messages that may be +waiting anywhere in Bongo's Queue. Their defaults are 10000 globally and 1000 +per local user. The per-user value must not exceed the global value. The +collector checks both limits before fetching each previously unseen message; +when either limit is reached, it records backpressure on the external account +and leaves that message at the provider. Successfully delivered Queue entries +no longer count toward either limit, so collection resumes on a later poll +after the user or administrator has freed the bottleneck. + +The authenticated Queue command `QCOUNT COLLECTED [user]` supplies this +pending count without treating unrelated SMTP or LMTP traffic as collector +load. Administrators can inspect the same values with: + +```sh +bongo-queuetool collected-count +bongo-queuetool collected-count username +``` diff --git a/docs/queue.md b/docs/queue.md index ce8c42a..e55e768 100644 --- a/docs/queue.md +++ b/docs/queue.md @@ -32,6 +32,19 @@ consuming the filesystem's final space. Per-user Store quota is separate: a full mailbox defers or rejects that recipient without corrupting other recipient state. +Collector backpressure is also separate from the byte reserve. The +`collector` document limits globally pending and per-user pending external +messages. Queue counts only entries carrying its durable collected-message +flag, including entries currently owned by a Queue worker. Once a limit is +reached, the collector does not download or delete the next source message. +`bongo-queuetool collected-count [user]` exposes the global count and, +optionally, the selected user's count. + +JSON integral configuration values and the generic agent loader preserve +signed 64-bit values. In particular, a large `minimumfreespace` value is not +truncated into a negative 32-bit number while Store configuration is mirrored +to disk. + A scanner outage is a delivery-policy decision, not permission to silently bypass inspection. ClamAV and SpamAssassin failures leave the message queued for the configured retry path. Likewise, a per-recipient LMTP temporary error diff --git a/include/bongoagent.h b/include/bongoagent.h index 3bf8e83..8d23a5a 100644 --- a/include/bongoagent.h +++ b/include/bongoagent.h @@ -24,6 +24,8 @@ #ifndef BONGOAGENT_H #define BONGOAGENT_H +#include + #include #include #include @@ -115,7 +117,7 @@ typedef struct _BongoConfigIntegerDestination { &(BongoConfigIntegerDestination){ (destination), sizeof(*(destination)) } } BOOL SetBongoConfigInteger(BongoConfigIntegerDestination *destination, - int value); + int64_t value); BOOL ReadBongoConfiguration(BongoConfigItem *config, char *filename); BOOL ProcessBongoConfiguration(BongoConfigItem *config, BongoJsonNode *node); diff --git a/include/bongojson.h b/include/bongojson.h index f9e49b9..08a1211 100644 --- a/include/bongojson.h +++ b/include/bongojson.h @@ -22,6 +22,9 @@ #ifndef BONGOJSON_H #define BONGOJSON_H +#include +#include + #include #include #include @@ -68,7 +71,7 @@ struct _BongoJsonNode { GArray *arrayVal; BOOL boolVal; double doubleVal; - int intVal; + int64_t intVal; char *stringVal; } value; }; @@ -83,7 +86,7 @@ BongoJsonNode *BongoJsonNodeNewObject(BongoJsonObject *val); BongoJsonNode *BongoJsonNodeNewArray(GArray *val); BongoJsonNode *BongoJsonNodeNewBool(BOOL val); BongoJsonNode *BongoJsonNodeNewDouble(double val); -BongoJsonNode *BongoJsonNodeNewInt(int val); +BongoJsonNode *BongoJsonNodeNewInt(int64_t val); BongoJsonNode *BongoJsonNodeNewString(const char *val); BongoJsonNode *BongoJsonNodeNewStringGive(char *val); @@ -142,11 +145,23 @@ DEF_ARRAY_GET(BongoJsonObject *, BONGO_JSON_OBJECT, Object); DEF_ARRAY_GET(GArray *, BONGO_JSON_ARRAY, Array); DEF_ARRAY_GET(BOOL, BONGO_JSON_BOOL, Bool); DEF_ARRAY_GET(double, BONGO_JSON_DOUBLE, Double); -DEF_ARRAY_GET(int, BONGO_JSON_INT, Int); DEF_ARRAY_GET(const char *, BONGO_JSON_STRING, String); #undef DEF_ARRAY_GET +static __inline BongoJsonResult +BongoJsonArrayGetInt(GArray *array, int i, int *val) +{ + BongoJsonNode *node = BongoJsonArrayGet(array, i); + if (node->type != BONGO_JSON_INT || + BongoJsonNodeAsInt(node) < INT_MIN || + BongoJsonNodeAsInt(node) > INT_MAX) { + return BONGO_JSON_BAD_TYPE; + } + *val = (int)BongoJsonNodeAsInt(node); + return BONGO_JSON_OK; +} + GArray *BongoJsonArrayDup(GArray *array); void BongoJsonArrayFree(GArray *array); @@ -179,6 +194,7 @@ BongoJsonResult BongoJsonObjectGetArray(BongoJsonObject *obj, const char *key, G BongoJsonResult BongoJsonObjectGetBool(BongoJsonObject *obj, const char *key, BOOL *val); BongoJsonResult BongoJsonObjectGetDouble(BongoJsonObject *obj, const char *key, double *val); BongoJsonResult BongoJsonObjectGetInt(BongoJsonObject *obj, const char *key, int *val); +BongoJsonResult BongoJsonObjectGetInt64(BongoJsonObject *obj, const char *key, int64_t *val); BongoJsonResult BongoJsonObjectGetString(BongoJsonObject *obj, const char *key, const char **val); BongoJsonResult BongoJsonObjectPut(BongoJsonObject *obj, const char *key, BongoJsonNode *node); @@ -223,6 +239,7 @@ BongoJsonResult BongoJsonJPathGetObject(BongoJsonNode *n, const char *path, Bong BongoJsonResult BongoJsonJPathGetArray(BongoJsonNode *n, const char *path, GArray **val); BongoJsonResult BongoJsonJPathGetBool(BongoJsonNode *n, const char *path, BOOL *val); BongoJsonResult BongoJsonJPathGetInt(BongoJsonNode *n, const char *path, int *val); +BongoJsonResult BongoJsonJPathGetInt64(BongoJsonNode *n, const char *path, int64_t *val); BongoJsonResult BongoJsonJPathGetString(BongoJsonNode *n, const char *path, char **val); BongoJsonResult BongoJsonJPathGetDouble(BongoJsonNode *n, const char *path, double *val); BongoJsonResult BongoJsonJPathGetFloat(BongoJsonNode *n, const char *path, float *val); diff --git a/man/bongo-queuetool.1 b/man/bongo-queuetool.1 index 55baa8c..470bea3 100644 --- a/man/bongo-queuetool.1 +++ b/man/bongo-queuetool.1 @@ -36,6 +36,13 @@ free-space reserve. The optional form also brackets the Queue query with backing-filesystem samples and prints the reserve and allocation-unit size for diagnostics. .TP +.BR collected-count " [" USER "]" +Print the number of pending messages accepted by the external collector. +With +.IR USER , +also print that local user's pending count. These values enforce the global +and per-user collector backpressure limits. +.TP .BR show " (s) " QUEUE-ID Write the routing envelope of one queue entry to standard output. .TP diff --git a/src/agents/collector/CMakeLists.txt b/src/agents/collector/CMakeLists.txt index dffa14d..bf647cf 100644 --- a/src/agents/collector/CMakeLists.txt +++ b/src/agents/collector/CMakeLists.txt @@ -4,6 +4,7 @@ add_executable(bongocollector collector.c external_store.c external_transport.c + pending-count.c ) target_link_libraries(bongocollector @@ -31,6 +32,15 @@ if(BUILD_TESTING) target_link_libraries(collector-transport-test CURL::libcurl) add_test(NAME collector-external-transport COMMAND collector-transport-test) + add_executable(collector-pending-count-test + tests/pending-count-test.c + pending-count.c + ) + target_include_directories(collector-pending-count-test PRIVATE + ${CMAKE_CURRENT_SOURCE_DIR} + ) + add_test(NAME collector-pending-count COMMAND collector-pending-count-test) + add_executable(collector-remote-transport-probe tests/vtestmail-transport-test.c external_transport.c) diff --git a/src/agents/collector/collector.c b/src/agents/collector/collector.c index b80487b..6c0670d 100644 --- a/src/agents/collector/collector.c +++ b/src/agents/collector/collector.c @@ -53,6 +53,9 @@ static int CollectExternalAccount(sqlite3 *database, CollectorRemoteIds ids = {0}; size_t index; size_t processed = 0; + uint64_t pending = 0; + uint64_t pending_for_owner = 0; + int have_pending_counts = 0; int secretResult; int result = -1; @@ -115,6 +118,30 @@ static int CollectExternalAccount(sqlite3 *database, } continue; } + if (!have_pending_counts) { + if (CollectorStorePendingCounts(account->owner, &pending, + &pending_for_owner, error, + sizeof(error)) != 0) { + goto failed; + } + have_pending_counts = 1; + } + if (pending >= Collector.maximumPendingMessages) { + snprintf(error, sizeof(error), + "collector global pending-message limit %u reached; " + "remote message was left at the provider", + Collector.maximumPendingMessages); + goto failed; + } + if (pending_for_owner >= + Collector.maximumPendingMessagesPerUser) { + snprintf(error, sizeof(error), + "collector pending-message limit %u reached for user %s; " + "remote message was left at the provider", + Collector.maximumPendingMessagesPerUser, + account->owner); + goto failed; + } message = tmpfile(); if (message == NULL) { snprintf(error, sizeof(error), "cannot create temporary external message file"); @@ -129,6 +156,8 @@ static int CollectExternalAccount(sqlite3 *database, goto failed; } fclose(message); + pending++; + pending_for_owner++; if (strcmp(account->delete_policy, "after_import") == 0) { delete_after = now; } else if (strcmp(account->delete_policy, "after_days") == 0 && @@ -227,6 +256,10 @@ ReadConfiguration(void) { char *configuration = NULL; const char *ca_file = ""; + int maximum_pending = + (int)COLLECTOR_DEFAULT_MAXIMUM_PENDING_MESSAGES; + int maximum_pending_per_user = + (int)COLLECTOR_DEFAULT_MAXIMUM_PENDING_MESSAGES_PER_USER; BongoJsonNode *node = NULL; BongoJsonObject *object; BOOL result = FALSE; @@ -248,11 +281,31 @@ ReadConfiguration(void) if (BongoJsonObjectGetString(object, "ca_file", &ca_file) != BONGO_JSON_OK) { ca_file = ""; } + if (BongoJsonObjectGetInt(object, "maximum_pending_messages", + &maximum_pending) != BONGO_JSON_OK) { + maximum_pending = + (int)COLLECTOR_DEFAULT_MAXIMUM_PENDING_MESSAGES; + } + if (BongoJsonObjectGetInt(object, "maximum_pending_messages_per_user", + &maximum_pending_per_user) != BONGO_JSON_OK) { + maximum_pending_per_user = + (int)COLLECTOR_DEFAULT_MAXIMUM_PENDING_MESSAGES_PER_USER; + } + if (maximum_pending < 1 || maximum_pending_per_user < 1 || + maximum_pending_per_user > maximum_pending) { + Log(LOG_ERROR, + "collector pending-message limits must be positive and the " + "per-user limit must not exceed the global limit"); + goto done; + } if (CollectorRemoteSetCaFile(ca_file) != 0) { Log(LOG_ERROR, "collector.ca_file must be an absolute path shorter than 4096 bytes"); goto done; } + Collector.maximumPendingMessages = (unsigned int)maximum_pending; + Collector.maximumPendingMessagesPerUser = + (unsigned int)maximum_pending_per_user; result = TRUE; done: diff --git a/src/agents/collector/collector.h b/src/agents/collector/collector.h index 5117023..45cdeaa 100644 --- a/src/agents/collector/collector.h +++ b/src/agents/collector/collector.h @@ -32,6 +32,8 @@ #define AGENT_NAME "bongocollector" #define AGENT_DN "Collector Agent" +#define COLLECTOR_DEFAULT_MAXIMUM_PENDING_MESSAGES 10000U +#define COLLECTOR_DEFAULT_MAXIMUM_PENDING_MESSAGES_PER_USER 1000U typedef struct { Connection *conn; @@ -47,6 +49,9 @@ typedef struct _CollectorGlobals { Connection *clientListener; void *clientMemPool; BongoThreadPool *clientThreadPool; + + unsigned int maximumPendingMessages; + unsigned int maximumPendingMessagesPerUser; } CollectorGlobals; extern CollectorGlobals Collector; diff --git a/src/agents/collector/external_store.c b/src/agents/collector/external_store.c index 7a073fc..c642615 100644 --- a/src/agents/collector/external_store.c +++ b/src/agents/collector/external_store.c @@ -65,6 +65,44 @@ SetError(char *error, size_t error_size, const char *message) } } +int +CollectorStorePendingCounts(const char *owner, uint64_t *total, + uint64_t *owner_total, char *error, + size_t error_size) +{ + Connection *connection = NULL; + char response[CONN_BUFSIZE + 1]; + int result = -1; + + if (!SafeNmapToken(owner) || total == NULL || owner_total == NULL) { + SetError(error, error_size, "invalid collector queue-count request"); + return -1; + } + connection = NMAPConnectQueue("127.0.0.1", NULL); + if (connection == NULL) { + SetError(error, error_size, "cannot connect to the Bongo queue"); + return -1; + } + if (!NMAPAuthenticateToQueue(connection, response, sizeof(response))) { + SetError(error, error_size, "cannot authenticate to the Bongo queue"); + goto done; + } + if (NMAPRunCommandF(connection, response, sizeof(response), + "QCOUNT COLLECTED %s\r\n", owner) != 1000 || + CollectorStoreParsePendingCounts(response, total, owner_total) != 0) { + SetError(error, error_size, + "cannot count pending externally collected messages"); + goto done; + } + SetError(error, error_size, ""); + result = 0; + +done: + NMAPQuit(connection); + ConnFree(connection); + return result; +} + int CollectorStoreSubmitFile(const char *owner, const char *folder, FILE *message, size_t message_size, diff --git a/src/agents/collector/external_store.h b/src/agents/collector/external_store.h index 00f733b..b69ee73 100644 --- a/src/agents/collector/external_store.h +++ b/src/agents/collector/external_store.h @@ -23,6 +23,7 @@ #define BONGO_COLLECTOR_EXTERNAL_STORE_H #include +#include #include #include @@ -30,5 +31,11 @@ int CollectorStoreSubmitFile(const char *owner, const char *folder, FILE *message, size_t message_size, time_t received_at, char *error, size_t error_size); +int CollectorStoreParsePendingCounts(const char *response, + uint64_t *total, + uint64_t *owner_total); +int CollectorStorePendingCounts(const char *owner, uint64_t *total, + uint64_t *owner_total, char *error, + size_t error_size); #endif diff --git a/src/agents/collector/pending-count.c b/src/agents/collector/pending-count.c new file mode 100644 index 0000000..3129052 --- /dev/null +++ b/src/agents/collector/pending-count.c @@ -0,0 +1,38 @@ +/**************************************************************************** + * + * Copyright (c) 2001 Novell, Inc. All Rights Reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of version 2 of the GNU General Public License + * as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, contact Novell, Inc. + * + ****************************************************************************/ + +#include + +#include +#include + +#include "external_store.h" + +int +CollectorStoreParsePendingCounts(const char *response, uint64_t *total, + uint64_t *owner_total) +{ + char description[2]; + + if (response == NULL || total == NULL || owner_total == NULL || + sscanf(response, "%" SCNu64 " %" SCNu64 " %1s", + total, owner_total, description) != 3) { + return -1; + } + return 0; +} diff --git a/src/agents/collector/tests/pending-count-test.c b/src/agents/collector/tests/pending-count-test.c new file mode 100644 index 0000000..322b186 --- /dev/null +++ b/src/agents/collector/tests/pending-count-test.c @@ -0,0 +1,38 @@ +/**************************************************************************** + * + * Copyright (c) 2001 Novell, Inc. All Rights Reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of version 2 of the GNU General Public License + * as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, contact Novell, Inc. + * + ****************************************************************************/ + +#include +#include + +#include "external_store.h" + +int +main(void) +{ + uint64_t total = 0; + uint64_t owner = 0; + + assert(CollectorStoreParsePendingCounts( + "12345 678 Collected queue counts", &total, &owner) == 0); + assert(total == UINT64_C(12345)); + assert(owner == UINT64_C(678)); + assert(CollectorStoreParsePendingCounts("1", &total, &owner) == -1); + assert(CollectorStoreParsePendingCounts("x 2 counts", &total, &owner) == -1); + assert(CollectorStoreParsePendingCounts(NULL, &total, &owner) == -1); + return 0; +} diff --git a/src/agents/queue/CMakeLists.txt b/src/agents/queue/CMakeLists.txt index b5bc4e8..7cc8964 100644 --- a/src/agents/queue/CMakeLists.txt +++ b/src/agents/queue/CMakeLists.txt @@ -1,4 +1,5 @@ add_executable(bongoqueue + collected-count.c conf.c delivery-policy.c disk-space.c @@ -41,6 +42,15 @@ if(BUILD_TESTING) ) add_test(NAME queue-delivery-policy COMMAND queue-delivery-policy-test) + add_executable(queue-collected-count-test + tests/collected-count-test.c + collected-count.c + ) + target_include_directories(queue-collected-count-test PRIVATE + ${CMAKE_CURRENT_SOURCE_DIR} + ) + add_test(NAME queue-collected-count COMMAND queue-collected-count-test) + add_executable(queue-disk-space-test tests/disk-space-test.c disk-space.c diff --git a/src/agents/queue/collected-count.c b/src/agents/queue/collected-count.c new file mode 100644 index 0000000..b09c023 --- /dev/null +++ b/src/agents/queue/collected-count.c @@ -0,0 +1,83 @@ +/**************************************************************************** + * + * Copyright (c) 2001 Novell, Inc. All Rights Reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of version 2 of the GNU General Public License + * as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, contact Novell, Inc. + * + ****************************************************************************/ + +#include + +#include +#include +#include +#include + +#include + +#include "collected-count.h" + +static BOOL +RecipientMatches(const char *line, const char *owner) +{ + const char *end; + size_t length; + + if (owner == NULL || owner[0] == '\0') + return FALSE; + end = line + 1; + while (*end != '\0' && *end != '\r' && *end != '\n' && + *end != ' ' && *end != '\t') { + end++; + } + length = (size_t)(end - (line + 1)); + return length == strlen(owner) && + memcmp(line + 1, owner, length) == 0; +} + +int +BongoQueueCollectedEnvelope(FILE *control, const char *owner, + BOOL *owner_matches) +{ + char line[4096]; + unsigned long flags = 0; + BOOL have_flags = FALSE; + BOOL matches = FALSE; + + if (control == NULL || owner_matches == NULL) + return -1; + rewind(control); + while (fgets(line, sizeof(line), control) != NULL) { + if (line[0] == QUEUE_FLAGS) { + char *end = NULL; + unsigned long parsed; + + errno = 0; + parsed = strtoul(line + 1, &end, 10); + if (errno != 0 || end == line + 1 || parsed > UINT_MAX || + (*end != '\0' && *end != '\r' && *end != '\n')) { + return -1; + } + flags = parsed; + have_flags = TRUE; + } else if ((line[0] == QUEUE_RECIP_LOCAL || + line[0] == QUEUE_RECIP_MBOX_LOCAL) && + RecipientMatches(line, owner)) { + matches = TRUE; + } + } + if (ferror(control)) + return -1; + *owner_matches = matches; + return have_flags && (flags & MSG_FLAG_COLLECTED_MESSAGE) != 0 ? 1 : 0; +} diff --git a/src/agents/queue/collected-count.h b/src/agents/queue/collected-count.h new file mode 100644 index 0000000..9052121 --- /dev/null +++ b/src/agents/queue/collected-count.h @@ -0,0 +1,29 @@ +/**************************************************************************** + * + * Copyright (c) 2001 Novell, Inc. All Rights Reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of version 2 of the GNU General Public License + * as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, contact Novell, Inc. + * + ****************************************************************************/ + +#ifndef BONGO_QUEUE_COLLECTED_COUNT_H +#define BONGO_QUEUE_COLLECTED_COUNT_H + +#include + +#include + +int BongoQueueCollectedEnvelope(FILE *control, const char *owner, + BOOL *owner_matches); + +#endif diff --git a/src/agents/queue/messages.h b/src/agents/queue/messages.h index 849cea1..19140f0 100644 --- a/src/agents/queue/messages.h +++ b/src/agents/queue/messages.h @@ -55,6 +55,10 @@ #define NMAP_QLIST_COMMAND "QLIST" #define NMAP_QLIST_HELP "QLIST - Lists committed message queue entries.\r\n" +#define NMAP_QCOUNT_COLLECTED_COMMAND "QCOUNT COLLECTED" +#define NMAP_QCOUNT_COLLECTED_HELP \ + "QCOUNT COLLECTED [user] - Counts pending externally collected messages.\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 20dea92..9b55769 100644 --- a/src/agents/queue/queue.c +++ b/src/agents/queue/queue.c @@ -27,6 +27,7 @@ #include #include "queue.h" +#include "collected-count.h" #include "disk-space.h" #include "mime.h" #include "messages.h" @@ -3472,6 +3473,93 @@ CommandQcrea(void *param) return(ConnWrite(client->conn, MSG1000ENTRYMADE, sizeof(MSG1000ENTRYMADE) - 1)); } +int +CommandQcountCollected(void *param) +{ + QueueClient *client = (QueueClient *)param; + const char *owner = NULL; + const char *arguments = + client->buffer + sizeof(NMAP_QCOUNT_COLLECTED_COMMAND) - 1; + XplDir *directory; + XplDir *entry; + GHashTable *counted; + uint64_t total = 0; + uint64_t owner_total = 0; + + if (*arguments == ' ' && arguments[1] != '\0') { + const unsigned char *cursor = + (const unsigned char *)(arguments + 1); + owner = arguments + 1; + while (*cursor != '\0') { + if (*cursor <= ' ' || *cursor == 0x7f) + return ConnWrite(client->conn, MSG3010BADARGC, + sizeof(MSG3010BADARGC) - 1); + cursor++; + } + } else if (*arguments != '\0') { + return ConnWrite(client->conn, MSG3010BADARGC, + sizeof(MSG3010BADARGC) - 1); + } + + counted = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, NULL); + directory = XplOpenDir(Conf.spoolPath); + if (directory == NULL) { + g_hash_table_destroy(counted); + return ConnWrite(client->conn, MSG4224CANTREAD, + sizeof(MSG4224CANTREAD) - 1); + } + while ((entry = XplReadDir(directory)) != NULL) { + const char *name = entry->d_nameDOS; + char id[8]; + char path[XPL_MAX_PATH + 1]; + FILE *control; + BOOL owner_matches = FALSE; + size_t position; + BOOL valid = TRUE; + int collected; + + if (strlen(name) != 12 || + (toupper((unsigned char)name[0]) != 'C' && + toupper((unsigned char)name[0]) != 'W') || + 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; + memcpy(id, name + 1, 7); + id[7] = '\0'; + if (g_hash_table_contains(counted, id)) + continue; + + QueueFormat(path, "%s/%s", Conf.spoolPath, name); + control = fopen(path, "rb"); + if (control == NULL) + continue; + collected = BongoQueueCollectedEnvelope(control, owner, &owner_matches); + fclose(control); + if (collected != 1) + continue; + g_hash_table_add(counted, g_strdup(id)); + total++; + if (owner_matches) + owner_total++; + } + XplCloseDir(directory); + g_hash_table_destroy(counted); + return ConnWriteF(client->conn, "1000 %" PRIu64 " %" PRIu64 + " Collected queue counts\r\n", total, owner_total); +} + int CommandQdele(void *param) { diff --git a/src/agents/queue/queue.h b/src/agents/queue/queue.h index 2dd4b40..9a019d0 100644 --- a/src/agents/queue/queue.h +++ b/src/agents/queue/queue.h @@ -114,6 +114,7 @@ int CommandQabrt(void *param); int CommandQbody(void *param); int CommandQbraw(void *param); int CommandQcopy(void *param); +int CommandQcountCollected(void *param); int CommandQcrea(void *param); int CommandQdele(void *param); int CommandQdone(void *param); diff --git a/src/agents/queue/queued.c b/src/agents/queue/queued.c index a92d04f..0a5c24b 100644 --- a/src/agents/queue/queued.c +++ b/src/agents/queue/queued.c @@ -67,6 +67,7 @@ static ProtocolCommand commands[] = { { NMAP_QBODY_COMMAND, NMAP_HELP_NOT_DEFINED, sizeof(NMAP_QBODY_COMMAND) - 1, CommandQbody, NULL, NULL, NULL, BlackCommand }, { NMAP_QBRAW_COMMAND, NMAP_HELP_NOT_DEFINED, sizeof(NMAP_QBRAW_COMMAND) - 1, CommandQbraw, NULL, NULL, NULL, BlackCommand }, { NMAP_QCOPY_COMMAND, NMAP_HELP_NOT_DEFINED, sizeof(NMAP_QCOPY_COMMAND) - 1, CommandQcopy, NULL, NULL, NULL, BlackCommand }, + { NMAP_QCOUNT_COLLECTED_COMMAND, NMAP_QCOUNT_COLLECTED_HELP, sizeof(NMAP_QCOUNT_COLLECTED_COMMAND) - 1, CommandQcountCollected, NULL, NULL, NULL, BlackCommand }, { NMAP_QCREA_COMMAND, NMAP_QCREA_HELP, sizeof(NMAP_QCREA_COMMAND) - 1, CommandQcrea, NULL, NULL, NULL, BlackCommand }, { NMAP_QDELE_COMMAND, NMAP_HELP_NOT_DEFINED, sizeof(NMAP_QDELE_COMMAND) - 1, CommandQdele, NULL, NULL, NULL, BlackCommand }, { NMAP_QDONE_COMMAND, NMAP_HELP_NOT_DEFINED, sizeof(NMAP_QDONE_COMMAND) - 1, CommandQdone, NULL, NULL, NULL, BlackCommand }, diff --git a/src/agents/queue/tests/collected-count-test.c b/src/agents/queue/tests/collected-count-test.c new file mode 100644 index 0000000..efa060c --- /dev/null +++ b/src/agents/queue/tests/collected-count-test.c @@ -0,0 +1,58 @@ +/**************************************************************************** + * + * Copyright (c) 2001 Novell, Inc. All Rights Reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of version 2 of the GNU General Public License + * as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, contact Novell, Inc. + * + ****************************************************************************/ + +#include +#include +#include + +#include + +#include "collected-count.h" + +static void +Check(const char *envelope, const char *owner, int expected_collected, + BOOL expected_owner) +{ + FILE *control = tmpfile(); + BOOL owner_matches = FALSE; + + assert(control != NULL); + assert(fwrite(envelope, 1, strlen(envelope), control) == strlen(envelope)); + assert(BongoQueueCollectedEnvelope(control, owner, &owner_matches) == + expected_collected); + assert(owner_matches == expected_owner); + fclose(control); +} + +int +main(void) +{ + char collected[256]; + + snprintf(collected, sizeof(collected), + "D1\r\nX%u\r\nMtest1 test1 32 INBOX 0\r\n", + (unsigned int)(MSG_FLAG_COLLECTED_MESSAGE | + MSG_FLAG_SOURCE_EXTERNAL)); + Check(collected, "test1", 1, TRUE); + Check(collected, "test", 1, FALSE); + Check("D1\r\nX0\r\nMtest1 test1 32 INBOX 0\r\n", "test1", 0, TRUE); + Check("D1\r\nX32768junk\r\nMtest1 test1 32 INBOX 0\r\n", "test1", -1, + FALSE); + Check("D1\r\nX32768\r\nLtest2 test2 0\r\n", "test2", 1, TRUE); + return 0; +} diff --git a/src/apps/config/config/collector b/src/apps/config/config/collector index ace9dd1..c2ebc11 100644 --- a/src/apps/config/config/collector +++ b/src/apps/config/config/collector @@ -1,4 +1,6 @@ { "version": 1, - "ca_file": "" + "ca_file": "", + "maximum_pending_messages": 10000, + "maximum_pending_messages_per_user": 1000 } diff --git a/src/apps/config/tests/test_configuration.py b/src/apps/config/tests/test_configuration.py index 42097a7..6bd973b 100644 --- a/src/apps/config/tests/test_configuration.py +++ b/src/apps/config/tests/test_configuration.py @@ -94,6 +94,23 @@ class ConfigurationModelTest(unittest.TestCase): errors(validate_all(configs, defaults, check_files=False)))) + def test_collector_pending_limits_are_positive_and_ordered(self): + defaults = templates() + configs = templates() + configs["collector"]["maximum_pending_messages"] = 10 + configs["collector"]["maximum_pending_messages_per_user"] = 11 + problems = errors(validate_all(configs, defaults, check_files=False)) + self.assertTrue(any( + issue.document == "collector" and + issue.path == "maximum_pending_messages_per_user" + for issue in problems)) + configs["collector"]["maximum_pending_messages_per_user"] = 5 + self.assertFalse(any( + issue.document == "collector" and + issue.path.startswith("maximum_pending_messages") + for issue in errors(validate_all( + configs, defaults, check_files=False)))) + def test_safe_reverse_proxy_scenario(self): defaults = templates() scenario = Scenario( diff --git a/src/apps/queuetool/bongo-queuetool b/src/apps/queuetool/bongo-queuetool index 0b0d16c..83af98a 100755 --- a/src/apps/queuetool/bongo-queuetool +++ b/src/apps/queuetool/bongo-queuetool @@ -94,6 +94,29 @@ class ShowCommand(Command): data = queue.RetrieveInfo(args[0]) sys.stdout.buffer.write(data) +class CollectedCountCommand(Command): + log = logging.getLogger("Bongo.QueueTool") + + def __init__(self): + Command.__init__( + self, "collected-count", + summary="Count pending externally collected messages", + usage="%prog %cmd []") + + def Run(self, options, args): + if len(args) > 1: + self.error("collected-count accepts at most one user") + queue = QueueClient(host=options.host, port=options.port) + try: + total, user_total = queue.CountCollected( + args[0] if args else None) + finally: + queue.Quit() + if args: + print(f"{total} {user_total}") + else: + print(total) + class ListCommand(Command): log = logging.getLogger("Bongo.QueueTool") @@ -364,6 +387,7 @@ class ExpireCommand(Command): parser.add_command(FlushCommand(), "Queue Management") parser.add_command(SpaceCommand(), "Queue Management") parser.add_command(ListCommand(), "Queue Management") +parser.add_command(CollectedCountCommand(), "Queue Management") parser.add_command(ShowCommand(), "Queue Management") parser.add_command(MessageCommand(), "Queue Management") parser.add_command(HoldLocalCommand(), "Queue Recovery and Diagnostics") diff --git a/src/apps/queuetool/tests/test_queue_client.py b/src/apps/queuetool/tests/test_queue_client.py index 6c64498..1c29b55 100644 --- a/src/apps/queuetool/tests/test_queue_client.py +++ b/src/apps/queuetool/tests/test_queue_client.py @@ -72,6 +72,22 @@ class QueueClientTests(unittest.TestCase): with self.assertRaisesRegex(ValueError, "between 0 and 999"): client.Create(1000) + def test_count_collected_returns_global_and_user_counts(self): + client = QueueClient.__new__(QueueClient) + client.stream = Stream( + [Response(1000, "47 12 Collected queue counts")], b"") + + self.assertEqual(client.CountCollected("test1"), (47, 12)) + self.assertEqual( + client.stream.commands, ["QCOUNT COLLECTED test1"]) + + def test_count_collected_rejects_user_injection(self): + client = QueueClient.__new__(QueueClient) + client.stream = Stream([], b"") + + with self.assertRaisesRegex(ValueError, "invalid queue user"): + client.CountCollected("test1\r\nQDELE 007-deadbee") + def test_retrieve_message_reads_declared_payload(self): client = QueueClient.__new__(QueueClient) client.stream = Stream( diff --git a/src/libs/cal/cal-convert.c b/src/libs/cal/cal-convert.c index fda5bc8..2cbd860 100644 --- a/src/libs/cal/cal-convert.c +++ b/src/libs/cal/cal-convert.c @@ -19,6 +19,8 @@ * ****************************************************************************/ +#include + #include #include #include @@ -689,7 +691,8 @@ IcalRecurPropertyToJsonInt(char *key, char *value, BongoJsonObject *obj) static void JsonRecurPropertyToIcalInt(char *key, BongoJsonNode *value, BongoStringBuilder *sb) { - BongoStringBuilderAppendF(sb, "%s=%d", key, BongoJsonNodeAsInt(value)); + BongoStringBuilderAppendF(sb, "%s=%" PRId64, key, + BongoJsonNodeAsInt(value)); } static void diff --git a/src/libs/json/CMakeLists.txt b/src/libs/json/CMakeLists.txt index 12b23c7..267b82d 100644 --- a/src/libs/json/CMakeLists.txt +++ b/src/libs/json/CMakeLists.txt @@ -17,4 +17,8 @@ if(BUILD_TESTING) add_executable(json-equality-test tests/equality-test.c) target_link_libraries(json-equality-test PRIVATE bongojson) add_test(NAME json-equality COMMAND json-equality-test) + + add_executable(json-int64-test tests/int64-test.c) + target_link_libraries(json-int64-test PRIVATE bongojson GLib2::GLib2) + add_test(NAME json-int64 COMMAND json-int64-test) endif() diff --git a/src/libs/json/json.c b/src/libs/json/json.c index 0bb6e38..65b2043 100644 --- a/src/libs/json/json.c +++ b/src/libs/json/json.c @@ -23,6 +23,8 @@ #include #include #include +#include +#include #define PRETTY_JSON 1 @@ -139,6 +141,19 @@ BongoJsonJPathGetBool(BongoJsonNode *n, const char *path, BOOL *val) { BongoJsonResult BongoJsonJPathGetInt(BongoJsonNode *n, const char *path, int *val) { + BongoJsonNode *result = BongoJsonJPath(n, path); + if (!result) return BONGO_JSON_NOT_FOUND; + if (result->type == BONGO_JSON_INT && + BongoJsonNodeAsInt(result) >= INT_MIN && + BongoJsonNodeAsInt(result) <= INT_MAX) { + *val = (int)BongoJsonNodeAsInt(result); + return BONGO_JSON_OK; + } + return BONGO_JSON_BAD_TYPE; +} + +BongoJsonResult +BongoJsonJPathGetInt64(BongoJsonNode *n, const char *path, int64_t *val) { BongoJsonNode *result = BongoJsonJPath(n, path); if (!result) return BONGO_JSON_NOT_FOUND; if (result->type == BONGO_JSON_INT) { @@ -228,7 +243,7 @@ BongoJsonNodeNewDouble(double val) } BongoJsonNode * -BongoJsonNodeNewInt(int val) +BongoJsonNodeNewInt(int64_t val) { BongoJsonNode *node = MemMalloc0(sizeof(BongoJsonNode)); node->type = BONGO_JSON_INT; @@ -393,7 +408,8 @@ BongoJsonNodeToStringBuilderInternal(BongoJsonNode *node, BongoStringBuilder *sb BongoStringBuilderAppendF(sb, "%f", BongoJsonNodeAsDouble(node)); break; case BONGO_JSON_INT : - BongoStringBuilderAppendF(sb, "%d", BongoJsonNodeAsInt(node)); + BongoStringBuilderAppendF(sb, "%" PRId64, + BongoJsonNodeAsInt(node)); break; case BONGO_JSON_STRING : BongoJsonQuoteStringToStringBuilder(BongoJsonNodeAsString(node), sb); @@ -834,16 +850,36 @@ BongoJsonObjectGetInt(BongoJsonObject *obj, const char *key, int *val) return res; } - if (node->type != BONGO_JSON_INT) { + if (node->type != BONGO_JSON_INT || + node->value.intVal < INT_MIN || node->value.intVal > INT_MAX) { *val = 0; return BONGO_JSON_BAD_TYPE; } - *val = node->value.intVal; + *val = (int)node->value.intVal; return BONGO_JSON_OK; } +BongoJsonResult +BongoJsonObjectGetInt64(BongoJsonObject *obj, const char *key, int64_t *val) +{ + BongoJsonNode *node; + BongoJsonResult res; + + res = BongoJsonObjectGet(obj, key, &node); + if (res != BONGO_JSON_OK) { + *val = 0; + return res; + } + if (node->type != BONGO_JSON_INT) { + *val = 0; + return BONGO_JSON_BAD_TYPE; + } + *val = node->value.intVal; + return BONGO_JSON_OK; +} + BongoJsonResult BongoJsonObjectGetString(BongoJsonObject *obj, const char *key, const char **val) { diff --git a/src/libs/json/parse.c b/src/libs/json/parse.c index 8ac2294..12e8a98 100644 --- a/src/libs/json/parse.c +++ b/src/libs/json/parse.c @@ -38,6 +38,8 @@ copies or substantial portions of the Software. #include #include #include "assert.h" +#include +#include /* FIXME: this just parses entire buffers, we really want something * that can process streams */ @@ -490,12 +492,13 @@ BongoJsonParserNextNode(BongoJsonParser *t) } if ((b >= '0' && b <= '9') || b == '.' || b == '-' || b == '+') { - int i; + int64_t i; double d; char *endp; - i = strtol(s, &endp, 10); - if (*endp == '\0') { + errno = 0; + i = strtoll(s, &endp, 10); + if (*endp == '\0' && errno != ERANGE) { MemFree(s); return BongoJsonNodeNewInt(i); } diff --git a/src/libs/json/tests/int64-test.c b/src/libs/json/tests/int64-test.c new file mode 100644 index 0000000..727113e --- /dev/null +++ b/src/libs/json/tests/int64-test.c @@ -0,0 +1,50 @@ +/**************************************************************************** + * + * Copyright (c) 2001 Novell, Inc. All Rights Reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of version 2 of the GNU General Public License + * as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, contact Novell, Inc. + * + ****************************************************************************/ + +#include +#include +#include + +#include + +int +main(void) +{ + const int64_t expected = INT64_C(68719476736); + BongoJsonNode *node = NULL; + BongoJsonObject *object; + int64_t actual = 0; + int narrow = 0; + char *serialized; + + assert(BongoJsonParseString( + "{\"minimumfreespace\":68719476736}", &node) == + BONGO_JSON_OK); + assert(node != NULL && node->type == BONGO_JSON_OBJECT); + object = BongoJsonNodeAsObject(node); + assert(BongoJsonObjectGetInt64( + object, "minimumfreespace", &actual) == BONGO_JSON_OK); + assert(actual == expected); + assert(BongoJsonObjectGetInt( + object, "minimumfreespace", &narrow) == BONGO_JSON_BAD_TYPE); + serialized = BongoJsonNodeToString(node); + assert(strstr(serialized, "68719476736") != NULL); + MemFree(serialized); + BongoJsonNodeFree(node); + return 0; +} diff --git a/src/libs/msgapi/msgapi.c b/src/libs/msgapi/msgapi.c index 5964338..a62c64a 100644 --- a/src/libs/msgapi/msgapi.c +++ b/src/libs/msgapi/msgapi.c @@ -29,6 +29,8 @@ #define CURRENT_PRODUCT_VERSION_MAJOR 5 #define CURRENT_PRODUCT_VERSION_MINOR 3 +#include + #include #include #include @@ -450,7 +452,8 @@ MsgGetConfigProperty(unsigned char *Buffer, unsigned char *Property) result = TRUE; break; case BONGO_JSON_INT: - snprintf((char *)Buffer, MSGSRV_CONFIG_MAX_PROP_CHARS + 1, "%d", + snprintf((char *)Buffer, MSGSRV_CONFIG_MAX_PROP_CHARS + 1, + "%" PRId64, BongoJsonNodeAsInt(value)); result = TRUE; break; diff --git a/src/libs/python/bongo/configuration/model.py b/src/libs/python/bongo/configuration/model.py index 7ee3188..10bef16 100644 --- a/src/libs/python/bongo/configuration/model.py +++ b/src/libs/python/bongo/configuration/model.py @@ -873,6 +873,16 @@ def _validate_scanners(configurations: Mapping[str, Any], issues: list[Issue]) - def _validate_collector(config: Mapping[str, Any], issues: list[Issue], check_files: bool) -> None: + maximum = config.get("maximum_pending_messages") + per_user = config.get("maximum_pending_messages_per_user") + _positive(issues, "collector", config, "maximum_pending_messages") + _positive(issues, "collector", config, + "maximum_pending_messages_per_user") + if (isinstance(maximum, int) and not isinstance(maximum, bool) and + isinstance(per_user, int) and not isinstance(per_user, bool) and + per_user > maximum): + _issue(issues, "collector", "maximum_pending_messages_per_user", + "must not exceed maximum_pending_messages") ca_file = config.get("ca_file", "") if not isinstance(ca_file, str): _issue(issues, "collector", "ca_file", diff --git a/src/libs/python/bongo/store/QueueClient.py b/src/libs/python/bongo/store/QueueClient.py index ae3bb54..4e688b3 100644 --- a/src/libs/python/bongo/store/QueueClient.py +++ b/src/libs/python/bongo/store/QueueClient.py @@ -36,6 +36,27 @@ class QueueClient: raise bongo.BongoError(r.message) return r + def CountCollected(self, user=None): + command = "QCOUNT COLLECTED" + if user is not None: + if (not user or any(character.isspace() for character in user) or + any(ord(character) < 0x20 or ord(character) == 0x7f + for character in user)): + raise ValueError("invalid queue user") + command += " %s" % user + self.stream.Write(command) + r = self.stream.GetResponse() + if r.code != 1000: + raise bongo.BongoError(r.message) + fields = r.message.split() + if len(fields) < 2: + raise bongo.BongoError("Malformed collected queue count response") + try: + return int(fields[0]), int(fields[1]) + except ValueError as error: + raise bongo.BongoError( + "Malformed collected queue count response") from error + def Delete(self, queue_id): self.stream.Write("QDELE %s" % queue_id) r = self.stream.GetResponse() diff --git a/src/libs/python/libbongo/json.c b/src/libs/python/libbongo/json.c index 5eeac67..064357f 100644 --- a/src/libs/python/libbongo/json.c +++ b/src/libs/python/libbongo/json.c @@ -243,7 +243,7 @@ JsonNodeToPy(BongoJsonNode *node, BOOL own, BOOL native) ret = Py_BuildValue("d", BongoJsonNodeAsDouble(node)); break; case BONGO_JSON_INT : - ret = Py_BuildValue("i", BongoJsonNodeAsInt(node)); + ret = PyLong_FromLongLong(BongoJsonNodeAsInt(node)); break; case BONGO_JSON_STRING : str = BongoJsonNodeAsString(node); diff --git a/src/libs/util/bongoagent.c b/src/libs/util/bongoagent.c index 41e14cb..5b91b88 100644 --- a/src/libs/util/bongoagent.c +++ b/src/libs/util/bongoagent.c @@ -173,7 +173,7 @@ SetBongoConfigItem(BongoConfigItem *schema, BongoJsonNode *node) { break; case BONGO_JSON_INT: { BongoConfigIntegerDestination *destination = schema->destination; - int value = BongoJsonNodeAsInt(node); + int64_t value = BongoJsonNodeAsInt(node); if (!SetBongoConfigInteger(destination, value)) { XplConsolePrintf("config: unsupported integer destination size %zu at %s\r\n", diff --git a/src/libs/util/config-integer.c b/src/libs/util/config-integer.c index b68d1c9..6ac98f8 100644 --- a/src/libs/util/config-integer.c +++ b/src/libs/util/config-integer.c @@ -22,7 +22,7 @@ #include BOOL -SetBongoConfigInteger(BongoConfigIntegerDestination *destination, int value) +SetBongoConfigInteger(BongoConfigIntegerDestination *destination, int64_t value) { if (!destination || !destination->pointer) return FALSE; diff --git a/src/libs/util/tests/config-integer-test.c b/src/libs/util/tests/config-integer-test.c index 79ee79c..b97c8ae 100644 --- a/src/libs/util/tests/config-integer-test.c +++ b/src/libs/util/tests/config-integer-test.c @@ -23,7 +23,8 @@ #include static void -CheckIntegerWidth(size_t offset, size_t width, int input, uint64_t expected) +CheckIntegerWidth(size_t offset, size_t width, int64_t input, + uint64_t expected) { unsigned char storage[32]; BongoConfigIntegerDestination destination; @@ -51,5 +52,7 @@ main(void) CheckIntegerWidth(1U, sizeof(uint32_t), -7, UINT64_C(0xfffffff9)); CheckIntegerWidth(1U, sizeof(uint64_t), -7, UINT64_C(0xfffffffffffffff9)); + CheckIntegerWidth(1U, sizeof(uint64_t), INT64_C(68719476736), + UINT64_C(68719476736)); return 0; }