From 8d7f651ddcf99201b09d75f06eee15e56817640b Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Wed, 28 Oct 2026 12:11:53 +0800 Subject: [PATCH] perf(delivery): use bounded native cold restore copies --- .github/workflows/engineering.yml | 3 + README.md | 4 +- backend/src/kf_oj/contracts/export.py | 3 + backend/src/kf_oj/contracts/restore_io.py | 129 ++++ backend/tests/test_delivery_io.py | 32 + backend/tests/test_restore_io_contracts.py | 23 + .../tests/test_restore_native_transport.py | 115 ++++ contracts/README.md | 11 + .../schemas/restore-io-reply.v1.schema.json | 75 +++ .../schemas/restore-io-request.v1.schema.json | 187 ++++++ judge/crates/contracts/src/lib.rs | 1 + judge/crates/contracts/src/restore_io.rs | 240 +++++++ .../contracts/tests/restore_io_examples.rs | 40 ++ judge/crates/sandbox/Cargo.toml | 6 +- .../crates/sandbox/src/restore_copy/files.rs | 502 ++++++++++++++ .../crates/sandbox/src/restore_copy/guard.rs | 206 ++++++ judge/crates/sandbox/src/restore_copy/main.rs | 401 +++++++++++ scripts/check_architecture.py | 32 +- scripts/release/README.md | 19 + scripts/release/build.py | 2 +- scripts/release/manifest.py | 5 + scripts/release/restore.py | 43 +- scripts/release/restore_native.py | 220 ++++++ tests/fixtures/restore-io/examples.json | 631 ++++++++++++++++++ 24 files changed, 2911 insertions(+), 19 deletions(-) create mode 100644 backend/src/kf_oj/contracts/restore_io.py create mode 100644 backend/tests/test_restore_io_contracts.py create mode 100644 backend/tests/test_restore_native_transport.py create mode 100644 contracts/schemas/restore-io-reply.v1.schema.json create mode 100644 contracts/schemas/restore-io-request.v1.schema.json create mode 100644 judge/crates/contracts/src/restore_io.rs create mode 100644 judge/crates/contracts/tests/restore_io_examples.rs create mode 100644 judge/crates/sandbox/src/restore_copy/files.rs create mode 100644 judge/crates/sandbox/src/restore_copy/guard.rs create mode 100644 judge/crates/sandbox/src/restore_copy/main.rs create mode 100644 scripts/release/restore_native.py create mode 100644 tests/fixtures/restore-io/examples.json diff --git a/.github/workflows/engineering.yml b/.github/workflows/engineering.yml index 585f6ee..ac9921e 100644 --- a/.github/workflows/engineering.yml +++ b/.github/workflows/engineering.yml @@ -111,6 +111,8 @@ jobs: tests/test_delivery_io.py \ tests/test_delivery_backup.py \ tests/test_delivery_restore_io.py \ + tests/test_restore_io_contracts.py \ + tests/test_restore_native_transport.py \ tests/test_operator_imports.py \ tests/test_health.py \ tests/test_identity.py \ @@ -126,6 +128,7 @@ jobs: cargo clippy --workspace --all-targets --locked -- -D warnings cargo test --workspace --lib --locked cargo test --package kf-contracts --locked + cargo test --package kf-sandbox --bin kf-restore-io --locked - name: Check and build Frontend timeout-minutes: 5 diff --git a/README.md b/README.md index 043ea74..7482f5c 100644 --- a/README.md +++ b/README.md @@ -37,12 +37,14 @@ Python 业务域分为 `identity`、`problems`、`submissions`、`community`、` Rust 分为 `kf-contracts`、`kf-sandbox`、`kf-engine`、`kf-worker`、`kf-coordinator` 五个 crate。 协调器拥有执行状态写入权，判题 Worker 不直接写评测数据库。 +Sandbox 内的 `kf-restore-io` 仅在固定离线恢复单元中复制和核验固定数据域， +通过共享有界协议接收文件记录；不启动程序、服务或写业务状态。 沙箱执行入口每次核验隔离与限额，任何未验证配置继续拒绝执行。 用户程序及隔离/超限验收只在独立、整体限额的无业务凭据测试 VM 内运行。 ## 安装与检查 -工具链基线：Python 3.13.16 + uv、Rust 1.99.0、Node 26.10.0 + pnpm 11.28.4。 +工具链基线：Python 3.13.16 + uv、Rust 1.99.0、Node 26.11.1 + pnpm 11.28.4。 Python 和 Rust 版本由工程配置固定，三套工程分别提交依赖锁文件。 开发与构建使用 Linux 原生文件系统。 diff --git a/backend/src/kf_oj/contracts/export.py b/backend/src/kf_oj/contracts/export.py index da08f3b..4284ea5 100644 --- a/backend/src/kf_oj/contracts/export.py +++ b/backend/src/kf_oj/contracts/export.py @@ -20,6 +20,7 @@ from kf_oj.contracts.problem_maintenance import ( MaintenanceReceipt, PublishCandidate, ) +from kf_oj.contracts.restore_io import RestoreIOReply, RestoreIORequest from kf_oj.contracts.submission_performance import SubmissionPerformance from kf_oj.core.settings import Settings @@ -57,6 +58,8 @@ def main() -> None: outputs[root / "contracts/schemas/submission-performance.v1.schema.json"] = performance for name, model in { "delivery-availability": AvailabilityRecord, + "restore-io-request": RestoreIORequest, + "restore-io-reply": RestoreIOReply, "problem-publish": PublishCandidate, "problem-availability": ChangeProblemAvailability, "problem-maintenance": MaintenanceReceipt, diff --git a/backend/src/kf_oj/contracts/restore_io.py b/backend/src/kf_oj/contracts/restore_io.py new file mode 100644 index 0000000..ca202d2 --- /dev/null +++ b/backend/src/kf_oj/contracts/restore_io.py @@ -0,0 +1,129 @@ +"""Private offline restore byte-copy protocol; conveys no execution authority.""" + +import re +from typing import Annotated, Literal + +from pydantic import ( + BaseModel, + BeforeValidator, + ConfigDict, + Field, + RootModel, + StrictInt, + model_validator, +) + +from kf_oj.contracts.judging import integer_version + +MAX_FILES = 98304 +MAX_LOGICAL_BYTES = 12 * 1024**3 +MAX_FILE_BYTES = 256 * 1024**2 +MAX_BATCH_BYTES = 64 * 1024**2 +DOMAINS = frozenset( + "postgres artifacts imports app-config judge-config factory operations releases " + "images dependencies proxy recovery".split() +) +Count = Annotated[StrictInt, Field(ge=0, le=MAX_FILES)] +ByteCount = Annotated[StrictInt, Field(ge=0, le=MAX_LOGICAL_BYTES)] + + +class Record(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) + + +class RestoreFile(Record): + path: Annotated[str, Field(min_length=1, max_length=512)] + uid: Annotated[StrictInt, Field(ge=0, le=1205)] + gid: Annotated[StrictInt, Field(ge=0, le=1205)] + mode: StrictInt + bytes: Annotated[StrictInt, Field(ge=0, le=MAX_FILE_BYTES)] + sha256: Annotated[str, Field(pattern=r"^[0-9a-f]{64}$")] + + @model_validator(mode="after") + def check_member(self) -> "RestoreFile": + parts = self.path.split("/") + if ( + len(parts) < 2 + or parts[0] not in DOMAINS + or any( + p in {"", ".", "..", ".git", ".ssh", ".localplan", "docs"} + or re.fullmatch(r"[A-Za-z0-9_.+@-]{1,255}", p) is None + for p in parts + ) + ): + raise ValueError("restore file must be below a fixed backup domain") + if self.uid not in {0, 1200, 1201, 1202, 1203, 1204, 1205} or self.gid not in ( + {0, 1203, 1204, 1205} if self.uid == 0 else {self.uid} + ): + raise ValueError("unsupported restore identity") + if self.mode not in {0o400, 0o444, 0o600, 0o640, 0o644, 0o555, 0o755}: + raise ValueError("unsupported restore file mode") + if self.sha256 in {"a" * 64, "0" * 64}: + raise ValueError("placeholder digest refused") + return self + + +class Open(Record): + schema_version: Annotated[Literal[1], BeforeValidator(integer_version)] + kind: Literal["restore.io.open"] + backup_id: Annotated[str, Field(pattern=r"^bk-[0-9a-f]{32}$")] + recovery_id: Annotated[str, Field(pattern=r"^rr-[0-9a-f]{32}$")] + expected_files: Annotated[StrictInt, Field(ge=1, le=MAX_FILES)] + expected_logical_bytes: ByteCount + deadline_monotonic_ns: Annotated[StrictInt, Field(ge=1, le=2**63 - 1)] + + +class Batch(Record): + schema_version: Annotated[Literal[1], BeforeValidator(integer_version)] + kind: Literal["restore.io.batch"] + batch_id: Annotated[StrictInt, Field(ge=1, le=MAX_FILES)] + files: Annotated[list[RestoreFile], Field(min_length=1, max_length=256)] + + @model_validator(mode="after") + def check_batch(self) -> "Batch": + if len(self.files) > 1 and sum(f.bytes for f in self.files) > MAX_BATCH_BYTES: + raise ValueError("restore batch byte bound exceeded") + if len({f.path for f in self.files}) != len(self.files): + raise ValueError("duplicate restore batch member") + return self + + +class Finish(Record): + schema_version: Annotated[Literal[1], BeforeValidator(integer_version)] + kind: Literal["restore.io.finish"] + + +class RestoreIORequest(RootModel[Annotated[Open | Batch | Finish, Field(discriminator="kind")]]): + pass + + +class RestoreIOReply(Record): + schema_version: Annotated[Literal[1], BeforeValidator(integer_version)] + kind: Literal["restore.io.reply"] + stage: Literal["opened", "verified", "finished", "refused"] + batch_id: Count + verified_files: Count + verified_logical_bytes: ByteCount + execution_verified: Annotated[ + Literal[False], + BeforeValidator(lambda value: value if type(value) is bool else _bad_boolean()), + ] + error: Literal["none", "protocol", "guard", "deadline", "filesystem", "bytes", "identity"] + + @model_validator(mode="after") + def check_reply(self) -> "RestoreIOReply": + if (self.stage == "refused") != (self.error != "none"): + raise ValueError("restore refusal state differs") + if self.stage == "opened" and ( + self.batch_id or self.verified_files or self.verified_logical_bytes + ): + raise ValueError("restore open cannot claim verified data") + if self.stage in {"verified", "finished"} and not ( + 1 <= self.batch_id <= self.verified_files + ): + raise ValueError("restore completion has no verified batch") + return self + + +def _bad_boolean() -> bool: + raise ValueError("execution verification must be a boolean false") diff --git a/backend/tests/test_delivery_io.py b/backend/tests/test_delivery_io.py index 5eaccf8..1cbdcf0 100644 --- a/backend/tests/test_delivery_io.py +++ b/backend/tests/test_delivery_io.py @@ -1437,3 +1437,35 @@ def test_proxy_previous_snapshot_retains_the_legacy_unit_and_binding_format(tmp_ unit.chmod(0o644) assert proxy_install.known_snapshot(directory) == selected assert "--previous-certificate" not in unit.read_text() + + +@pytest.mark.parametrize("support", ["legacy", "complete", "binary-only", "transport-only"]) +def test_native_restore_capability_requires_a_complete_pair_but_keeps_legacy( + ordinary_bundle, support +): + bundle, _ = ordinary_bundle + data = json.loads((bundle / "manifest.json").read_bytes()) + names = { + "legacy": [], + "complete": ["bin/kf-restore-io", "delivery/scripts/release/restore_native.py"], + "binary-only": ["bin/kf-restore-io"], + "transport-only": ["delivery/scripts/release/restore_native.py"], + }[support] + parents = set(data["directories"]) + for name in names: + data["files"].append( + { + "path": name, + "bytes": 1, + "sha256": "b" * 64, + "mode": "0555" if name.startswith("bin/") else "0444", + } + ) + data["payload_bytes"] += 1 + parents.update(str(p) for p in Path(name).parents if str(p) != ".") + data["directories"] = sorted(parents) + if support in {"legacy", "complete"}: + assert manifest.validate(data) is data + else: + with pytest.raises(policy.DeliveryError, match="incomplete application bundle"): + manifest.validate(data) diff --git a/backend/tests/test_restore_io_contracts.py b/backend/tests/test_restore_io_contracts.py new file mode 100644 index 0000000..f75db60 --- /dev/null +++ b/backend/tests/test_restore_io_contracts.py @@ -0,0 +1,23 @@ +"""Shared ordinary wire examples; no privileged copy or execution acceptance.""" + +import json +from pathlib import Path + +import pytest +from pydantic import ValidationError + +from kf_oj.contracts.restore_io import RestoreIOReply, RestoreIORequest + +CASES = json.loads( + (Path(__file__).resolve().parents[2] / "tests/fixtures/restore-io/examples.json").read_bytes() +) + + +@pytest.mark.parametrize("case", CASES, ids=lambda c: c["name"]) +def test_shared_restore_io_examples(case): + model = RestoreIORequest if case["direction"] == "request" else RestoreIOReply + if case["valid"]: + assert model.model_validate(case["value"]).model_dump(mode="json") == case["value"] + else: + with pytest.raises(ValidationError): + model.model_validate(case["value"]) diff --git a/backend/tests/test_restore_native_transport.py b/backend/tests/test_restore_native_transport.py new file mode 100644 index 0000000..f77e0ca --- /dev/null +++ b/backend/tests/test_restore_native_transport.py @@ -0,0 +1,115 @@ +"""Ordinary pipe fixtures only; no Root, filesystem-copy or VM acceptance.""" + +import json +import subprocess +import sys +import time + +import pytest +from scripts.release import restore_native +from scripts.release.policy import DeliveryError + + +@pytest.fixture +def transport(tmp_path, monkeypatch): + helper = tmp_path / "bin/kf-restore-io" + helper.parent.mkdir() + helper.write_bytes(b"ordinary metadata fixture, never executed\n") + helper.chmod(0o555) + monkeypatch.setattr(restore_native.preflight, "protected", lambda *_a, **_kw: None) + original = subprocess.Popen + children = [] + + def start(program): + def popen(args, **options): + assert args == [str(helper), "--stdio-v1"] + assert options["start_new_session"] and options["close_fds"] + assert set(options["env"]) == {"PATH", "LC_ALL", "LD_LIBRARY_PATH"} + child = original([sys.executable, "-c", program], **options) + children.append(child) + return child + + monkeypatch.setattr(restore_native.subprocess, "Popen", popen) + + yield tmp_path, start, children + assert all(p.poll() is not None for p in children) + assert all(h.closed for p in children for h in (p.stdin, p.stdout, p.stderr)) + + +def test_transport_counts_and_child_join_without_execution_claim(transport): + payload, start, children = transport + start(""" +import json,sys +count=total=batch=0 +for line in sys.stdin: + value=json.loads(line) + kind=value['kind'] + stage='opened' + if kind=='restore.io.batch': + count+=len(value['files']) + total+=sum(r['bytes'] for r in value['files']) + batch=value['batch_id'] + stage='verified' + if kind=='restore.io.finish':stage='finished' + reply={'schema_version':1,'kind':'restore.io.reply','stage':stage,'batch_id':batch,'verified_files':count,'verified_logical_bytes':total,'execution_verified':False,'error':'none'} + print(json.dumps(reply),flush=True) + if stage=='finished':break +""") + session = restore_native.Session( + payload, "bk-" + "1" * 32, "rr-" + "2" * 32, 2, 7, time.monotonic() + 10 + ) + try: + session.materialize( + [ + {"path": "artifacts/a", "bytes": 3, "sha256": "1" * 64}, + {"path": "artifacts/b", "bytes": 4, "sha256": "2" * 64}, + ] + ) + session.finish() + assert session.finished and children[0].returncode == 0 + finally: + session.close() + + +@pytest.mark.parametrize( + "response", + [ + b"", + b"not-json\n", + b"{}\n{}\n", + b"x" * (64 * 1024 + 1), + json.dumps( + { + "schema_version": 1, + "kind": "restore.io.reply", + "stage": "opened", + "batch_id": 0, + "verified_files": 0, + "verified_logical_bytes": 0, + "execution_verified": True, + "error": "none", + } + ).encode() + + b"\n", + ], +) +def test_transport_refuses_eof_malformed_unbounded_or_authority_reply(transport, response): + payload, start, _ = transport + start( + "import sys;sys.stdin.readline();sys.stdout.buffer.write(" + + repr(response) + + ");sys.stdout.flush()" + ) + with pytest.raises(DeliveryError): + restore_native.Session( + payload, "bk-" + "1" * 32, "rr-" + "2" * 32, 1, 0, time.monotonic() + 10 + ) + + +def test_transport_deadline_reaps_its_owned_waiting_child(transport): + payload, start, _ = transport + start("import sys,time;sys.stdin.readline();time.sleep(10)") + then = time.monotonic() + with pytest.raises(DeliveryError): + restore_native.Session(payload, "bk-" + "1" * 32, "rr-" + "2" * 32, 1, 0, then + 0.2) + assert time.monotonic() - then < 5 diff --git a/contracts/README.md b/contracts/README.md index e6074a3..d9c8313 100644 --- a/contracts/README.md +++ b/contracts/README.md @@ -38,3 +38,14 @@ Coordinator 必须在事务锁内核对当前租约及状态规则，并将结 Python/Rust 的 `manifest_bytes` 输出同一 UTF-8 JSON：递归排序对象键，无额外空白/换行。 清单摘要是这些确切字节的 SHA-256，不能使用数据库 JSONB 的输出顺序或任意 prettify 后的字节。 共享 `canonical-manifest.v1.txt` 为合成示例的字节验收，平台内部响应复用这些序列化字节。 + +## 离线恢复字节复制协议 + +`restore-io-request/reply` v1 用于 Python 恢复操作器与 Sandbox 的 `kf-restore-io`。 +共享 37 项合法/非法元数据样例在 `tests/fixtures/restore-io/examples.json`。 +请求仅含固定域内规范文件名、既有 OS 身份/模式、字节数和摘要；不接受源/目标宿主路径、 +命令或服务操作。每行请求最多 256 KiB、回复 64 KiB；重复/未知字段、严格整数违规及 +总量/批次/状态不一致均拒绝。单批至多 256 文件、64 MiB，较大单文件独占批次且不超过 +256 MiB；整个已验证清单至多 98304 文件、12 GiB。接收端仍独立核对固定 Root 单元、 +真实限额、只读备份卷与停止的写入者。`execution_verified` 始终为 false；复制回复 +不能证明业务恢复、正常就绪、判题或恢复时间目标。Schema 描述结构，跨字段规则需模型校验。 diff --git a/contracts/schemas/restore-io-reply.v1.schema.json b/contracts/schemas/restore-io-reply.v1.schema.json new file mode 100644 index 0000000..889cafd --- /dev/null +++ b/contracts/schemas/restore-io-reply.v1.schema.json @@ -0,0 +1,75 @@ +{ + "$id": "urn:kf-oj:restore-io-reply:v1", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "additionalProperties": false, + "properties": { + "batch_id": { + "maximum": 98304, + "minimum": 0, + "title": "Batch Id", + "type": "integer" + }, + "error": { + "enum": [ + "none", + "protocol", + "guard", + "deadline", + "filesystem", + "bytes", + "identity" + ], + "title": "Error", + "type": "string" + }, + "execution_verified": { + "const": false, + "title": "Execution Verified", + "type": "boolean" + }, + "kind": { + "const": "restore.io.reply", + "title": "Kind", + "type": "string" + }, + "schema_version": { + "const": 1, + "title": "Schema Version", + "type": "integer" + }, + "stage": { + "enum": [ + "opened", + "verified", + "finished", + "refused" + ], + "title": "Stage", + "type": "string" + }, + "verified_files": { + "maximum": 98304, + "minimum": 0, + "title": "Verified Files", + "type": "integer" + }, + "verified_logical_bytes": { + "maximum": 12884901888, + "minimum": 0, + "title": "Verified Logical Bytes", + "type": "integer" + } + }, + "required": [ + "schema_version", + "kind", + "stage", + "batch_id", + "verified_files", + "verified_logical_bytes", + "execution_verified", + "error" + ], + "title": "RestoreIOReply", + "type": "object" +} diff --git a/contracts/schemas/restore-io-request.v1.schema.json b/contracts/schemas/restore-io-request.v1.schema.json new file mode 100644 index 0000000..37aa128 --- /dev/null +++ b/contracts/schemas/restore-io-request.v1.schema.json @@ -0,0 +1,187 @@ +{ + "$defs": { + "Batch": { + "additionalProperties": false, + "properties": { + "batch_id": { + "maximum": 98304, + "minimum": 1, + "title": "Batch Id", + "type": "integer" + }, + "files": { + "items": { + "$ref": "#/$defs/RestoreFile" + }, + "maxItems": 256, + "minItems": 1, + "title": "Files", + "type": "array" + }, + "kind": { + "const": "restore.io.batch", + "title": "Kind", + "type": "string" + }, + "schema_version": { + "const": 1, + "title": "Schema Version", + "type": "integer" + } + }, + "required": [ + "schema_version", + "kind", + "batch_id", + "files" + ], + "title": "Batch", + "type": "object" + }, + "Finish": { + "additionalProperties": false, + "properties": { + "kind": { + "const": "restore.io.finish", + "title": "Kind", + "type": "string" + }, + "schema_version": { + "const": 1, + "title": "Schema Version", + "type": "integer" + } + }, + "required": [ + "schema_version", + "kind" + ], + "title": "Finish", + "type": "object" + }, + "Open": { + "additionalProperties": false, + "properties": { + "backup_id": { + "pattern": "^bk-[0-9a-f]{32}$", + "title": "Backup Id", + "type": "string" + }, + "deadline_monotonic_ns": { + "maximum": 9223372036854775807, + "minimum": 1, + "title": "Deadline Monotonic Ns", + "type": "integer" + }, + "expected_files": { + "maximum": 98304, + "minimum": 1, + "title": "Expected Files", + "type": "integer" + }, + "expected_logical_bytes": { + "maximum": 12884901888, + "minimum": 0, + "title": "Expected Logical Bytes", + "type": "integer" + }, + "kind": { + "const": "restore.io.open", + "title": "Kind", + "type": "string" + }, + "recovery_id": { + "pattern": "^rr-[0-9a-f]{32}$", + "title": "Recovery Id", + "type": "string" + }, + "schema_version": { + "const": 1, + "title": "Schema Version", + "type": "integer" + } + }, + "required": [ + "schema_version", + "kind", + "backup_id", + "recovery_id", + "expected_files", + "expected_logical_bytes", + "deadline_monotonic_ns" + ], + "title": "Open", + "type": "object" + }, + "RestoreFile": { + "additionalProperties": false, + "properties": { + "bytes": { + "maximum": 268435456, + "minimum": 0, + "title": "Bytes", + "type": "integer" + }, + "gid": { + "maximum": 1205, + "minimum": 0, + "title": "Gid", + "type": "integer" + }, + "mode": { + "title": "Mode", + "type": "integer" + }, + "path": { + "maxLength": 512, + "minLength": 1, + "title": "Path", + "type": "string" + }, + "sha256": { + "pattern": "^[0-9a-f]{64}$", + "title": "Sha256", + "type": "string" + }, + "uid": { + "maximum": 1205, + "minimum": 0, + "title": "Uid", + "type": "integer" + } + }, + "required": [ + "path", + "uid", + "gid", + "mode", + "bytes", + "sha256" + ], + "title": "RestoreFile", + "type": "object" + } + }, + "$id": "urn:kf-oj:restore-io-request:v1", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "discriminator": { + "mapping": { + "restore.io.batch": "#/$defs/Batch", + "restore.io.finish": "#/$defs/Finish", + "restore.io.open": "#/$defs/Open" + }, + "propertyName": "kind" + }, + "oneOf": [ + { + "$ref": "#/$defs/Open" + }, + { + "$ref": "#/$defs/Batch" + }, + { + "$ref": "#/$defs/Finish" + } + ], + "title": "RestoreIORequest" +} diff --git a/judge/crates/contracts/src/lib.rs b/judge/crates/contracts/src/lib.rs index b68d60b..47e10dd 100644 --- a/judge/crates/contracts/src/lib.rs +++ b/judge/crates/contracts/src/lib.rs @@ -6,6 +6,7 @@ use uuid::Uuid; pub mod judging; pub mod judging_v2; pub mod judging_versions; +pub mod restore_io; pub mod state; pub const SCHEMA_VERSION: u16 = 1; diff --git a/judge/crates/contracts/src/restore_io.rs b/judge/crates/contracts/src/restore_io.rs new file mode 100644 index 0000000..c75d4ee --- /dev/null +++ b/judge/crates/contracts/src/restore_io.rs @@ -0,0 +1,240 @@ +//! Private offline restore protocol. No OS operations or execution authority. + +use crate::SchemaVersion; +use serde::{Deserialize, Serialize}; +use std::collections::HashSet; + +pub const MAX_FILES: usize = 98304; +pub const MAX_LOGICAL_BYTES: u64 = 12 * 1024 * 1024 * 1024; +pub const MAX_FILE_BYTES: u64 = 256 * 1024 * 1024; +pub const MAX_BATCH_BYTES: u64 = 64 * 1024 * 1024; +pub const MAX_REQUEST_BYTES: usize = 256 * 1024; +pub const MAX_REPLY_BYTES: usize = 64 * 1024; + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RestoreFile { + pub path: String, + pub uid: u32, + pub gid: u32, + pub mode: u32, + pub bytes: u64, + pub sha256: String, +} + +impl RestoreFile { + pub fn validate(&self) -> Result<(), &'static str> { + let mut parts = self.path.split('/'); + if self.path.len() > 512 + || !matches!( + parts.next(), + Some( + "postgres" + | "artifacts" + | "imports" + | "app-config" + | "judge-config" + | "factory" + | "operations" + | "releases" + | "images" + | "dependencies" + | "proxy" + | "recovery" + ) + ) + { + return Err("restore file domain differs"); + } + let mut count = 0; + for part in parts { + count += 1; + if part.is_empty() + || part.len() > 255 + || matches!(part, "." | ".." | ".git" | ".ssh" | ".localplan" | "docs") + || !part + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b"_.+@-".contains(&b)) + { + return Err("restore path component differs"); + } + } + if count == 0 + || !matches!(self.uid, 0 | 1200..=1205) + || !(if self.uid == 0 { + matches!(self.gid, 0 | 1203..=1205) + } else { + self.gid == self.uid + }) + || !matches!( + self.mode, + 0o400 | 0o444 | 0o600 | 0o640 | 0o644 | 0o555 | 0o755 + ) + || self.bytes > MAX_FILE_BYTES + || self.sha256.len() != 64 + || !self + .sha256 + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) + || self.sha256 == "0".repeat(64) + || self.sha256 == "a".repeat(64) + { + return Err("restore member metadata differs"); + } + Ok(()) + } +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(tag = "kind", deny_unknown_fields)] +pub enum Request { + #[serde(rename = "restore.io.open")] + Open { + schema_version: SchemaVersion, + backup_id: String, + recovery_id: String, + expected_files: u32, + expected_logical_bytes: u64, + deadline_monotonic_ns: u64, + }, + #[serde(rename = "restore.io.batch")] + Batch { + schema_version: SchemaVersion, + batch_id: u32, + files: Vec, + }, + #[serde(rename = "restore.io.finish")] + Finish { schema_version: SchemaVersion }, +} + +pub fn identifier(value: &str, prefix: &str) -> bool { + value.len() == prefix.len() + 32 + && value.starts_with(prefix) + && value[prefix.len()..] + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) +} + +impl Request { + pub fn validate(&self) -> Result<(), &'static str> { + match self { + Self::Open { + backup_id, + recovery_id, + expected_files, + expected_logical_bytes, + deadline_monotonic_ns, + .. + } => { + if !identifier(backup_id, "bk-") + || !identifier(recovery_id, "rr-") + || !(1..=MAX_FILES as u32).contains(expected_files) + || *expected_logical_bytes > MAX_LOGICAL_BYTES + || !(1..=i64::MAX as u64).contains(deadline_monotonic_ns) + { + return Err("restore open bounds differ"); + } + } + Self::Batch { + batch_id, files, .. + } => { + if !(1..=MAX_FILES as u32).contains(batch_id) + || files.is_empty() + || files.len() > 256 + { + return Err("restore batch count differs"); + } + let mut seen = HashSet::new(); + for file in files { + file.validate()?; + if !seen.insert(&file.path) { + return Err("duplicate restore file"); + } + } + if files.len() > 1 && files.iter().map(|f| f.bytes).sum::() > MAX_BATCH_BYTES { + return Err("restore batch byte bound exceeded"); + } + } + Self::Finish { .. } => {} + } + Ok(()) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum Stage { + Opened, + Verified, + Finished, + Refused, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum Error { + None, + Protocol, + Guard, + Deadline, + Filesystem, + Bytes, + Identity, +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Reply { + pub schema_version: SchemaVersion, + pub kind: ReplyKind, + pub stage: Stage, + pub batch_id: u32, + pub verified_files: u32, + pub verified_logical_bytes: u64, + pub execution_verified: bool, + pub error: Error, +} + +#[derive(Debug, Serialize, Deserialize)] +pub enum ReplyKind { + #[serde(rename = "restore.io.reply")] + RestoreIO, +} + +impl Reply { + pub fn validate(&self) -> Result<(), &'static str> { + if self.execution_verified + || self.batch_id > MAX_FILES as u32 + || self.verified_files > MAX_FILES as u32 + || self.verified_logical_bytes > MAX_LOGICAL_BYTES + || (self.stage == Stage::Refused) != (self.error != Error::None) + || (self.stage == Stage::Opened + && (self.batch_id != 0 + || self.verified_files != 0 + || self.verified_logical_bytes != 0)) + || (matches!(self.stage, Stage::Verified | Stage::Finished) + && !(1..=self.verified_files).contains(&self.batch_id)) + { + return Err("restore reply state differs"); + } + Ok(()) + } +} + +pub fn decode_request(raw: &[u8]) -> Result { + if raw.len() > MAX_REQUEST_BYTES { + return Err("restore request too large"); + } + let value: Request = serde_json::from_slice(raw).map_err(|_| "invalid restore request")?; + value.validate()?; + Ok(value) +} + +pub fn decode_reply(raw: &[u8]) -> Result { + if raw.len() > MAX_REPLY_BYTES { + return Err("restore reply too large"); + } + let value: Reply = serde_json::from_slice(raw).map_err(|_| "invalid restore reply")?; + value.validate()?; + Ok(value) +} diff --git a/judge/crates/contracts/tests/restore_io_examples.rs b/judge/crates/contracts/tests/restore_io_examples.rs new file mode 100644 index 0000000..4fcfb82 --- /dev/null +++ b/judge/crates/contracts/tests/restore_io_examples.rs @@ -0,0 +1,40 @@ +use kf_contracts::restore_io::{decode_reply, decode_request}; +use serde::Deserialize; + +#[derive(Deserialize)] +struct Case { + name: String, + direction: String, + valid: bool, + value: serde_json::Value, +} + +#[test] +fn shared_restore_io_examples() { + let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../../../tests/fixtures/restore-io/examples.json"); + let cases: Vec = serde_json::from_slice(&std::fs::read(path).unwrap()).unwrap(); + assert_eq!(cases.len(), 37); + for case in cases { + let raw = serde_json::to_vec(&case.value).unwrap(); + let result = match case.direction.as_str() { + "request" => decode_request(&raw).map(|v| serde_json::to_value(v).unwrap()), + "reply" => decode_reply(&raw).map(|v| serde_json::to_value(v).unwrap()), + _ => panic!("unknown fixture direction"), + }; + assert_eq!(result.is_ok(), case.valid, "{}", case.name); + if let Ok(actual) = result { + assert_eq!(actual, case.value, "{}", case.name); + } + } +} + +#[test] +fn duplicate_fields_and_oversized_wire_are_refused() { + assert!( + decode_request(br#"{"kind":"restore.io.finish","schema_version":1,"schema_version":1}"#) + .is_err() + ); + assert!(decode_request(&vec![b' '; 256 * 1024 + 1]).is_err()); + assert!(decode_reply(&vec![b' '; 64 * 1024 + 1]).is_err()); +} diff --git a/judge/crates/sandbox/Cargo.toml b/judge/crates/sandbox/Cargo.toml index 249ba1f..6417fd5 100644 --- a/judge/crates/sandbox/Cargo.toml +++ b/judge/crates/sandbox/Cargo.toml @@ -16,7 +16,7 @@ clap.workspace = true serde.workspace = true serde_json.workspace = true sha2 = "=0.10.9" -rustix = { version = "1.1.5", features = ["net", "process", "fs", "thread", "mount", "system", "stdio", "pipe", "event"] } +rustix = { version = "1.1.5", features = ["net", "process", "fs", "thread", "mount", "system", "stdio", "pipe", "event", "time"] } # Fixed BPF compilation/installation through safe APIs; libc supplies syscall # numbers and the audited fork/exec primitive. Both dependencies stay inside @@ -30,3 +30,7 @@ workspace = true [[example]] name = "identity_probe" test = false + +[[bin]] +name = "kf-restore-io" +path = "src/restore_copy/main.rs" diff --git a/judge/crates/sandbox/src/restore_copy/files.rs b/judge/crates/sandbox/src/restore_copy/files.rs new file mode 100644 index 0000000..908b0f4 --- /dev/null +++ b/judge/crates/sandbox/src/restore_copy/files.rs @@ -0,0 +1,502 @@ +//! Anchored regular-file copies, filesystem error cursors and complete byte readback. + +use super::{Error, Result, remaining}; +use kf_contracts::restore_io::RestoreFile; +use rustix::{ + fs::{AtFlags, Mode, OFlags, ResolveFlags}, + process::{Gid, Uid}, +}; +use sha2::{Digest, Sha256}; +use std::{ + collections::HashMap, + fs::{File, Metadata}, + io::{Read, Seek, Write}, + os::unix::fs::MetadataExt, + path::{Path, PathBuf}, +}; + +const CHUNK: usize = 1024 * 1024; +const RESOLVE: ResolveFlags = ResolveFlags::BENEATH + .union(ResolveFlags::NO_SYMLINKS) + .union(ResolveFlags::NO_MAGICLINKS); +pub type Stamp = [u64; 9]; + +fn stamp(info: &Metadata) -> Result { + let ns = |s: i64, n: i64| { + u64::try_from(i128::from(s) * 1_000_000_000 + i128::from(n)).map_err(|_| Error::Identity) + }; + Ok([ + info.dev(), + info.ino(), + info.mode().into(), + info.nlink(), + info.uid().into(), + info.gid().into(), + info.size(), + ns(info.mtime(), info.mtime_nsec())?, + ns(info.ctime(), info.ctime_nsec())?, + ]) +} + +struct Anchor { + path: PathBuf, + file: File, + device: u64, + inode: u64, +} +impl Anchor { + fn open(path: PathBuf) -> Result { + let file = File::from( + rustix::fs::openat2( + rustix::fs::CWD, + &path, + OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC, + Mode::empty(), + ResolveFlags::NO_SYMLINKS | ResolveFlags::NO_MAGICLINKS, + ) + .map_err(|_| Error::Filesystem)?, + ); + let info = file.metadata().map_err(|_| Error::Filesystem)?; + Ok(Self { + path, + file, + device: info.dev(), + inode: info.ino(), + }) + } + fn unchanged(&self) -> Result<()> { + let fresh = Self::open(self.path.clone())?; + if (self.device, self.inode) != (fresh.device, fresh.inode) { + return Err(Error::Identity); + } + Ok(()) + } + fn file(&self, relative: &str, flags: OFlags) -> Result { + rustix::fs::openat2( + &self.file, + relative, + flags | OFlags::CLOEXEC | OFlags::NOFOLLOW, + if flags.contains(OFlags::CREATE) { + Mode::from_raw_mode(0o600) + } else { + Mode::empty() + }, + RESOLVE, + ) + .map(File::from) + .map_err(|_| Error::Filesystem) + } + fn observed(&self, relative: &str) -> Result { + // Parents were created/checked by the locked Root operator on the + // stopped node. The actual byte opens additionally use openat2 with + // BENEATH/NO_SYMLINKS. Final metadata checks include the leaf's type. + let s = rustix::fs::statat(&self.file, relative, AtFlags::SYMLINK_NOFOLLOW) + .map_err(|_| Error::Filesystem)?; + let ns = |sec: i64, nano: i64| { + u64::try_from(i128::from(sec) * 1_000_000_000 + i128::from(nano)) + .map_err(|_| Error::Identity) + }; + Ok([ + s.st_dev, + s.st_ino, + s.st_mode.into(), + s.st_nlink, + s.st_uid.into(), + s.st_gid.into(), + u64::try_from(s.st_size).map_err(|_| Error::Identity)?, + ns(s.st_mtime, s.st_mtime_nsec as i64)?, + ns(s.st_ctime, s.st_ctime_nsec as i64)?, + ]) + } +} + +pub struct Roots { + source: Anchor, + targets: HashMap<&'static str, Anchor>, + source_owner: (u32, u32), + #[cfg(test)] + ordinary_fixture: bool, +} +impl Roots { + pub fn production(backup_id: &str, recovery_id: &str) -> Result { + let source = Path::new("/var/lib/kf-onlinejudge/backups") + .join(backup_id) + .join("objects"); + let targets = [ + ( + "postgres", + PathBuf::from("/var/lib/kf-onlinejudge/postgres/data"), + ), + ( + "artifacts", + PathBuf::from("/var/lib/kf-onlinejudge/artifacts"), + ), + ("imports", PathBuf::from("/var/lib/kf-onlinejudge/imports")), + ("factory", PathBuf::from("/var/lib/kf-judge/factory")), + ("releases", PathBuf::from("/opt/kf-onlinejudge/releases")), + ("images", PathBuf::from("/var/lib/kf-judge/images")), + ( + "history", + Path::new("/var/lib/kf-onlinejudge/recovery") + .join(recovery_id) + .join("history"), + ), + ]; + let snapshot = Anchor::open(source.parent().ok_or(Error::Guard)?.to_owned())?; + let roots = Self::open(source, targets)?; + for anchor in [&snapshot, &roots.source] { + let info = anchor.file.metadata().map_err(|_| Error::Guard)?; + if (info.uid(), info.gid(), info.mode() & 0o7777) != (0, 0, 0o700) { + return Err(Error::Guard); + } + } + Ok(roots) + } + fn open( + source: PathBuf, + targets: [(&'static str, PathBuf); N], + ) -> Result { + Ok(Self { + source: Anchor::open(source)?, + targets: targets + .into_iter() + .map(|(key, path)| Ok((key, Anchor::open(path)?))) + .collect::>()?, + source_owner: (0, 0), + #[cfg(test)] + ordinary_fixture: false, + }) + } + fn target<'a>(&'a self, row: &'a RestoreFile) -> Result<(&'a Anchor, &'a str, u32, u32, u32)> { + let (domain, suffix) = row.path.split_once('/').ok_or(Error::Protocol)?; + match self.targets.get(domain) { + Some(anchor) => Ok((anchor, suffix, row.uid, row.gid, row.mode)), + None => Ok(( + self.targets.get("history").ok_or(Error::Protocol)?, + &row.path, + 0, + 0, + 0o600, + )), + } + } + pub fn sync(&self, deadline: u64) -> Result<()> { + let mut seen = Vec::new(); + for anchor in self.targets.values() { + remaining(deadline)?; + anchor.unchanged()?; + if !seen.contains(&anchor.device) { + #[cfg(not(test))] + rustix::fs::syncfs(&anchor.file).map_err(|_| Error::Filesystem)?; + // Ordinary tests may fsync only their own tiny files/directories, + // never syncfs the host. Real error-cursor validation is VM-only. + #[cfg(test)] + { + assert!(self.ordinary_fixture); + for entry in std::fs::read_dir(&anchor.path).map_err(|_| Error::Filesystem)? { + let entry = entry.map_err(|_| Error::Filesystem)?; + File::open(entry.path()) + .and_then(|f| f.sync_all()) + .map_err(|_| Error::Filesystem)?; + } + anchor.file.sync_all().map_err(|_| Error::Filesystem)?; + } + seen.push(anchor.device); + } + remaining(deadline)?; + } + Ok(()) + } + pub fn unchanged(&self) -> Result<()> { + self.source.unchanged()?; + for anchor in self.targets.values() { + anchor.unchanged()?; + } + Ok(()) + } + pub fn copy(&self, row: RestoreFile, deadline: u64) -> Result { + remaining(deadline)?; + let (anchor, name, uid, gid, mode) = self.target(&row)?; + match rustix::fs::openat2( + &anchor.file, + name, + OFlags::PATH | OFlags::CLOEXEC | OFlags::NOFOLLOW, + Mode::empty(), + RESOLVE, + ) { + Ok(fd) => { + let info = File::from(fd).metadata().map_err(|_| Error::Filesystem)?; + member(&info, &row, uid, gid, mode)?; + return Ok(Pending { + witness: stamp(&info)?, + row, + }); + } + Err(rustix::io::Errno::NOENT) => {} + Err(_) => return Err(Error::Filesystem), + } + let mut input = self + .source + .file(&row.sha256, OFlags::RDONLY | OFlags::NONBLOCK)?; + let before = input.metadata().map_err(|_| Error::Filesystem)?; + member( + &before, + &row, + self.source_owner.0, + self.source_owner.1, + 0o600, + )?; + let original = stamp(&before)?; + if self.source.observed(&row.sha256)? != original { + return Err(Error::Identity); + } + let mut output = anchor.file(name, OFlags::WRONLY | OFlags::CREATE | OFlags::EXCL)?; + let empty = output.metadata().map_err(|_| Error::Filesystem)?; + if !empty.is_file() + || empty.nlink() != 1 + || empty.size() != 0 + || input.stream_position().map_err(|_| Error::Filesystem)? != 0 + || output.stream_position().map_err(|_| Error::Filesystem)? != 0 + { + return Err(Error::Identity); + } + transfer(&mut input, &mut output, row.bytes, deadline)?; + if stamp(&input.metadata().map_err(|_| Error::Filesystem)?)? != original + || self.source.observed(&row.sha256)? != original + { + return Err(Error::Identity); + } + let current = output.metadata().map_err(|_| Error::Filesystem)?; + let changed = (current.uid(), current.gid()) != (uid, gid); + if changed { + rustix::fs::fchown(&output, Some(Uid::from_raw(uid)), Some(Gid::from_raw(gid))) + .map_err(|_| Error::Filesystem)?; + } + let mode_changed = changed || current.mode() & 0o7777 != mode; + if mode_changed { + rustix::fs::fchmod(&output, Mode::from_raw_mode(mode)) + .map_err(|_| Error::Filesystem)?; + } + let final_info = if mode_changed { + output.metadata().map_err(|_| Error::Filesystem)? + } else { + current + }; + member(&final_info, &row, uid, gid, mode)?; + let witness = stamp(&final_info)?; + drop(output); + if anchor.observed(name)? != witness { + return Err(Error::Identity); + } + Ok(Pending { row, witness }) + } + pub fn read_group(&self, group: Vec, deadline: u64) -> Result> { + let mut previous = None; + let mut reference = None; + let mut buffer = vec![0; CHUNK]; + let mut counts = HashMap::new(); + for pending in &group { + *counts + .entry((&pending.row.sha256, pending.row.bytes)) + .or_insert(0) += 1; + } + // Keep one immutable, actually hashed <=8MiB reference per reader. Every + // other destination is still read completely and compared byte for byte. + let repeated: Vec<_> = group + .iter() + .map(|p| counts[&(&p.row.sha256, p.row.bytes)] > 1) + .collect(); + drop(counts); + for (pending, repeated) in group.iter().zip(repeated) { + remaining(deadline)?; + let row = &pending.row; + let key = (&row.sha256, row.bytes); + if previous != Some(key) { + reference = None; + previous = Some(key); + } + let small = repeated && row.bytes <= 8 * 1024 * 1024; + let retain = small && reference.is_none(); + let (anchor, name, uid, gid, mode) = self.target(row)?; + if anchor.observed(name)? != pending.witness { + return Err(Error::Identity); + } + let mut input = anchor.file(name, OFlags::RDONLY | OFlags::NONBLOCK)?; + let info = input.metadata().map_err(|_| Error::Filesystem)?; + member(&info, row, uid, gid, mode)?; + if stamp(&info)? != pending.witness { + return Err(Error::Identity); + } + let retained = verify( + &mut input, + row, + deadline, + if small { reference.as_deref() } else { None }, + retain, + &mut buffer, + )?; + if stamp(&input.metadata().map_err(|_| Error::Filesystem)?)? != pending.witness + || anchor.observed(name)? != pending.witness + { + return Err(Error::Identity); + } + if let Some(data) = retained { + reference = Some(data); + } + } + remaining(deadline)?; + Ok(group) + } + pub fn final_file(&self, pending: &Pending) -> Result<()> { + let (anchor, name, _, _, _) = self.target(&pending.row)?; + if anchor.observed(name)? != pending.witness { + return Err(Error::Identity); + } + Ok(()) + } +} + +pub struct Pending { + pub row: RestoreFile, + pub witness: Stamp, +} + +fn member(info: &Metadata, row: &RestoreFile, uid: u32, gid: u32, mode: u32) -> Result<()> { + if !info.is_file() + || info.nlink() != 1 + || info.size() != row.bytes + || (info.uid(), info.gid(), info.mode() & 0o7777) != (uid, gid, mode) + { + return Err(Error::Identity); + } + Ok(()) +} + +fn transfer(input: &mut File, output: &mut File, bytes: u64, deadline: u64) -> Result<()> { + let mut total = 0; + let mut kernel = true; + let mut buffer = Vec::new(); + while total < bytes { + remaining(deadline)?; + let maximum = (bytes - total).min(CHUNK as u64) as usize; + let count = if kernel { + match rustix::fs::sendfile(&*output, &*input, None, maximum) { + Ok(n) => n, + Err(rustix::io::Errno::INTR) => continue, + Err( + rustix::io::Errno::NOSYS + | rustix::io::Errno::INVAL + | rustix::io::Errno::OPNOTSUPP + | rustix::io::Errno::XDEV, + ) => { + kernel = false; + continue; + } + Err(_) => return Err(Error::Filesystem), + } + } else { + if buffer.is_empty() { + buffer.resize(CHUNK, 0); + } + let count = input + .read(&mut buffer[..maximum]) + .map_err(|_| Error::Filesystem)?; + output + .write_all(&buffer[..count]) + .map_err(|_| Error::Filesystem)?; + count + }; + if count == 0 || count > maximum { + return Err(Error::Bytes); + } + total += count as u64; + } + remaining(deadline)?; + if input.read(&mut [0; 1]).map_err(|_| Error::Filesystem)? != 0 { + return Err(Error::Bytes); + } + Ok(()) +} + +fn verify( + input: &mut File, + row: &RestoreFile, + deadline: u64, + reference: Option<&[u8]>, + retain: bool, + buffer: &mut [u8], +) -> Result>> { + if reference.is_some_and(|r| r.len() as u64 != row.bytes) { + return Err(Error::Bytes); + } + let mut hasher = reference.is_none().then(Sha256::new); + let mut retained = retain.then(|| Vec::with_capacity(row.bytes as usize)); + let mut total = 0; + loop { + remaining(deadline)?; + let count = input.read(buffer).map_err(|_| Error::Filesystem)?; + if count == 0 { + break; + } + let end = total + count; + if end as u64 > row.bytes { + return Err(Error::Bytes); + } + if reference.is_some_and(|r| r[total..end] != buffer[..count]) { + return Err(Error::Bytes); + } + if let Some(hash) = &mut hasher { + hash.update(&buffer[..count]); + } + if let Some(data) = &mut retained { + data.extend_from_slice(&buffer[..count]); + } + total = end; + } + if total as u64 != row.bytes + || hasher.is_some_and(|h| format!("{:x}", h.finalize()) != row.sha256) + { + return Err(Error::Bytes); + } + remaining(deadline)?; + Ok(retained) +} + +#[cfg(test)] +pub struct Fixture(PathBuf); +#[cfg(test)] +impl Fixture { + pub fn new() -> Self { + use std::sync::atomic::{AtomicU64, Ordering}; + static NEXT: AtomicU64 = AtomicU64::new(0); + let root = std::env::temp_dir().join(format!( + "kf-restore-trusted-{}-{}", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + std::fs::create_dir(&root).unwrap(); + std::fs::create_dir(root.join("source")).unwrap(); + std::fs::create_dir(root.join("target")).unwrap(); + Self(root) + } + pub fn source(&self) -> PathBuf { + self.0.join("source") + } + pub fn target(&self) -> PathBuf { + self.0.join("target") + } + pub fn roots(&self) -> Roots { + let mut roots = Roots::open(self.source(), [("artifacts", self.target())]) + .unwrap_or_else(|_| panic!("trusted test root")); + let info = std::fs::metadata(self.source()).unwrap(); + roots.source_owner = (info.uid(), info.gid()); + roots.ordinary_fixture = true; + roots + } +} +#[cfg(test)] +impl Drop for Fixture { + fn drop(&mut self) { + std::fs::remove_dir_all(&self.0).unwrap(); + } +} diff --git a/judge/crates/sandbox/src/restore_copy/guard.rs b/judge/crates/sandbox/src/restore_copy/guard.rs new file mode 100644 index 0000000..a033fef --- /dev/null +++ b/judge/crates/sandbox/src/restore_copy/guard.rs @@ -0,0 +1,206 @@ +//! Fixed production identity, stopped-writer and storage guards. No subprocesses. + +use super::{Error, Result}; +use rustix::process::{Resource, getrlimit}; +use std::{collections::HashMap, fs, io::Read, os::unix::fs::MetadataExt, path::Path}; + +const CGROUP: &str = "/sys/fs/cgroup/kfonlinejudge.slice"; +const BACKUPS: &str = "/var/lib/kf-onlinejudge/backups"; + +pub fn text(path: impl AsRef, bound: usize) -> Result { + let mut data = Vec::new(); + fs::File::open(path) + .map_err(|_| Error::Guard)? + .take(bound as u64 + 1) + .read_to_end(&mut data) + .map_err(|_| Error::Guard)?; + if data.len() > bound { + return Err(Error::Guard); + } + String::from_utf8(data).map_err(|_| Error::Guard) +} + +pub fn stopped() -> Result<()> { + for name in [ + "proxy", + "api", + "outbox", + "worker", + "problem-import", + "coordinator", + "sandbox-factory", + "postgres", + "rabbitmq", + ] { + let path = Path::new(CGROUP).join(format!("kf-{name}.service/cgroup.procs")); + match fs::symlink_metadata(&path) { + Ok(info) if info.is_file() && !info.file_type().is_symlink() => { + if !text(path, 8192)?.trim().is_empty() { + return Err(Error::Guard); + } + } + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + _ => return Err(Error::Guard), + } + } + let mut count = 0; + for entry in fs::read_dir("/proc").map_err(|_| Error::Guard)? { + let entry = entry.map_err(|_| Error::Guard)?; + if !entry + .file_name() + .as_encoded_bytes() + .iter() + .all(u8::is_ascii_digit) + { + continue; + } + count += 1; + if count > 1024 { + return Err(Error::Guard); + } + let path = entry.path().join("status"); + let raw = match text(&path, 16384) { + Ok(v) => v, + Err(_) if !path.exists() => continue, + Err(e) => return Err(e), + }; + let uid = raw + .lines() + .find_map(|s| s.strip_prefix("Uid:")) + .ok_or(Error::Guard)?; + let values: Vec<_> = uid.split_whitespace().map(str::parse::).collect(); + if values.len() != 4 + || values + .iter() + .any(|v| matches!(v, Ok(1200..=1205 | 100000) | Err(_))) + { + return Err(Error::Guard); + } + } + for name in ["tasks", "identities", "captures"] { + let path = Path::new("/var/lib/kf-judge").join(name); + match fs::read_dir(&path) { + Ok(mut entries) => { + if entries.next().is_some() { + return Err(Error::Guard); + } + } + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + _ => return Err(Error::Guard), + } + } + match fs::symlink_metadata("/var/lib/kf-judge/factory/control.sock") { + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), + _ => Err(Error::Guard), + } +} + +pub fn require() -> Result<()> { + let raw = text("/proc/self/status", 16384)?; + let fields: HashMap<_, _> = raw.lines().filter_map(|s| s.split_once(':')).collect(); + for key in ["Uid", "Gid"] { + if fields + .get(key) + .ok_or(Error::Guard)? + .split_whitespace() + .collect::>() + != ["0", "0", "0", "0"] + { + return Err(Error::Guard); + } + } + if fields + .get("Groups") + .ok_or(Error::Guard)? + .split_whitespace() + .any(|v| v != "0") + || fields.get("NoNewPrivs").ok_or(Error::Guard)?.trim() != "1" + { + return Err(Error::Guard); + } + for key in ["CapInh", "CapPrm", "CapEff", "CapBnd", "CapAmb"] { + if u64::from_str_radix(fields.get(key).ok_or(Error::Guard)?.trim(), 16) != Ok(15) { + return Err(Error::Guard); + } + } + if text("/proc/self/cgroup", 4096)?.trim() != "0::/kfonlinejudge.slice/kf-node-restore.service" + { + return Err(Error::Guard); + } + for (group, limits) in [ + ( + format!("{CGROUP}/kf-node-restore.service"), + ["536870912", "0", "32", "200000 100000"], + ), + ( + CGROUP.to_owned(), + ["4026531840", "0", "384", "200000 100000"], + ), + ] { + for (key, expected) in ["memory.max", "memory.swap.max", "pids.max", "cpu.max"] + .into_iter() + .zip(limits) + { + if text(Path::new(&group).join(key), 128)?.trim() != expected { + return Err(Error::Guard); + } + } + } + for (resource, maximum) in [ + (Resource::Core, 0), + (Resource::Fsize, 256 * 1024 * 1024), + (Resource::Nofile, 256), + (Resource::Stack, 16 * 1024 * 1024), + ] { + let observed = getrlimit(resource); + if observed.current != Some(maximum) || observed.maximum != Some(maximum) { + return Err(Error::Guard); + } + } + let metadata = fs::symlink_metadata(BACKUPS).map_err(|_| Error::Guard)?; + if !metadata.is_dir() + || metadata.uid() != 0 + || metadata.gid() != 0 + || metadata.mode() & 0o7777 != 0o700 + { + return Err(Error::Guard); + } + let mounts = text("/proc/self/mountinfo", 256 * 1024)?; + let rows: Vec> = mounts + .lines() + .map(|r| r.split_whitespace().collect()) + .collect(); + let found: Vec<_> = rows.iter().filter(|r| r.get(4) == Some(&BACKUPS)).collect(); + if found.len() != 1 { + return Err(Error::Guard); + } + let row = found[0]; + let sep = row.iter().position(|v| *v == "-").ok_or(Error::Guard)?; + if sep < 6 + || row.len() <= sep + 3 + || row[3] != "/" + || row[sep + 1] != "ext4" + || row[6..sep] + .iter() + .any(|v| v.starts_with("shared:") || v.starts_with("master:")) + || rows.iter().filter(|r| r.get(2) == Some(&row[2])).count() != 1 + { + return Err(Error::Guard); + } + let options: Vec<_> = row[5].split(',').chain(row[sep + 3].split(',')).collect(); + if ["ro", "nosuid", "nodev", "noexec"] + .iter() + .any(|v| !options.contains(v)) + { + return Err(Error::Guard); + } + let info = rustix::fs::statvfs(BACKUPS).map_err(|_| Error::Guard)?; + let bytes = info + .f_frsize + .checked_mul(info.f_blocks) + .ok_or(Error::Guard)?; + if !(16 * 1024_u64.pow(3) * 9 / 10..=16 * 1024_u64.pow(3)).contains(&bytes) { + return Err(Error::Guard); + } + stopped() +} diff --git a/judge/crates/sandbox/src/restore_copy/main.rs b/judge/crates/sandbox/src/restore_copy/main.rs new file mode 100644 index 0000000..b15578f --- /dev/null +++ b/judge/crates/sandbox/src/restore_copy/main.rs @@ -0,0 +1,401 @@ +//! Bounded offline restore data plane. Never starts a program or a service. + +mod files; +mod guard; + +use files::{Pending, Roots}; +use kf_contracts::{ + SchemaVersion, + restore_io::{self, Error, Reply, ReplyKind, Request, Stage}, +}; +use std::{ + collections::HashSet, + io::{self, BufRead, Read, Write}, + sync::{Arc, mpsc}, + thread::{self, JoinHandle}, +}; + +type Result = std::result::Result; + +fn now() -> Result { + let t = rustix::time::clock_gettime(rustix::time::ClockId::Monotonic); + u64::try_from(i128::from(t.tv_sec) * 1_000_000_000 + i128::from(t.tv_nsec)) + .map_err(|_| Error::Deadline) +} +fn remaining(deadline: u64) -> Result<()> { + if now()? >= deadline { + return Err(Error::Deadline); + } + Ok(()) +} + +enum Job { + Copy(Vec, u64), + Read(Vec, u64), + Stop, +} +enum Completed { + Copied(Result>), + Read(Result>), +} +struct Worker { + sender: mpsc::Sender, + receiver: mpsc::Receiver, + join: Option>, +} +struct Pool { + workers: Vec, +} +impl Pool { + fn new(roots: Arc) -> Result { + let mut pool = Self { + workers: Vec::new(), + }; + for _ in 0..2 { + let (sender, jobs) = mpsc::channel(); + let (completed, receiver) = mpsc::channel(); + let roots = Arc::clone(&roots); + let join = thread::Builder::new() + .name("restore-io".into()) + .stack_size(1024 * 1024) + .spawn(move || { + while let Ok(job) = jobs.recv() { + let result = match job { + Job::Copy(rows, deadline) => Completed::Copied( + rows.into_iter().map(|r| roots.copy(r, deadline)).collect(), + ), + Job::Read(rows, deadline) => { + Completed::Read(roots.read_group(rows, deadline)) + } + Job::Stop => break, + }; + if completed.send(result).is_err() { + break; + } + } + }) + .map_err(|_| Error::Filesystem)?; + pool.workers.push(Worker { + sender, + receiver, + join: Some(join), + }); + } + Ok(pool) + } + fn materialize( + &self, + rows: Vec, + roots: &Roots, + deadline: u64, + ) -> Result> { + let size = rows.len().div_ceil(2); + let mut groups = vec![Vec::new(), Vec::new()]; + for (i, row) in rows.into_iter().enumerate() { + groups[i / size].push(row); + } + for (worker, group) in self.workers.iter().zip(groups) { + worker + .sender + .send(Job::Copy(group, deadline)) + .map_err(|_| Error::Filesystem)?; + } + let mut pending = Vec::with_capacity(size * 2); + let mut error = None; + // Observe both completions even when the first failed. No sync/readback + // and no success response can outrun an outstanding copy worker. + for worker in &self.workers { + match worker.receiver.recv() { + Ok(Completed::Copied(Ok(rows))) => pending.extend(rows), + Ok(Completed::Copied(Err(e))) => { + error.get_or_insert(e); + } + _ => { + error.get_or_insert(Error::Filesystem); + } + } + } + if let Some(e) = error { + return Err(e); + } + roots.sync(deadline)?; + let mut groups = vec![Vec::new(), Vec::new()]; + for (i, row) in pending.into_iter().enumerate() { + groups[i / size].push(row); + } + for (worker, group) in self.workers.iter().zip(groups) { + worker + .sender + .send(Job::Read(group, deadline)) + .map_err(|_| Error::Filesystem)?; + } + let mut verified = Vec::with_capacity(size * 2); + for worker in &self.workers { + match worker.receiver.recv() { + Ok(Completed::Read(Ok(rows))) => verified.extend(rows), + Ok(Completed::Read(Err(e))) => { + error.get_or_insert(e); + } + _ => { + error.get_or_insert(Error::Filesystem); + } + } + } + if let Some(e) = error { + return Err(e); + } + remaining(deadline)?; + Ok(verified) + } +} +impl Drop for Pool { + fn drop(&mut self) { + for worker in &self.workers { + let _ = worker.sender.send(Job::Stop); + } + for worker in &mut self.workers { + if let Some(join) = worker.join.take() { + let _ = join.join(); + } + } + } +} + +struct Session { + // Pool drops/joins before the shared descriptors or verified inventory. + pool: Pool, + roots: Arc, + deadline: u64, + expected_files: u32, + expected_bytes: u64, + batch_id: u32, + bytes: u64, + seen: HashSet, + verified: Vec, +} +impl Session { + fn open(request: Request) -> Result { + let Request::Open { + backup_id, + recovery_id, + expected_files, + expected_logical_bytes, + deadline_monotonic_ns, + .. + } = request + else { + return Err(Error::Protocol); + }; + remaining(deadline_monotonic_ns)?; + if deadline_monotonic_ns + .checked_sub(now()?) + .is_none_or(|n| n > 1700 * 1_000_000_000) + { + return Err(Error::Deadline); + } + let roots = Arc::new(Roots::production(&backup_id, &recovery_id)?); + let pool = Pool::new(Arc::clone(&roots))?; + Ok(Self { + pool, + roots, + deadline: deadline_monotonic_ns, + expected_files, + expected_bytes: expected_logical_bytes, + batch_id: 0, + bytes: 0, + seen: HashSet::with_capacity(expected_files as usize), + verified: Vec::with_capacity(expected_files as usize), + }) + } + fn reply(&self, stage: Stage, error: Error) -> Reply { + Reply { + schema_version: SchemaVersion, + kind: ReplyKind::RestoreIO, + stage, + batch_id: self.batch_id, + verified_files: self.verified.len() as u32, + verified_logical_bytes: self.bytes, + execution_verified: false, + error, + } + } + fn batch(&mut self, batch_id: u32, rows: Vec) -> Result<()> { + remaining(self.deadline)?; + let bytes: u64 = rows.iter().map(|r| r.bytes).sum(); + if batch_id != self.batch_id + 1 + || self.verified.len() + rows.len() > self.expected_files as usize + || self + .bytes + .checked_add(bytes) + .is_none_or(|n| n > self.expected_bytes) + || rows.iter().any(|r| !self.seen.insert(r.path.clone())) + { + return Err(Error::Protocol); + } + let verified = self.pool.materialize(rows, &self.roots, self.deadline)?; + self.verified.extend(verified); + self.bytes += bytes; + self.batch_id = batch_id; + Ok(()) + } + fn finish(&self) -> Result<()> { + remaining(self.deadline)?; + if self.verified.len() != self.expected_files as usize || self.bytes != self.expected_bytes + { + return Err(Error::Protocol); + } + guard::require()?; + self.roots.unchanged()?; + for pending in &self.verified { + remaining(self.deadline)?; + self.roots.final_file(pending)?; + } + self.roots.unchanged()?; + guard::stopped()?; + remaining(self.deadline)?; + Ok(()) + } +} + +fn request(input: &mut impl BufRead) -> Result { + let mut bytes = Vec::new(); + input + .take(restore_io::MAX_REQUEST_BYTES as u64 + 1) + .read_until(b'\n', &mut bytes) + .map_err(|_| Error::Protocol)?; + if bytes.last() != Some(&b'\n') || bytes.len() > restore_io::MAX_REQUEST_BYTES { + return Err(Error::Protocol); + } + restore_io::decode_request(&bytes).map_err(|_| Error::Protocol) +} +fn respond(output: &mut impl Write, value: &Reply) -> Result<()> { + value.validate().map_err(|_| Error::Protocol)?; + let mut bytes = serde_json::to_vec(value).map_err(|_| Error::Protocol)?; + bytes.push(b'\n'); + if bytes.len() > restore_io::MAX_REPLY_BYTES { + return Err(Error::Protocol); + } + output + .write_all(&bytes) + .and_then(|_| output.flush()) + .map_err(|_| Error::Protocol) +} + +fn run(input: &mut impl BufRead, output: &mut impl Write) -> Result<()> { + guard::require()?; + let mut session = Session::open(request(input)?)?; + respond(output, &session.reply(Stage::Opened, Error::None))?; + loop { + let action = request(input); + let result = match action { + Ok(Request::Batch { + batch_id, files, .. + }) => session.batch(batch_id, files), + Ok(Request::Finish { .. }) => { + let result = session.finish(); + match result { + Ok(()) => { + // Join both workers before the finished byte-copy reply. + let reply = session.reply(Stage::Finished, Error::None); + drop(session); + respond(output, &reply)?; + return Ok(()); + } + Err(e) => Err(e), + } + } + _ => Err(Error::Protocol), + }; + match result { + Ok(()) => respond(output, &session.reply(Stage::Verified, Error::None))?, + Err(e) => { + let reply = session.reply(Stage::Refused, e); + drop(session); + respond(output, &reply)?; + return Err(e); + } + } + } +} + +fn main() -> std::process::ExitCode { + if std::env::args_os().skip(1).collect::>() != ["--stdio-v1"] { + return std::process::ExitCode::FAILURE; + } + let result = run(&mut io::stdin().lock(), &mut io::stdout().lock()); + if result.is_ok() { + std::process::ExitCode::SUCCESS + } else { + std::process::ExitCode::FAILURE + } +} + +#[cfg(test)] +mod tests { + use super::*; + use sha2::{Digest, Sha256}; + use std::{fs, os::unix::fs::PermissionsExt}; + + // Trusted tiny ordinary fixtures exercise bytes/joining, without bypassing + // the production CLI guards or claiming Root/unit/isolation acceptance. + #[test] + fn joined_copy_sync_full_readback_and_final_witnesses() { + let fixture = files::Fixture::new(); + let roots = Arc::new(fixture.roots()); + let data = b"ordinary trusted restore fixture\n".repeat(128); + let hash = format!("{:x}", Sha256::digest(&data)); + fs::write(fixture.source().join(&hash), &data).unwrap(); + fs::set_permissions( + fixture.source().join(&hash), + fs::Permissions::from_mode(0o600), + ) + .unwrap(); + fs::write(fixture.target().join("existing"), &data).unwrap(); + fs::set_permissions( + fixture.target().join("existing"), + fs::Permissions::from_mode(0o644), + ) + .unwrap(); + let owner = fs::metadata(fixture.source()).unwrap(); + use std::os::unix::fs::MetadataExt; + let rows = ["existing", "one", "two", "three"] + .into_iter() + .map(|name| restore_io::RestoreFile { + path: format!("artifacts/{name}"), + uid: owner.uid(), + gid: owner.gid(), + mode: 0o644, + bytes: data.len() as u64, + sha256: hash.clone(), + }) + .collect(); + let deadline = now().unwrap() + 10_000_000_000; + let pool = Pool::new(Arc::clone(&roots)).unwrap_or_else(|_| panic!("worker pool")); + let verified = pool + .materialize(rows, &roots, deadline) + .unwrap_or_else(|_| panic!("verified copy")); + assert_eq!(verified.len(), 4); + for row in &verified { + roots + .final_file(row) + .unwrap_or_else(|_| panic!("final witness")); + let name = row.row.path.strip_prefix("artifacts/").unwrap(); + assert_eq!(fs::read(fixture.target().join(name)).unwrap(), data); + } + drop(pool); + roots.unchanged().unwrap_or_else(|_| panic!("root anchors")); + } + + #[test] + fn request_line_requires_finite_complete_frame() { + assert!(matches!( + request(&mut &b"{\"schema_version\":1,\"kind\":\"restore.io.finish\"}\n"[..]), + Ok(Request::Finish { .. }) + )); + assert!( + request(&mut &b"{\"schema_version\":1,\"kind\":\"restore.io.finish\"}"[..]).is_err() + ); + assert!(request(&mut &b""[..]).is_err()); + assert!(request(&mut &vec![b' '; restore_io::MAX_REQUEST_BYTES + 1][..]).is_err()); + } +} diff --git a/scripts/check_architecture.py b/scripts/check_architecture.py index b9466cd..b8bd1a6 100644 --- a/scripts/check_architecture.py +++ b/scripts/check_architecture.py @@ -108,8 +108,38 @@ def release_violations() -> list[str]: return violations +def restore_copy_violations() -> list[str]: + """Keep the fixed offline data plane inside Sandbox and free of execution.""" + root = ROOT / "judge/crates/sandbox" + manifest = tomllib.loads((root / "Cargo.toml").read_text(encoding="utf-8")) + entries = [row for row in manifest.get("bin", []) if row.get("name") == "kf-restore-io"] + if entries != [{"name": "kf-restore-io", "path": "src/restore_copy/main.rs"}]: + return ["kf-restore-io must remain the declared offline Sandbox entry"] + violations = [] + for path in sorted((root / "src/restore_copy").glob("*.rs")): + text = path.read_text(encoding="utf-8") + # A source boundary check, complemented by actual Root/cgroup/IO VM tests. + for forbidden in ( + "Command", + "std::net", + "kf_sandbox::", + "kf_engine::", + "kf_worker::", + "kf_coordinator::", + "libc::", + "execve", + "execv(", + "fork(", + ): + if forbidden in text: + violations.append(f"{path.relative_to(ROOT)}: offline copy cannot use {forbidden}") + return violations + + def main() -> None: - violations = python_violations() + rust_violations() + release_violations() + violations = ( + python_violations() + rust_violations() + release_violations() + restore_copy_violations() + ) if violations: print("\n".join(violations), file=sys.stderr) raise SystemExit(1) diff --git a/scripts/release/README.md b/scripts/release/README.md index f5a94a3..c34d962 100644 --- a/scripts/release/README.md +++ b/scripts/release/README.md @@ -411,6 +411,25 @@ all normal services and tasks are stopped. The stopped-state checks precede data verification and copying and run again before completion. Original identities, modes and replay IDs are preserved. Its PostgreSQL target is writable; the fresh RabbitMQ volume is checked through a read-only mount and receives no copied broker state. +Version 4 delegates its file data plane to the selected bundle's `kf-restore-io` +binary inside the existing 512 MiB/200% CPU/32-task restore unit. A single native +process retains two joined worker threads and the verified file witnesses; it +never forks a copy of Python's full inventory. Its only input is the versioned +private stdio protocol, with fixed source/target domains and no command or path +overrides. The binary independently checks Root capabilities, actual unit/node +limits, the read-only backup volume and stopped writers. It starts no programs +or services. Python retains its filesystem error cursors from before directory +creation; the native child holds its own cursors before file writes. Both cursor +sets are synchronized before completion. The original deadlines remain. +Each batch copies, joins both workers, flushes the affected filesystems, then +reads every target. Final directory sync and stopped/selected-identity checks +precede the native inventory's complete metadata recheck and thread joins. +Transport, copy, byte, sync or identity failure retains the partial target and +returns no successful data receipt. There is no fallback after a native failure. +Version 3 keeps the prior per-file implementation. Older complete bundles remain +eligible for compatible software rollback; new native support requires both its +binary and Python transport module in the bundle. + Each restored file is flushed and read back in full. Within a joined half-batch, repeated digest/size groups can retain at most 8 MiB of reference bytes per reader. The first target receives an actual SHA-256 check; every later target is fully diff --git a/scripts/release/build.py b/scripts/release/build.py index 4b8b0a2..e24798d 100644 --- a/scripts/release/build.py +++ b/scripts/release/build.py @@ -340,7 +340,7 @@ def build(repo: Path, inputs: Path, output: Path, work: Path) -> dict: name = raw.decode() relative_name("delivery/" + name) copy(repo / name, payload / "delivery" / name) - for name in ("kf-sandbox", "kf-worker", "kf-coordinator"): + for name in ("kf-sandbox", "kf-worker", "kf-coordinator", "kf-restore-io"): copy(repo / "judge/target/release" / name, payload / "bin" / name) for name in sorted(records): if name.startswith("python/lib/") or name == "python/bin/python3.13": diff --git a/scripts/release/manifest.py b/scripts/release/manifest.py index fe0a427..abbbe85 100644 --- a/scripts/release/manifest.py +++ b/scripts/release/manifest.py @@ -349,6 +349,11 @@ def validate(data: dict) -> dict: "frontend/admin/index.html", "deploy/release/contract.v1.json", } + # Older deliveries remain valid rollback candidates. New native restore + # support must be complete in both directions when either member is present. + restore_pair = {"bin/kf-restore-io", "delivery/scripts/release/restore_native.py"} + if seen & restore_pair: + required.update(restore_pair) required.update(f"images/{name}/manifest.json" for name in data["images"]) if not required <= seen: raise DeliveryError("incomplete application bundle") diff --git a/scripts/release/restore.py b/scripts/release/restore.py index 687e113..fc78efc 100644 --- a/scripts/release/restore.py +++ b/scripts/release/restore.py @@ -23,6 +23,7 @@ from . import ( manifest, preflight, restore_io, + restore_native, restore_policy, ) from .policy import DeliveryError, canonical, digest, sha256 @@ -295,6 +296,7 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict: "restore.py", "restore_policy.py", "restore_io.py", + "restore_native.py", "backup.py", "backup_manifest.py", "backup_objects.py", @@ -345,6 +347,7 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict: } deadline = time.monotonic() + 1700 filesystems = None + native = None try: fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) stopped() @@ -398,15 +401,15 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict: if value["schema_version"] == 4: assert filesystems is not None - verified = restore_io.materialize( - value["files"], - lambda item: backup_manifest.member_path(root, value, item), - lambda item: _target(item["path"], item, recovery), - readback, - filesystems, + native = restore_native.Session( + payload, + backup_id, + recovery_id, + len(value["files"]), + backup_manifest.logical_bytes(value), deadline, - group_readback=readback_group, ) + native.materialize(value["files"]) else: witnesses = file_copy.bounded_copies(value["files"], copy, deadline=deadline) verified = { @@ -439,11 +442,17 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict: stopped() if install.selection() != selected: raise DeliveryError("selected recovery identity changed") - for item in value["files"]: - if time.monotonic() >= deadline: - raise DeliveryError("recovery verification deadline exceeded") - path, uid, gid, mode = _target(item["path"], item, recovery) - unchanged_file(path, verified[item["path"]], uid, gid, mode) + if native is not None: + # The native child retained every post-sync full-byte witness. It + # rechecks the complete inventory after final directory sync and + # stopped/selection proofs, then joins before this receipt exists. + native.finish() + else: + for item in value["files"]: + if time.monotonic() >= deadline: + raise DeliveryError("recovery verification deadline exceeded") + path, uid, gid, mode = _target(item["path"], item, recovery) + unchanged_file(path, verified[item["path"]], uid, gid, mode) if filesystems is not None: filesystems.close() filesystems = None @@ -461,10 +470,14 @@ def restore(recovery_id: str, backup_id: str, expected: str) -> dict: return record finally: try: - if filesystems is not None: - filesystems.close() + if native is not None: + native.close() finally: - os.close(descriptor) + try: + if filesystems is not None: + filesystems.close() + finally: + os.close(descriptor) def launch(recovery_id: str, backup_id: str, expected: str) -> dict: diff --git a/scripts/release/restore_native.py b/scripts/release/restore_native.py new file mode 100644 index 0000000..b5a7da6 --- /dev/null +++ b/scripts/release/restore_native.py @@ -0,0 +1,220 @@ +"""One bounded native data-copy child in the existing exclusive restore unit.""" + +import json +import os +import selectors +import signal +import stat +import subprocess +import time +from pathlib import Path + +from kf_oj.contracts.restore_io import RestoreIOReply + +from . import file_copy, preflight, restore_io +from .policy import DeliveryError, invalid_constant, unique_object + +MAX_REQUEST_BYTES = 256 * 1024 +MAX_REPLY_BYTES = 64 * 1024 +MAX_STDERR_BYTES = 16 * 1024 + + +class Session: + """No fallback after partial writes; all verified witnesses stay in the child.""" + + def __init__( + self, + payload: Path, + backup_id: str, + recovery_id: str, + expected_files: int, + expected_bytes: int, + deadline: float, + ): + self.deadline = deadline + self.batch_id = self.files = self.bytes = 0 + self.expected_files, self.expected_bytes = expected_files, expected_bytes + self.finished = False + self.process = None + self.stderr_bytes = 0 + helper = payload / "bin/kf-restore-io" + preflight.protected(helper) + info = helper.lstat() + if ( + not stat.S_ISREG(info.st_mode) + or info.st_nlink != 1 + or stat.S_IMODE(info.st_mode) != 0o555 + ): + raise DeliveryError("selected native recovery helper differs") + witness = file_copy._identity(info) + self.selector = selectors.DefaultSelector() + try: + restore_io.remaining(self.deadline) + self.process = subprocess.Popen( + [str(helper), "--stdio-v1"], + cwd="/", + env={ + "PATH": "/usr/bin:/bin", + "LC_ALL": "C.UTF-8", + "LD_LIBRARY_PATH": f"{payload}/lib:{payload}/python/lib", + }, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + start_new_session=True, + close_fds=True, + ) + if file_copy._identity(helper.lstat()) != witness: + raise DeliveryError("selected native recovery helper changed") + assert self.process.stdin and self.process.stdout and self.process.stderr + for handle in (self.process.stdin, self.process.stdout, self.process.stderr): + os.set_blocking(handle.fileno(), False) + self.selector.register(self.process.stdout, selectors.EVENT_READ, "stdout") + self.selector.register(self.process.stderr, selectors.EVENT_READ, "stderr") + self._exchange( + { + "schema_version": 1, + "kind": "restore.io.open", + "backup_id": backup_id, + "recovery_id": recovery_id, + "expected_files": expected_files, + "expected_logical_bytes": expected_bytes, + "deadline_monotonic_ns": int(deadline * 1_000_000_000), + }, + "opened", + ) + except BaseException: + self.close() + raise + + def _exchange(self, value: dict, stage: str) -> None: + restore_io.remaining(self.deadline) + if self.process is None or self.finished: + raise DeliveryError("native recovery session is closed") + raw = json.dumps(value, separators=(",", ":"), ensure_ascii=True).encode() + b"\n" + if len(raw) > MAX_REQUEST_BYTES: + raise DeliveryError("native recovery request exceeds bound") + assert self.process.stdin + self.selector.register(self.process.stdin, selectors.EVENT_WRITE, "stdin") + offset = 0 + output = bytearray() + try: + while True: + restore_io.remaining(self.deadline) + ready = self.selector.select(max(0, self.deadline - time.monotonic())) + if not ready: + raise DeliveryError("native recovery copy deadline exceeded") + for key, _ in ready: + if key.data == "stdin": + try: + count = os.write(key.fd, memoryview(raw)[offset:]) + except BlockingIOError: + continue + if count <= 0: + raise DeliveryError("native recovery request pipe closed") + offset += count + if offset == len(raw): + self.selector.unregister(self.process.stdin) + continue + try: + block = os.read(key.fd, 8192) + except BlockingIOError: + continue + if key.data == "stderr": + self.stderr_bytes += len(block) + if self.stderr_bytes > MAX_STDERR_BYTES: + raise DeliveryError("native recovery diagnostic exceeds bound") + if not block: + self.selector.unregister(key.fileobj) + continue + if not block: + raise DeliveryError("native recovery stopped before verification") + output.extend(block) + if len(output) > MAX_REPLY_BYTES: + raise DeliveryError("native recovery reply exceeds bound") + if b"\n" in output: + if offset != len(raw) or output[-1:] != b"\n" or output.count(b"\n") != 1: + raise DeliveryError("native recovery reply framing differs") + reply = RestoreIOReply.model_validate( + json.loads( + output, + object_pairs_hook=unique_object, + parse_constant=invalid_constant, + ) + ) + if ( + reply.stage != stage + or reply.error != "none" + or (reply.batch_id, reply.verified_files, reply.verified_logical_bytes) + != (self.batch_id, self.files, self.bytes) + ): + raise DeliveryError("native recovery file verification refused") + restore_io.remaining(self.deadline) + return + except (OSError, ValueError, RecursionError) as error: + raise DeliveryError("native recovery transport or reply differs") from error + finally: + if self.process.stdin in self.selector.get_map(): + self.selector.unregister(self.process.stdin) + + def materialize(self, rows: list[dict]) -> None: + for batch in restore_io.batches(rows): + self.batch_id += 1 + self.files += len(batch) + self.bytes += sum(item["bytes"] for item in batch) + if self.files > self.expected_files or self.bytes > self.expected_bytes: + raise DeliveryError("native recovery inventory exceeds original receipt") + # The complete private manifest was already strictly validated. + # The native decoder independently validates each row before use. + self._exchange( + { + "schema_version": 1, + "kind": "restore.io.batch", + "batch_id": self.batch_id, + "files": batch, + }, + "verified", + ) + + def finish(self) -> None: + if (self.files, self.bytes) != (self.expected_files, self.expected_bytes): + raise DeliveryError("native recovery inventory incomplete") + self._exchange({"schema_version": 1, "kind": "restore.io.finish"}, "finished") + assert self.process is not None and self.process.stdin + self.process.stdin.close() + restore_io.remaining(self.deadline) + try: + code = self.process.wait(timeout=max(0, self.deadline - time.monotonic())) + except subprocess.TimeoutExpired as error: + raise DeliveryError("native recovery child did not join") from error + if code != 0: + raise DeliveryError("native recovery child failed after verification") + restore_io.remaining(self.deadline) + self.finished = True + + def close(self) -> None: + process = self.process + try: + if process is not None and process.poll() is None: + # This child starts its own session and never spawns processes. + # Its only descendants are its two native threads. + try: + if os.getpgid(process.pid) != process.pid: + raise DeliveryError("native recovery child ownership differs") + os.killpg(process.pid, signal.SIGTERM) + except ProcessLookupError: + pass + try: + process.wait(timeout=3) + except subprocess.TimeoutExpired: + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + process.wait(timeout=2) + finally: + self.selector.close() + if process is not None: + for handle in (process.stdin, process.stdout, process.stderr): + if handle is not None: + handle.close() diff --git a/tests/fixtures/restore-io/examples.json b/tests/fixtures/restore-io/examples.json new file mode 100644 index 0000000..5fe0fca --- /dev/null +++ b/tests/fixtures/restore-io/examples.json @@ -0,0 +1,631 @@ +[ + { + "name": "open", + "direction": "request", + "valid": true, + "value": { + "schema_version": 1, + "kind": "restore.io.open", + "backup_id": "bk-11111111111111111111111111111111", + "recovery_id": "rr-22222222222222222222222222222222", + "expected_files": 1, + "expected_logical_bytes": 4, + "deadline_monotonic_ns": 1700000000000 + } + }, + { + "name": "batch", + "direction": "request", + "valid": true, + "value": { + "schema_version": 1, + "kind": "restore.io.batch", + "batch_id": 1, + "files": [ + { + "path": "artifacts/trusted.txt", + "uid": 1200, + "gid": 1200, + "mode": 420, + "bytes": 4, + "sha256": "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7" + } + ] + } + }, + { + "name": "finish", + "direction": "request", + "valid": true, + "value": { + "schema_version": 1, + "kind": "restore.io.finish" + } + }, + { + "name": "opened", + "direction": "reply", + "valid": true, + "value": { + "schema_version": 1, + "kind": "restore.io.reply", + "stage": "opened", + "batch_id": 0, + "verified_files": 0, + "verified_logical_bytes": 0, + "execution_verified": false, + "error": "none" + } + }, + { + "name": "verified", + "direction": "reply", + "valid": true, + "value": { + "schema_version": 1, + "kind": "restore.io.reply", + "stage": "verified", + "batch_id": 1, + "verified_files": 1, + "verified_logical_bytes": 4, + "execution_verified": false, + "error": "none" + } + }, + { + "name": "finished", + "direction": "reply", + "valid": true, + "value": { + "schema_version": 1, + "kind": "restore.io.reply", + "stage": "finished", + "batch_id": 1, + "verified_files": 1, + "verified_logical_bytes": 4, + "execution_verified": false, + "error": "none" + } + }, + { + "name": "refused", + "direction": "reply", + "valid": true, + "value": { + "schema_version": 1, + "kind": "restore.io.reply", + "stage": "refused", + "batch_id": 0, + "verified_files": 0, + "verified_logical_bytes": 0, + "execution_verified": false, + "error": "identity" + } + }, + { + "name": "bad-open-schema_version-True", + "direction": "request", + "valid": false, + "value": { + "schema_version": true, + "kind": "restore.io.open", + "backup_id": "bk-11111111111111111111111111111111", + "recovery_id": "rr-22222222222222222222222222222222", + "expected_files": 1, + "expected_logical_bytes": 4, + "deadline_monotonic_ns": 1700000000000 + } + }, + { + "name": "bad-open-expected_files-0", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.open", + "backup_id": "bk-11111111111111111111111111111111", + "recovery_id": "rr-22222222222222222222222222222222", + "expected_files": 0, + "expected_logical_bytes": 4, + "deadline_monotonic_ns": 1700000000000 + } + }, + { + "name": "bad-open-expected_files-98305", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.open", + "backup_id": "bk-11111111111111111111111111111111", + "recovery_id": "rr-22222222222222222222222222222222", + "expected_files": 98305, + "expected_logical_bytes": 4, + "deadline_monotonic_ns": 1700000000000 + } + }, + { + "name": "bad-open-expected_logical_bytes-12884901889", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.open", + "backup_id": "bk-11111111111111111111111111111111", + "recovery_id": "rr-22222222222222222222222222222222", + "expected_files": 1, + "expected_logical_bytes": 12884901889, + "deadline_monotonic_ns": 1700000000000 + } + }, + { + "name": "bad-open-deadline_monotonic_ns-1.0", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.open", + "backup_id": "bk-11111111111111111111111111111111", + "recovery_id": "rr-22222222222222222222222222222222", + "expected_files": 1, + "expected_logical_bytes": 4, + "deadline_monotonic_ns": 1.0 + } + }, + { + "name": "bad-open-backup_id-bk-AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.open", + "backup_id": "bk-AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", + "recovery_id": "rr-22222222222222222222222222222222", + "expected_files": 1, + "expected_logical_bytes": 4, + "deadline_monotonic_ns": 1700000000000 + } + }, + { + "name": "bad-open-recovery_id-rr-../etc", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.open", + "backup_id": "bk-11111111111111111111111111111111", + "recovery_id": "rr-../etc", + "expected_files": 1, + "expected_logical_bytes": 4, + "deadline_monotonic_ns": 1700000000000 + } + }, + { + "name": "bad-open-extra-1", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.open", + "backup_id": "bk-11111111111111111111111111111111", + "recovery_id": "rr-22222222222222222222222222222222", + "expected_files": 1, + "expected_logical_bytes": 4, + "deadline_monotonic_ns": 1700000000000, + "extra": 1 + } + }, + { + "name": "bad-file-path-artifacts/../x", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.batch", + "batch_id": 1, + "files": [ + { + "path": "artifacts/../x", + "uid": 1200, + "gid": 1200, + "mode": 420, + "bytes": 4, + "sha256": "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7" + } + ] + } + }, + { + "name": "bad-file-path-artifacts/x/.ssh/id", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.batch", + "batch_id": 1, + "files": [ + { + "path": "artifacts/x/.ssh/id", + "uid": 1200, + "gid": 1200, + "mode": 420, + "bytes": 4, + "sha256": "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7" + } + ] + } + }, + { + "name": "bad-file-path-artifacts", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.batch", + "batch_id": 1, + "files": [ + { + "path": "artifacts", + "uid": 1200, + "gid": 1200, + "mode": 420, + "bytes": 4, + "sha256": "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7" + } + ] + } + }, + { + "name": "bad-file-path-x/y", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.batch", + "batch_id": 1, + "files": [ + { + "path": "x/y", + "uid": 1200, + "gid": 1200, + "mode": 420, + "bytes": 4, + "sha256": "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7" + } + ] + } + }, + { + "name": "bad-file-path-artifacts/你好", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.batch", + "batch_id": 1, + "files": [ + { + "path": "artifacts/你好", + "uid": 1200, + "gid": 1200, + "mode": 420, + "bytes": 4, + "sha256": "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7" + } + ] + } + }, + { + "name": "bad-file-uid-1000", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.batch", + "batch_id": 1, + "files": [ + { + "path": "artifacts/trusted.txt", + "uid": 1000, + "gid": 1200, + "mode": 420, + "bytes": 4, + "sha256": "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7" + } + ] + } + }, + { + "name": "bad-file-gid-1201", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.batch", + "batch_id": 1, + "files": [ + { + "path": "artifacts/trusted.txt", + "uid": 1200, + "gid": 1201, + "mode": 420, + "bytes": 4, + "sha256": "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7" + } + ] + } + }, + { + "name": "bad-file-mode-2541", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.batch", + "batch_id": 1, + "files": [ + { + "path": "artifacts/trusted.txt", + "uid": 1200, + "gid": 1200, + "mode": 2541, + "bytes": 4, + "sha256": "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7" + } + ] + } + }, + { + "name": "bad-file-bytes--1", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.batch", + "batch_id": 1, + "files": [ + { + "path": "artifacts/trusted.txt", + "uid": 1200, + "gid": 1200, + "mode": 420, + "bytes": -1, + "sha256": "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7" + } + ] + } + }, + { + "name": "bad-file-bytes-268435457", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.batch", + "batch_id": 1, + "files": [ + { + "path": "artifacts/trusted.txt", + "uid": 1200, + "gid": 1200, + "mode": 420, + "bytes": 268435457, + "sha256": "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7" + } + ] + } + }, + { + "name": "bad-file-bytes-False", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.batch", + "batch_id": 1, + "files": [ + { + "path": "artifacts/trusted.txt", + "uid": 1200, + "gid": 1200, + "mode": 420, + "bytes": false, + "sha256": "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7" + } + ] + } + }, + { + "name": "bad-file-sha256-0000000000000000000000000000000000000000000000000000000000000000", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.batch", + "batch_id": 1, + "files": [ + { + "path": "artifacts/trusted.txt", + "uid": 1200, + "gid": 1200, + "mode": 420, + "bytes": 4, + "sha256": "0000000000000000000000000000000000000000000000000000000000000000" + } + ] + } + }, + { + "name": "duplicate-file", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.batch", + "batch_id": 1, + "files": [ + { + "path": "artifacts/trusted.txt", + "uid": 1200, + "gid": 1200, + "mode": 420, + "bytes": 4, + "sha256": "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7" + }, + { + "path": "artifacts/trusted.txt", + "uid": 1200, + "gid": 1200, + "mode": 420, + "bytes": 4, + "sha256": "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7" + } + ] + } + }, + { + "name": "multi-large", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.batch", + "batch_id": 1, + "files": [ + { + "path": "artifacts/trusted.txt", + "uid": 1200, + "gid": 1200, + "mode": 420, + "bytes": 67108864, + "sha256": "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7" + }, + { + "path": "artifacts/b", + "uid": 1200, + "gid": 1200, + "mode": 420, + "bytes": 4, + "sha256": "3a6eb0790f39ac87c94f3856b2dd2c5d110e6811602261a9a923d3bb23adc8b7" + } + ] + } + }, + { + "name": "empty-batch", + "direction": "request", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.batch", + "batch_id": 1, + "files": [] + } + }, + { + "name": "bad-reply-execution_verified-True", + "direction": "reply", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.reply", + "stage": "opened", + "batch_id": 0, + "verified_files": 0, + "verified_logical_bytes": 0, + "execution_verified": true, + "error": "none" + } + }, + { + "name": "bad-reply-execution_verified-0", + "direction": "reply", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.reply", + "stage": "opened", + "batch_id": 0, + "verified_files": 0, + "verified_logical_bytes": 0, + "execution_verified": 0, + "error": "none" + } + }, + { + "name": "bad-reply-stage-unknown", + "direction": "reply", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.reply", + "stage": "unknown", + "batch_id": 0, + "verified_files": 0, + "verified_logical_bytes": 0, + "execution_verified": false, + "error": "none" + } + }, + { + "name": "bad-reply-error-bytes", + "direction": "reply", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.reply", + "stage": "opened", + "batch_id": 0, + "verified_files": 0, + "verified_logical_bytes": 0, + "execution_verified": false, + "error": "bytes" + } + }, + { + "name": "bad-reply-verified_files-1", + "direction": "reply", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.reply", + "stage": "opened", + "batch_id": 0, + "verified_files": 1, + "verified_logical_bytes": 0, + "execution_verified": false, + "error": "none" + } + }, + { + "name": "bad-reply-schema_version-1.0", + "direction": "reply", + "valid": false, + "value": { + "schema_version": 1.0, + "kind": "restore.io.reply", + "stage": "opened", + "batch_id": 0, + "verified_files": 0, + "verified_logical_bytes": 0, + "execution_verified": false, + "error": "none" + } + }, + { + "name": "bad-reply-extra-1", + "direction": "reply", + "valid": false, + "value": { + "schema_version": 1, + "kind": "restore.io.reply", + "stage": "opened", + "batch_id": 0, + "verified_files": 0, + "verified_logical_bytes": 0, + "execution_verified": false, + "error": "none", + "extra": 1 + } + } +]