feat(delivery): compose bounded local database and broker services

This commit is contained in:
2026-10-13 02:49:19 +08:00
parent 40b9a69e5a
commit 480db0e74f
3 changed files with 496 additions and 0 deletions
+419
View File
@@ -0,0 +1,419 @@
"""Finite local DB/broker composition; installed limits require separate readback."""
import argparse
import base64
import hashlib
import json
import os
import resource
import secrets
import stat
from pathlib import Path
from urllib.parse import urlsplit
from . import configuration, install, preflight, service_install
from .policy import DeliveryError, canonical, digest
from .services import python_environment
QUEUE = "kf.judge.evaluations.v1"
QUEUE_ARGUMENTS = {
"x-max-length": 64,
"x-max-length-bytes": 32768,
"x-overflow": "reject-publish",
"x-message-ttl": 600000,
}
STORAGE = {
"postgres": {"path": "/var/lib/kf-onlinejudge/postgres", "filesystem_bytes": 2 * 1024**3},
"rabbitmq": {"path": "/var/lib/kf-onlinejudge/rabbitmq", "filesystem_bytes": 512 * 1024**2},
}
def check_runtime(name: str) -> dict:
uid = {"postgres": 1203, "rabbitmq": 1204}[name]
if (
os.getuid() != uid
or os.geteuid() != uid
or os.getgid() != uid
or not set(os.getgroups()) <= {uid}
):
raise DeliveryError("dedicated ordinary dependency identity required")
status = Path("/proc/self/status").read_text()
values = dict(line.split(":", 1) for line in status.splitlines() if ":" in line)
if values.get("NoNewPrivs", "").strip() != "1" or any(
not values.get(key, "").strip() or int(values[key].strip(), 16) != 0
for key in ("CapInh", "CapPrm", "CapEff", "CapBnd", "CapAmb")
):
raise DeliveryError("dependency privilege boundary differs")
group = preflight.current_cgroup()
if str(group) != "/sys/fs/cgroup/kfonlinejudge.slice/kf-" + name + ".service":
raise DeliveryError("dependency cgroup identity differs")
limits = {
key: (group / key).read_text().strip()
for key in (
"memory.max",
"memory.swap.max",
"pids.max",
"cpu.max",
)
}
if limits != {
"memory.max": "536870912",
"memory.swap.max": "0",
"pids.max": "64",
"cpu.max": "100000 100000",
}:
raise DeliveryError("effective dependency budgets differ")
node = Path("/sys/fs/cgroup/kfonlinejudge.slice")
actual_node = {key: (node / key).read_text().strip() for key in limits}
if actual_node != {
"memory.max": "4026531840",
"memory.swap.max": "0",
"pids.max": "384",
"cpu.max": "200000 100000",
}:
raise DeliveryError("effective dependency node budget differs")
for key, maximum in (
(resource.RLIMIT_CORE, 0),
(resource.RLIMIT_FSIZE, 256 * 1024**2),
(resource.RLIMIT_NOFILE, 256),
(resource.RLIMIT_STACK, 16 * 1024**2),
):
if resource.getrlimit(key) != (maximum, maximum):
raise DeliveryError("effective dependency process limits differ")
directory = Path(STORAGE[name]["path"])
info = directory.lstat()
if (
not stat.S_ISDIR(info.st_mode)
or info.st_uid != uid
or info.st_gid != uid
or stat.S_IMODE(info.st_mode) != 0o700
):
raise DeliveryError("private dependency data identity/mode differs")
# A parent filesystem and soft DB/broker alarm must not pass as the
# separately bounded data filesystem required by the service profile.
mounts = [line.split() for line in Path("/proc/self/mountinfo").read_text().splitlines()]
if not any(len(row) > 6 and row[4] == str(directory) and row[3] == "/" for row in mounts):
raise DeliveryError("separate dependency storage mount required")
fs = os.statvfs(directory)
capacity = fs.f_frsize * fs.f_blocks
required = STORAGE[name]["filesystem_bytes"]
if not required * 9 // 10 <= capacity <= required:
raise DeliveryError("effective dependency filesystem capacity differs")
return {
"dependency": name,
"actual_cgroup_limits": limits,
"data_filesystem_bytes": capacity,
"execution_verified": False,
}
def salted_password(password: str) -> str:
salt = secrets.token_bytes(4)
return base64.b64encode(salt + hashlib.sha256(salt + password.encode()).digest()).decode()
def broker_definitions(connections: dict) -> dict:
connections = configuration.validate_connections(connections)
users = []
permissions = []
queue = r"^kf\.judge\.evaluations\.v1$"
for role in ("outbox", "worker"):
endpoint = urlsplit(connections["amqp"][role])
users.append(
{
"name": "kf_" + role,
"password_hash": salted_password(endpoint.password),
"hashing_algorithm": "rabbit_password_hashing_sha256",
"tags": [],
"limits": {"max-connections": 2, "max-channels": 4},
}
)
permissions.append(
{
"user": "kf_" + role,
"vhost": "kf",
"configure": queue,
"write": r"^(amq\.default|kf\.judge\.evaluations\.v1)$"
if role == "outbox"
else "^$",
"read": queue if role == "worker" else "^$",
}
)
return {
"users": users,
# In 4.3.6 add_vhost does not apply the exported `limits` member.
# Fixed default_limits.vhosts in rabbitmq.conf applies on creation;
# existing-vhost limits must be read back before service readiness.
"vhosts": [{"name": "kf", "metadata": {"default_queue_type": "classic"}}],
"permissions": permissions,
"queues": [
{
"name": QUEUE,
"vhost": "kf",
"durable": True,
"auto_delete": False,
"arguments": QUEUE_ARGUMENTS,
}
],
}
def render(release: str, batch: str) -> dict[str, bytes]:
# These identities are normally supplied by an installed selected receipt.
import re
if not re.fullmatch(r"sd-[0-9a-f]{12}", release) or not re.fullmatch(
r"cfg-[0-9a-f]{24}", batch
):
raise DeliveryError("invalid dependency configuration binding")
base = f"/etc/kf-onlinejudge/dependencies/{release}.{batch}"
pg = (
"\n".join(
[
"data_directory = '/var/lib/kf-onlinejudge/postgres/data'",
f"hba_file = '{base}/pg_hba.conf'",
"listen_addresses = '127.0.0.1'",
"port = 5432",
"unix_socket_directories = '/run/postgresql'",
"unix_socket_permissions = 0777",
"password_encryption = 'scram-sha-256'",
"max_connections = 32",
"superuser_reserved_connections = 3",
"shared_buffers = '64MB'",
"work_mem = '1MB'",
"hash_mem_multiplier = 1",
"maintenance_work_mem = '32MB'",
"autovacuum_work_mem = '32MB'",
"autovacuum_max_workers = 2",
"max_worker_processes = 4",
"max_parallel_workers = 0",
"max_wal_senders = 0",
"temp_file_limit = '16MB'",
"max_files_per_process = 128",
"statement_timeout = '10s'",
"lock_timeout = '2s'",
"idle_in_transaction_session_timeout = '60s'",
"max_wal_size = '64MB'",
"min_wal_size = '32MB'",
"logging_collector = off",
"log_statement = 'none'",
"log_min_duration_statement = -1",
"log_min_error_statement = 'panic'",
"log_connections = ''",
"log_disconnections = off",
]
)
+ "\n"
)
hba = (
"local all all scram-sha-256\n"
"host all all 127.0.0.1/32 scram-sha-256\n"
"host all all 0.0.0.0/0 reject\n"
"host all all ::0/0 reject\n"
)
broker = (
"\n".join(
[
"listeners.tcp.1 = 127.0.0.1:5672",
"connection_max = 8",
"channel_max = 8",
"channel_max_per_node = 16",
"ranch_connection_max = 8",
"max_message_size = 4096",
"handshake_timeout = 5000",
"heartbeat = 30",
"consumer_timeout = 1800000",
"vm_memory_high_watermark.absolute = 256MiB",
"total_memory_available_override_value = 512MiB",
"disk_free_limit.absolute = 64MiB",
"log.console = true",
"log.console.level = warning",
"log.file = false",
"definitions.import_backend = local_filesystem",
f"definitions.local.path = {base}/definitions.json",
"default_limits.vhosts.1.pattern = ^kf$",
"default_limits.vhosts.1.max_connections = 4",
"default_limits.vhosts.1.max_queues = 1",
]
)
+ "\n"
)
files = {
"postgresql.conf": pg.encode(),
"pg_hba.conf": hba.encode(),
"rabbitmq.conf": broker.encode(),
"enabled_plugins": b"[].\n",
}
for name, user in {"postgres": "postgres", "rabbitmq": "rabbitmq"}.items():
payload = install.RELEASES / release / "payload"
values = [
"[Unit]",
"Description=KF bounded local " + name,
"After=local-fs.target",
"StartLimitIntervalSec=600",
"StartLimitBurst=3",
"[Service]",
"Type=exec",
"User=" + user,
"Group=" + user,
"Slice=kfonlinejudge.slice",
"UMask=0077",
"MemoryMax=512M",
"MemorySwapMax=0",
"TasksMax=64",
"CPUQuota=100%",
"LimitCORE=0",
"LimitFSIZE=256M",
"LimitNOFILE=256",
"LimitSTACK=16M",
"TimeoutStartSec=120",
"TimeoutStopSec=30",
"Restart=on-failure",
"RestartSec=30",
"RuntimeMaxSec=24h",
"KillMode=control-group",
"NoNewPrivileges=yes",
"CapabilityBoundingSet=",
"AmbientCapabilities=",
"ProtectSystem=strict",
"ProtectHome=yes",
"ProtectControlGroups=yes",
"PrivateTmp=yes",
"PrivateDevices=yes",
"RestrictAddressFamilies=AF_UNIX AF_INET",
"RestrictSUIDSGID=yes",
"LockPersonality=yes",
"ReadWritePaths=" + STORAGE[name]["path"],
"RequiresMountsFor=" + STORAGE[name]["path"],
"StandardOutput=journal",
"StandardError=journal",
"LogRateLimitIntervalSec=30s",
"LogRateLimitBurst=100",
"ExecStartPre=+/usr/bin/env -i PATH=/usr/bin:/bin LC_ALL=C "
+ python_environment(payload)
+ f" {payload}/python/bin/python3.13"
+ " -s -P -m scripts.release.services check-binding"
+ f" --release-id {release} --configuration-id {batch}",
"ExecStartPre=/usr/bin/env -i PATH=/usr/bin:/bin LC_ALL=C "
+ python_environment(payload)
+ f" {payload}/python/bin/python3.13"
+ " -s -P -m scripts.release.dependencies --check-service "
+ name,
]
if name == "postgres":
values += [
"RuntimeDirectory=postgresql",
"RuntimeDirectoryMode=0755",
"ExecStart=/usr/bin/env -i PATH=/usr/bin:/bin LC_ALL=C.UTF-8"
f" /usr/bin/postgres -c config_file={base}/postgresql.conf",
]
else:
values += [
"WorkingDirectory=/var/lib/kf-onlinejudge/rabbitmq",
"ExecStart=/usr/bin/env -i PATH=/usr/bin:/bin LC_ALL=C.UTF-8"
" HOME=/var/lib/kf-onlinejudge/rabbitmq ERL_EPMD_ADDRESS=127.0.0.1"
" RABBITMQ_NODENAME=kf@localhost RABBITMQ_DIST_PORT=25672"
f" RABBITMQ_CONFIG_FILE={base}/rabbitmq.conf"
f" RABBITMQ_ENABLED_PLUGINS_FILE={base}/enabled_plugins"
" RABBITMQ_MNESIA_BASE=/var/lib/kf-onlinejudge/rabbitmq/data"
' RABBITMQ_LOGS=- "RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS=+S 2:2 +A 4 +P 1024 +Q 1024'
' -kernel inet_dist_use_interface {127,0,0,1}"'
" /usr/lib/rabbitmq/sbin/rabbitmq-server",
]
values += ["[Install]", "WantedBy=multi-user.target"]
files["kf-" + name + ".service"] = ("\n".join(values) + "\n").encode()
return files
def generate(connections_path: Path, output: Path) -> dict:
if os.getuid() != 0 or os.geteuid() != 0:
raise DeliveryError("dependency composition requires root operator")
selected, original = service_install.current_receipt()
payload = install.RELEASES / selected["release_id"] / "payload"
installed = payload / "delivery/scripts/release/dependencies.py"
preflight.protected(installed)
if digest(installed) != digest(Path(__file__)):
raise DeliveryError("dependency composition differs from selected delivery")
configuration.private(connections_path)
from .policy import read_json
connections = configuration.validate_connections(read_json(connections_path, 16384))
for role in ("outbox", "worker"):
# Bound to the same already-installed credential batch, not a second
# set of broker identities hidden in the dependency definition file.
if role == "outbox":
env = configuration.environment_read(
configuration.installed_path(
selected["release_id"],
selected["configuration_id"],
"outbox.env",
).read_bytes()
)
expected = env["KF_OJ_AMQP_URL"]
else:
expected = read_json(
configuration.installed_path(
selected["release_id"],
selected["configuration_id"],
"worker.json",
),
32768,
)["amqp_url"]
if connections["amqp"][role] != expected:
raise DeliveryError("dependency broker credentials differ from selected batch")
files = render(selected["release_id"], selected["configuration_id"])
files["definitions.json"] = canonical(broker_definitions(connections))
if output.exists() or output.is_symlink():
raise DeliveryError("dependency output must be new")
preflight.protected(output.parent, directory=True)
output.mkdir(mode=0o700)
hashes = {}
for name, contents in files.items():
path = output / name
path.write_bytes(contents)
path.chmod(0o600)
hashes[name] = digest(path)
result = {
"schema_version": 1,
"source_sha": original["source_sha"],
"configuration_id": selected["configuration_id"],
"files": hashes,
"required_storage_mounts": STORAGE,
"installed": False,
"storage_limits_verified": False,
"services_started": False,
"execution_verified": False,
}
path = output / "receipt.json"
path.write_bytes(canonical(result))
path.chmod(0o600)
return result
def main() -> None:
parser = argparse.ArgumentParser(description="Bounded local dependency composition")
parser.add_argument("--connections", type=Path)
parser.add_argument("--output", type=Path)
parser.add_argument("--check-service", choices=STORAGE)
args = parser.parse_args()
try:
if args.check_service is not None:
if args.connections is not None or args.output is not None:
raise DeliveryError("unexpected runtime check override")
result = check_runtime(args.check_service)
else:
if args.connections is None or args.output is None:
raise DeliveryError("explicit dependency composition inputs required")
result = generate(args.connections, args.output)
print(json.dumps(result, sort_keys=True))
except (OSError, ValueError) as error:
parser.exit(
1,
"Dependency composition refused: "
+ (str(error) if isinstance(error, DeliveryError) else type(error).__name__)
+ "\n",
)
if __name__ == "__main__":
main()