From 721c293382a2641d7ef01bd5db538c40150f7d78 Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Thu, 15 Oct 2026 21:50:57 +0800 Subject: [PATCH] fix(delivery): fit bounded health probes into startup budgets --- backend/tests/test_delivery.py | 149 +++++++++++++++++++++++++---- scripts/release/README.md | 13 ++- scripts/release/broker_health.py | 85 ++-------------- scripts/release/broker_snapshot.py | 96 +++++++++++++++++++ scripts/release/node_health.py | 1 + scripts/release/operator_units.py | 8 +- 6 files changed, 251 insertions(+), 101 deletions(-) create mode 100644 scripts/release/broker_snapshot.py diff --git a/backend/tests/test_delivery.py b/backend/tests/test_delivery.py index 929ca64..d658a14 100644 --- a/backend/tests/test_delivery.py +++ b/backend/tests/test_delivery.py @@ -2233,46 +2233,159 @@ def test_proxy_render_is_fixed_to_a_selected_snapshot_without_business_credentia assert "EnvironmentFile=" not in text and "LoadCredential=" not in text -@pytest.mark.parametrize("name", ["queues", "vhost_limits", "users", "permissions"]) @pytest.mark.parametrize("failure", ["deadline", "exit", "output"]) -def test_broker_query_failure_identifies_fixed_operation_without_leaking_process_data( - monkeypatch, name, failure -): - import subprocess +def test_broker_snapshot_failure_keeps_process_details_private(monkeypatch, failure): from types import SimpleNamespace - from scripts.release import broker_health + from scripts.release import broker_snapshot secret = "synthetic-private-process-detail" def failed(command, *, stdout, stderr, timeout, env): - assert stderr == subprocess.DEVNULL - assert timeout == 8 - assert command[-1] == "--formatter=json" + assert stderr == subprocess.DEVNULL and timeout == 30 + assert command == [ + str(broker_snapshot.CTL), + "-q", + "-n", + "kf@localhost", + "--timeout", + "5", + "eval", + broker_snapshot.SNAPSHOT_EXPRESSION, + "--formatter=json", + ] + assert env["ERL_EPMD_ADDRESS"] == "127.0.0.1" + assert "rpc:call(node(), erlang, apply," in command[7] + assert command[7].endswith("end, []], 5000).") if failure == "deadline": raise subprocess.TimeoutExpired([secret], timeout, stderr=secret.encode()) if failure == "output": stdout.write(b"x" * 32769) return SimpleNamespace(returncode=1 if failure == "exit" else 0) - monkeypatch.setattr(broker_health.subprocess, "run", failed) + monkeypatch.setattr(broker_snapshot.subprocess, "run", failed) suffix = {"deadline": "exceeded deadline", "exit": "failed", "output": "output exceeds bound"} with pytest.raises(policy.DeliveryError) as raised: - broker_health.command(name) - assert str(raised.value) == "broker " + name + " query " + suffix[failure] + broker_snapshot.command() + assert str(raised.value) == "broker snapshot query " + suffix[failure] assert secret not in str(raised.value) +@pytest.mark.parametrize("change", [None, "node", "result", "rpc_failure", "extra", "duplicate"]) +def test_broker_snapshot_accepts_only_the_pinned_node_envelope(monkeypatch, change): + from types import SimpleNamespace + + from scripts.release import broker_snapshot + + snapshot = {"queues": [], "vhost_limits": [], "users": [], "permissions": []} + envelope = {"result": "ok", "node": "kf@localhost", "value": json.dumps(snapshot)} + if change == "node": + envelope["node"] = "other@localhost" + elif change == "result": + envelope["result"] = "error" + elif change == "rpc_failure": + envelope["value"] = ["badrpc", "timeout"] + elif change == "extra": + envelope["unknown"] = True + raw = json.dumps(envelope).encode() + if change == "duplicate": + raw = b'{"node":"kf@localhost",' + raw[1:] + + def query(command, *, stdout, **kwargs): + stdout.write(raw) + return SimpleNamespace(returncode=0) + + monkeypatch.setattr(broker_snapshot.subprocess, "run", query) + if change is None: + assert broker_snapshot.command() == snapshot + else: + with pytest.raises(policy.DeliveryError): + broker_snapshot.command() + + @pytest.mark.parametrize("name", ["delete_queue", "users --delete", None]) -def test_broker_query_selector_never_accepts_extra_commands_or_flags(monkeypatch, name): - from scripts.release import broker_health +def test_broker_snapshot_has_no_command_or_expression_selector(monkeypatch, name): + from scripts.release import broker_snapshot def unexpected(*args, **kwargs): - pytest.fail("unknown selector launched a process") + pytest.fail("caller text launched a process") - monkeypatch.setattr(broker_health.subprocess, "run", unexpected) - with pytest.raises(policy.DeliveryError, match="unknown broker metadata query"): - broker_health.command(name) + monkeypatch.setattr(broker_snapshot.subprocess, "run", unexpected) + with pytest.raises(TypeError): + broker_snapshot.command(name) + + +@pytest.mark.parametrize( + "kind,value", [("longstr", "classic"), ("longstr", "quorum"), ("long", "classic")] +) +def test_broker_arguments_allow_only_exact_automatic_classic_type(kind, value): + args = [ + [k, "longstr" if k == "x-overflow" else "long", v] + for k, v in dependencies.QUEUE_ARGUMENTS.items() + ] + args.append(["x-queue-type", kind, value]) + if (kind, value) == ("longstr", "classic"): + assert broker_health.queue_arguments(args) == dependencies.QUEUE_ARGUMENTS + supplied = {**dependencies.QUEUE_ARGUMENTS, "x-queue-type": "classic"} + assert broker_health.queue_arguments(supplied) == dependencies.QUEUE_ARGUMENTS + assert supplied["x-queue-type"] == "classic" + args[-1] = args[0].copy() + with pytest.raises(policy.DeliveryError): + broker_health.queue_arguments(args) + + +@pytest.mark.parametrize( + "name", ["broker-health", "factory-health", "import-health", "outbox-health"] +) +@pytest.mark.parametrize("cpu", ["100000 100000", "max 100000", "200000 100000"]) +def test_fixed_health_guard_still_requires_exact_helper_and_node_cpu(monkeypatch, name, cpu): + import resource + + from scripts.release import operator_units + + _, uid, memory, pids, expected_cpu, _ = operator_units.PROFILES[name] + assert expected_cpu == 100 + monkeypatch.setattr(operator_units.os, "getuid", lambda: uid) + monkeypatch.setattr(operator_units.os, "geteuid", lambda: uid) + monkeypatch.setattr(operator_units.os, "getgid", lambda: uid) + monkeypatch.setattr(operator_units.os, "getgroups", lambda: [uid]) + group = Path("/sys/fs/cgroup/kfonlinejudge.slice/kf-" + name + ".service") + monkeypatch.setattr(operator_units.preflight, "current_cgroup", lambda: group) + observed = { + "memory.max": str(memory * 1024**2), + "memory.swap.max": "0", + "pids.max": str(pids), + "cpu.max": cpu, + } + node = { + "memory.max": str(3840 * 1024**2), + "memory.swap.max": "0", + "pids.max": "384", + "cpu.max": "200000 100000", + } + + def read(path, *args, **kwargs): + if str(path) == "/proc/self/status": + return "NoNewPrivs: 1\n" + "\n".join( + k + ": 0" for k in ("CapInh", "CapPrm", "CapEff", "CapBnd", "CapAmb") + ) + return (observed if path.parent == group else node)[path.name] + + monkeypatch.setattr(Path, "read_text", read) + limits = { + resource.RLIMIT_CORE: 0, + resource.RLIMIT_FSIZE: operator_units.FILE_LIMIT_MIB.get(name, 16) * 1024**2, + resource.RLIMIT_NOFILE: 256, + resource.RLIMIT_STACK: 16 * 1024**2, + } + monkeypatch.setattr( + operator_units.resource, "getrlimit", lambda key: (limits[key], limits[key]) + ) + if cpu == "100000 100000": + operator_units.guarded(name) + else: + with pytest.raises(policy.DeliveryError, match="budgets differ"): + operator_units.guarded(name) @pytest.mark.parametrize( diff --git a/scripts/release/README.md b/scripts/release/README.md index 04ee5a8..6d5ea3d 100644 --- a/scripts/release/README.md +++ b/scripts/release/README.md @@ -415,9 +415,16 @@ end-to-end target still requires measured acceptance. Fixed health helpers include interpreter startup in their wall-clock deadline. Outbox and import queries have 60 seconds each; factory status has 30 seconds and broker metadata has 45 seconds. Node readiness has 300 seconds for the sequential -checks. Each helper keeps its fixed CPU, memory, process and privilege limits. -These operator deadlines do not change student execution budgets. An expired or -failed query keeps admission closed. +checks. Each ordinary helper has one CPU within the node's two-CPU ceiling; +its memory, process and privilege limits remain fixed. These operator budgets +do not change student execution budgets. Startup opens admission only after all +required readiness queries succeed. + +Broker metadata uses one fixed read-only RPC with a five-second remote deadline +and a 30-second CLI process deadline, within the helper's 45-second limit. It reads +queue metadata, vhost limits, usernames/tags and permissions; it does not read +messages or password hashes. Readiness requires exactly the configured classic +queue and budgets, including RabbitMQ's optional `x-queue-type=classic` argument. The broker helper has a 64 MiB per-file ceiling for Erlang JIT mappings. [OTP 27.3.4](https://github.com/erlang/otp/blob/OTP-27.3.4/erts/emulator/beam/jit/beam_jit_main.cpp) diff --git a/scripts/release/broker_health.py b/scripts/release/broker_health.py index a1035df..2a9de2a 100644 --- a/scripts/release/broker_health.py +++ b/scripts/release/broker_health.py @@ -4,56 +4,33 @@ import argparse import json import stat import subprocess -import tempfile from pathlib import Path +from . import broker_snapshot, operator_units, preflight from . import dependency_policy as dependencies -from . import operator_units, preflight from .policy import DeliveryError, canonical, fields, integer -CTL = Path("/usr/lib/rabbitmq/sbin/rabbitmqctl") -QUERIES = { - "queues": [ - "list_queues", - "-p", - "kf", - "name", - "durable", - "auto_delete", - "exclusive", - "type", - "state", - "arguments", - "messages_ready", - "messages_unacknowledged", - "message_bytes", - "consumers", - "policy", - "operator_policy", - "effective_policy_definition", - ], - "vhost_limits": ["list_vhost_limits", "--global"], - "users": ["list_users"], - "permissions": ["list_permissions", "-p", "kf"], -} +CTL = broker_snapshot.CTL def queue_arguments(value) -> dict: if isinstance(value, dict): - result = value - elif isinstance(value, list) and len(value) == 4: + result = dict(value) + elif isinstance(value, list) and len(value) in (4, 5): result = {} for row in value: if not isinstance(row, list) or len(row) != 3 or row[0] in result: raise DeliveryError("duplicate/unknown broker argument") name, kind, argument = row - if name not in dependencies.QUEUE_ARGUMENTS or kind != ( - "longstr" if name == "x-overflow" else "long" + if name not in {*dependencies.QUEUE_ARGUMENTS, "x-queue-type"} or kind != ( + "longstr" if name in {"x-overflow", "x-queue-type"} else "long" ): raise DeliveryError("broker argument type differs") result[name] = argument else: raise DeliveryError("broker argument shape differs") + if "x-queue-type" in result and result.pop("x-queue-type") != "classic": + raise DeliveryError("actual broker queue type differs") if canonical(result) != canonical(dependencies.QUEUE_ARGUMENTS): raise DeliveryError("actual broker queue budgets differ") return result @@ -148,49 +125,6 @@ def validate_snapshot(value: dict) -> dict: } -def command(name: str): - if name not in QUERIES: - raise DeliveryError("unknown broker metadata query") - # A file-backed capped stream prevents malicious metadata/log output from - # allocating an unbounded capture buffer. The unit adds a hard file limit. - env = { - "PATH": "/usr/bin:/bin", - "LC_ALL": "C.UTF-8", - "HOME": dependencies.STORAGE["rabbitmq"]["path"], - "ERL_EPMD_ADDRESS": "127.0.0.1", - "RABBITMQ_NODENAME": "kf@localhost", - "RABBITMQ_CTL_ERL_ARGS": ( - "+S 1:1 +A 2 +P 1024 +Q 1024 -kernel inet_dist_use_interface {127,0,0,1}" - ), - } - with tempfile.TemporaryFile() as output: - try: - result = subprocess.run( - [ - str(CTL), - "-q", - "-n", - "kf@localhost", - "--timeout", - "5", - *QUERIES[name], - "--formatter=json", - ], - stdout=output, - stderr=subprocess.DEVNULL, - timeout=8, - env=env, - ) - except subprocess.TimeoutExpired as error: - raise DeliveryError("broker " + name + " query exceeded deadline") from error - if result.returncode: - raise DeliveryError("broker " + name + " query failed") - if output.tell() > 32768: - raise DeliveryError("broker " + name + " query output exceeds bound") - output.seek(0) - return operator_units.decode(output.read(32769)) - - def observe() -> dict: operator_units.guarded("broker-health") preflight.protected(CTL) @@ -202,8 +136,7 @@ def observe() -> dict: or (info.st_uid, info.st_gid, stat.S_IMODE(info.st_mode)) != (1204, 1204, 0o400) ): raise DeliveryError("ordinary private broker cookie required") - data = {name: command(name) for name in QUERIES} - return validate_snapshot(data) + return validate_snapshot(broker_snapshot.command()) def main() -> None: diff --git a/scripts/release/broker_snapshot.py b/scripts/release/broker_snapshot.py new file mode 100644 index 0000000..6291395 --- /dev/null +++ b/scripts/release/broker_snapshot.py @@ -0,0 +1,96 @@ +"""One fixed read-only RabbitMQ 4.3.6 metadata RPC; no caller expressions.""" + +import subprocess +import tempfile +from pathlib import Path + +from . import dependency_policy, operator_units +from .policy import DeliveryError, fields + +CTL = Path("/usr/lib/rabbitmq/sbin/rabbitmqctl") +# Fixed getters only: no caller text, mutation, message reads or password hashes. +SNAPSHOT_EXPRESSION = r""" +rpc:call(node(), erlang, apply, [fun() -> + Count = rabbit_amqqueue:count(<<"kf">>), + true = (Count =:= 0 orelse Count =:= 1), + Normalize = fun + (name, {resource, <<"kf">>, queue, Value}) -> Value; + (arguments, Value) -> [[K, atom_to_binary(T, utf8), V] || {K,T,V} <- Value]; + (type, rabbit_classic_queue) -> <<"classic">>; + (K, Value) when K =:= policy; K =:= operator_policy -> + case Value of undefined -> <<>>; none -> <<>>; '' -> <<>>; _ -> Value end; + (effective_policy_definition, Value) when is_list(Value) -> maps:from_list(Value); + (tags, Value) -> [case T of + A when is_atom(A) -> atom_to_binary(A,utf8); _ -> T end || T <- Value]; + (_, Value) when is_atom(Value), Value =/= true, Value =/= false -> + atom_to_binary(Value,utf8); + (_, Value) -> Value + end, + ToMap = fun(Items) -> maps:from_list( + [{atom_to_binary(K,utf8),Normalize(K,V)} || {K,V} <- Items]) end, + Queues = [ToMap(Row) || Row <- rabbit_amqqueue:info_all(<<"kf">>, + [name,durable,auto_delete,exclusive,type,state,arguments,messages_ready, + messages_unacknowledged,message_bytes,consumers,policy,operator_policy, + effective_policy_definition])], + Limits = [#{<<"vhost">> => V, + <<"limits">> => iolist_to_binary(rabbit_json:encode(maps:from_list(L)))} + || {V,L} <- rabbit_vhost_limit:list()], + Users = [ToMap(Row) || Row <- rabbit_auth_backend_internal:list_users()], + Permissions = [ToMap(Row) || Row <- + rabbit_auth_backend_internal:list_vhost_permissions(<<"kf">>)], + iolist_to_binary(rabbit_json:encode(#{<<"queues">> => Queues, <<"vhost_limits">> => Limits, + <<"users">> => Users, <<"permissions">> => Permissions})) +end, []], 5000). +""".strip() + + +def command() -> dict: + env = { + "PATH": "/usr/bin:/bin", + "LC_ALL": "C.UTF-8", + "HOME": dependency_policy.STORAGE["rabbitmq"]["path"], + "ERL_EPMD_ADDRESS": "127.0.0.1", + "RABBITMQ_NODENAME": "kf@localhost", + "RABBITMQ_CTL_ERL_ARGS": ( + "+S 1:1 +A 2 +P 1024 +Q 1024 -kernel inet_dist_use_interface {127,0,0,1}" + ), + } + with tempfile.TemporaryFile() as output: + try: + result = subprocess.run( + [ + str(CTL), + "-q", + "-n", + "kf@localhost", + "--timeout", + "5", + "eval", + SNAPSHOT_EXPRESSION, + "--formatter=json", + ], + stdout=output, + stderr=subprocess.DEVNULL, + timeout=30, + env=env, + ) + except subprocess.TimeoutExpired as error: + raise DeliveryError("broker snapshot query exceeded deadline") from error + if result.returncode: + raise DeliveryError("broker snapshot query failed") + if output.tell() > 32768: + raise DeliveryError("broker snapshot query output exceeds bound") + output.seek(0) + value = operator_units.decode(output.read(32769)) + fields(value, {"result", "node", "value"}, "broker snapshot envelope") + if ( + value["result"] != "ok" + or value["node"] != "kf@localhost" + or not isinstance(value["value"], str) + ): + raise DeliveryError("broker snapshot result or node differs") + return fields( + operator_units.decode(value["value"].encode()), + {"queues", "vhost_limits", "users", "permissions"}, + "broker snapshot", + ) diff --git a/scripts/release/node_health.py b/scripts/release/node_health.py index 97b978d..84346f0 100644 --- a/scripts/release/node_health.py +++ b/scripts/release/node_health.py @@ -66,6 +66,7 @@ def binding() -> tuple[dict, dict, Path]: "availability.py", "node_health.py", "broker_health.py", + "broker_snapshot.py", "factory_health.py", "import_health.py", "operator_units.py", diff --git a/scripts/release/operator_units.py b/scripts/release/operator_units.py index 5291877..064aa59 100644 --- a/scripts/release/operator_units.py +++ b/scripts/release/operator_units.py @@ -11,10 +11,10 @@ from .policy import DeliveryError, invalid_constant, unique_object PROFILES = { "node-health": ("root", 0, 128, 16, 50, 300), - "broker-health": ("rabbitmq", 1204, 256, 64, 50, 45), - "factory-health": ("kf-worker", 1202, 32, 4, 10, 30), - "import-health": ("kf-oj", 1200, 128, 16, 25, 60), - "outbox-health": ("kf-oj", 1200, 128, 16, 25, 60), + "broker-health": ("rabbitmq", 1204, 256, 64, 100, 45), + "factory-health": ("kf-worker", 1202, 32, 4, 100, 30), + "import-health": ("kf-oj", 1200, 128, 16, 100, 60), + "outbox-health": ("kf-oj", 1200, 128, 16, 100, 60), "node-maintenance": ("root", 0, 256, 32, 50, 900), "node-observation": ("root", 0, 128, 16, 25, 360), }