From 502d09980ffed746e5006136ea79eead19030f24 Mon Sep 17 00:00:00 2001 From: KiriAky 107 Date: Wed, 9 Sep 2026 14:36:55 +0800 Subject: [PATCH] =?UTF-8?q?test(sync):=20=E5=AE=8C=E6=88=90=20S-06=20?= =?UTF-8?q?=E5=AE=89=E5=85=A8=E9=AA=8C=E6=94=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/development/OpenNexus生产验收Runner.md | 2 +- scripts/acceptance_cases/s06_sync_security.py | 479 ++++++++++++++++++ .../acceptance_cases/sync_production_stack.py | 172 ++++--- scripts/phase3_acceptance.py | 15 + server sync/sync_server/app.py | 12 +- server sync/tests/test_production_storage.py | 4 +- 6 files changed, 617 insertions(+), 67 deletions(-) create mode 100644 scripts/acceptance_cases/s06_sync_security.py diff --git a/docs/development/OpenNexus生产验收Runner.md b/docs/development/OpenNexus生产验收Runner.md index deae8c5..df9d335 100644 --- a/docs/development/OpenNexus生产验收Runner.md +++ b/docs/development/OpenNexus生产验收Runner.md @@ -24,4 +24,4 @@ python scripts/phase3-production-acceptance.py ` 报告目录包含 `summary.json`、`case-manifest.json`、`junit.xml`、`cases/.json` 和脱敏的 `logs/.log`。摘要记录 commit、各锁文件 SHA-256、配置摘要与已提供安装产物摘要。日志将仓库、数据根、报告根、用户主目录和配置声明的秘密值替换为占位符，并限制为 10 MiB。报告目录必须为空，避免单例复跑覆盖原始证据。 -当前 runner 与失败闭合行为已实现，A-02/A-03 Sidecar、B-01/B-02 凭据、D-01 扩展包、S-01/S-02/S-03/S-08 Sync 客户端以及 S-04/S-05 Sync 服务 driver 已登记；其余 19 个生产验收 ID 尚未登记，运行时会生成 `NOT_IMPLEMENTED` 证据并退出 1。这用于阻止误报，不是这些用例的验收通过。 +当前 runner 与失败闭合行为已实现，A-02/A-03 Sidecar、B-01/B-02 凭据、D-01 扩展包、S-01/S-02/S-03/S-08 Sync 客户端以及 S-04/S-05/S-06 Sync 服务 driver 已登记；其余 18 个生产验收 ID 尚未登记，运行时会生成 `NOT_IMPLEMENTED` 证据并退出 1。这用于阻止误报，不是这些用例的验收通过。 diff --git a/scripts/acceptance_cases/s06_sync_security.py b/scripts/acceptance_cases/s06_sync_security.py new file mode 100644 index 0000000..6c9a11c --- /dev/null +++ b/scripts/acceptance_cases/s06_sync_security.py @@ -0,0 +1,479 @@ +"""S-06 authorization, revocation, rate-limit, and readiness acceptance.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import secrets +import time +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +from sync_production_stack import ROOT, SyncProductionStack, call, sha256 + +ROUNDS = 100 + + +def result(case_id: str, status: str, reason: str, facts: dict) -> dict: + passed = status == "PASSED" + assertions = [ + ( + "the oracle uses real PostgreSQL, MinIO, and both production workers", + facts.get("dependencies") is True and facts.get("worker_count") == 2, + ), + ( + "100 cross-account or cross-vault hash guesses reveal no object", + facts.get("hash_rejections") == ROUNDS and facts.get("content_leaks") == 0, + ), + ( + "a revoked device is rejected on the next request and for 100 requests", + facts.get("immediate_revocation") is True + and facts.get("revoked_rejections") == ROUNDS + and facts.get("revoked_refresh_rejections") == ROUNDS, + ), + ( + "100 expired access-token requests are rejected and refresh rotates once", + facts.get("expired_token_rejections") == ROUNDS + and facts.get("refresh_replay_rejections") == ROUNDS + and facts.get("rotated_access_works") is True, + ), + ( + "foreign-device and expired upload links reject 100 requests each", + facts.get("foreign_upload_rejections") == ROUNDS + and facts.get("expired_upload_rejections") == ROUNDS, + ), + ( + "database, S3, and staging faults and recoveries are visible within eight seconds", + facts.get("readiness_transitions") == 6 + and facts.get("ready_failure_max_ms", 8001) < 8000 + and facts.get("ready_recovery_max_ms", 8001) < 8000, + ), + ( + "100 limited logins return 429 with Retry-After after the fixed threshold", + facts.get("prelimit_rejections") == 10 + and facts.get("rate_limited_responses") == ROUNDS + and facts.get("retry_after_responses") == ROUNDS, + ), + ] + evidence = "real PostgreSQL 17, MinIO S3, two Uvicorn workers, and black-box HTTP" + return { + "schema": 1, + "case_id": case_id, + "status": status, + "reason": reason, + "assertions": [ + { + "name": name, + "status": "PASSED" if passed and actual else "FAILED", + "evidence": evidence, + } + for name, actual in assertions + ], + "metrics": { + "hash_rejections": facts.get("hash_rejections", 0), + "revoked_rejections": facts.get("revoked_rejections", 0), + "expired_token_rejections": facts.get("expired_token_rejections", 0), + "upload_link_rejections": facts.get("foreign_upload_rejections", 0) + + facts.get("expired_upload_rejections", 0), + "rate_limited_responses": facts.get("rate_limited_responses", 0), + "ready_failure_max_ms": facts.get("ready_failure_max_ms", 0), + "ready_recovery_max_ms": facts.get("ready_recovery_max_ms", 0), + "worker_count": facts.get("worker_count", 0), + }, + "files": [ + {"path": relative, "sha256": sha256(ROOT / relative)} + for relative in ( + "server sync/sync_server/app.py", + "server sync/sync_server/database.py", + "server sync/sync_server/readiness.py", + "server sync/sync_server/storage.py", + "server sync/tests/test_protocol.py", + "server sync/tests/test_production_storage.py", + "server sync/tests/test_readiness.py", + "scripts/acceptance_cases/sync_production_stack.py", + "scripts/acceptance_cases/s06_sync_security.py", + ) + ], + "revisions": [ + { + "scope": "runtime", + "postgres": facts.get("postgres_version"), + "minio": facts.get("minio_version"), + }, + { + "scope": "authorization matrix", + "hash": facts.get("hash_rejections", 0), + "revoked": facts.get("revoked_rejections", 0), + "revoked_refresh": facts.get("revoked_refresh_rejections", 0), + "expired_token": facts.get("expired_token_rejections", 0), + "foreign_upload": facts.get("foreign_upload_rejections", 0), + "expired_upload": facts.get("expired_upload_rejections", 0), + }, + { + "scope": "readiness", + "transitions": facts.get("readiness_transitions", 0), + "failure_max_ms": facts.get("ready_failure_max_ms", 0), + "recovery_max_ms": facts.get("ready_recovery_max_ms", 0), + "dependencies": facts.get("readiness_dependencies", {}), + }, + { + "scope": "login limiting", + "prelimit_401": facts.get("prelimit_rejections", 0), + "limited_429": facts.get("rate_limited_responses", 0), + "retry_after": facts.get("retry_after_responses", 0), + }, + { + "scope": "last checkpoint", + "stage": facts.get("active_stage"), + "iteration": facts.get("active_iteration", -1), + }, + ], + } + + +def auth(session: dict) -> dict: + return {"Authorization": "Bearer " + session["access_token"]} + + +def upload(base: str, headers: dict, data: bytes) -> str: + digest = hashlib.sha256(data).hexdigest() + code, pending, _ = call( + "POST", base + "/uploads", headers=headers, body={"content_hash": digest, "size": len(data)} + ) + assert code == 200 and pending["complete"] is False + path = base + "/uploads/" + pending["upload_id"] + assert call("PUT", path + "?offset=0", headers=headers, body=data)[1] == { + "offset": len(data) + } + completed = call("POST", path + "/complete", headers=headers) + assert completed[0] == 200 and completed[1]["content_hash"] == digest + return digest + + +def await_cluster_ready(origin: str, expected: int, *, timeout=8) -> int: + started = time.monotonic() + deadline = started + timeout + observed: set[str] = set() + + def probe(_): + try: + return call( + "GET", + origin + "/ready", + headers={"Connection": "close"}, + timeout=5, + ) + except (OSError, TimeoutError): + return None + + while time.monotonic() < deadline: + with ThreadPoolExecutor(max_workers=8) as pool: + samples = list(pool.map(probe, range(8))) + for sample in samples: + if sample is not None and sample[0] == expected: + worker = sample[2].get("x-opennexus-worker") + if worker: + observed.add(worker) + if len(observed) == 2: + return int((time.monotonic() - started) * 1000) + time.sleep(0.05) + raise RuntimeError("READINESS_TRANSITION_TIMEOUT") + + +def assert_error(response, status: int, code: str) -> None: + assert response[0] == status and response[1]["error"]["code"] == code + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--config", required=True) + parser.add_argument("--output", required=True) + args = parser.parse_args() + output = Path(args.output) + output.parent.mkdir(parents=True, exist_ok=True) + case_id = os.environ.get("OPENNEXUS_ACCEPTANCE_CASE_ID", "") + facts: dict = {} + reason = "" + status = "FAILED" + stack = None + staging_available = None + try: + config = json.loads(Path(args.config).read_text(encoding="utf-8")) + stack = SyncProductionStack( + config, Path(os.environ["OPENNEXUS_ACCEPTANCE_DATA_ROOT"]), "s06" + ).start() + facts.update( + { + "dependencies": True, + "postgres_version": stack.postgres_version, + "minio_version": stack.minio_version, + } + ) + + owner_session = stack.login(stack.username, stack.password, "S-06 owner") + owner_auth = auth(owner_session) + _, owner_base, owner_vault = stack.create_vault_for_auth(owner_auth, "S-06 owner") + _, sibling_base, _ = stack.create_vault_for_auth(owner_auth, "S-06 sibling") + other_username, other_password = stack.add_user("s06-other") + other_session = stack.login(other_username, other_password, "S-06 other account") + other_auth = auth(other_session) + stack.create_vault_for_auth(other_auth, "S-06 other vault") + object_data = b"OpenNexus S-06 private object" + object_hash = upload(owner_base, owner_auth, object_data) + + facts["active_stage"] = "hash-isolation" + hash_rejections = content_leaks = 0 + for index in range(ROUNDS): + facts["active_iteration"] = index + if index % 2: + response = call("GET", owner_base + "/objects/" + object_hash, headers=other_auth) + assert_error(response, 404, "VAULT_NOT_FOUND") + else: + response = call("GET", sibling_base + "/objects/" + object_hash, headers=owner_auth) + assert_error(response, 404, "OBJECT_NOT_FOUND") + hash_rejections += 1 + content_leaks += int(object_data in json.dumps(response[1]).encode()) + facts.update({"hash_rejections": hash_rejections, "content_leaks": content_leaks}) + + facts["active_stage"] = "device-revocation" + revoked = stack.login(stack.username, stack.password, "S-06 revoked device") + revoke = call( + "DELETE", + stack.origin + "/sync/v1/devices/" + revoked["device_id"], + headers=owner_auth, + ) + assert revoke[0] == 204 + revoked_auth = auth(revoked) + rejected = call("GET", owner_base + "/changes", headers=revoked_auth) + assert_error(rejected, 401, "SESSION_EXPIRED") + facts["immediate_revocation"] = True + revoked_rejections = 1 + for index in range(1, ROUNDS): + facts["active_iteration"] = index + response = call("GET", owner_base + "/changes", headers=revoked_auth) + assert_error(response, 401, "SESSION_EXPIRED") + revoked_rejections += 1 + revoked_refresh_rejections = 0 + for index in range(ROUNDS): + response = call( + "POST", + stack.origin + "/sync/v1/auth/refresh", + body={"refresh_token": revoked["refresh_token"]}, + ) + assert_error(response, 401, "SESSION_EXPIRED") + revoked_refresh_rejections += 1 + facts.update( + { + "revoked_rejections": revoked_rejections, + "revoked_refresh_rejections": revoked_refresh_rejections, + } + ) + + facts["active_stage"] = "expired-access" + expired = stack.login(stack.username, stack.password, "S-06 expired access") + access_digest = hashlib.sha256(expired["access_token"].encode()).hexdigest() + changed = stack.sql_scalar( + "WITH changed AS (UPDATE sessions SET expires=0 " + f"WHERE token='{access_digest}' RETURNING 1) SELECT COUNT(*) FROM changed" + ) + assert changed == 1 + expired_rejections = 0 + expired_auth = auth(expired) + for index in range(ROUNDS): + facts["active_iteration"] = index + response = call("GET", owner_base + "/changes", headers=expired_auth) + assert_error(response, 401, "SESSION_EXPIRED") + expired_rejections += 1 + facts["expired_token_rejections"] = expired_rejections + + facts["active_stage"] = "refresh-rotation" + rotating = stack.login(stack.username, stack.password, "S-06 refresh rotation") + rotated = call( + "POST", + stack.origin + "/sync/v1/auth/refresh", + body={"refresh_token": rotating["refresh_token"]}, + ) + assert rotated[0] == 200 + refresh_replays = 0 + for index in range(ROUNDS): + response = call( + "POST", + stack.origin + "/sync/v1/auth/refresh", + body={"refresh_token": rotating["refresh_token"]}, + ) + assert_error(response, 401, "SESSION_EXPIRED") + refresh_replays += 1 + facts["refresh_replay_rejections"] = refresh_replays + rotated_access = call("GET", owner_base + "/changes", headers=auth(rotated[1])) + facts["rotated_access_works"] = rotated_access[0] == 200 + + facts["active_stage"] = "upload-link-isolation" + upload_data = b"OpenNexus S-06 pending upload" + upload_hash = hashlib.sha256(upload_data).hexdigest() + pending = call( + "POST", + owner_base + "/uploads", + headers=owner_auth, + body={"content_hash": upload_hash, "size": len(upload_data)}, + )[1] + upload_id = pending["upload_id"] + upload_path = owner_base + "/uploads/" + upload_id + foreign = stack.login(stack.username, stack.password, "S-06 foreign device") + + def rejected_upload_request(index: int, headers: dict): + kind = index % 4 + if kind == 0: + return call("GET", upload_path, headers=headers) + if kind == 1: + return call("PUT", upload_path + "?offset=0", headers=headers, body=b"x") + if kind == 2: + return call("POST", upload_path + "/complete", headers=headers) + return call("DELETE", upload_path, headers=headers) + + foreign_upload_rejections = 0 + for index in range(ROUNDS): + facts["active_iteration"] = index + response = rejected_upload_request(index, auth(foreign)) + assert_error(response, 404, "UPLOAD_EXPIRED") + foreign_upload_rejections += 1 + expired_rows = stack.sql_scalar( + "WITH changed AS (UPDATE uploads SET expires=0 " + f"WHERE id='{upload_id}' RETURNING 1) SELECT COUNT(*) FROM changed" + ) + assert expired_rows == 1 + expired_upload_rejections = 0 + for index in range(ROUNDS): + facts["active_iteration"] = index + response = rejected_upload_request(index, owner_auth) + assert_error(response, 404, "UPLOAD_EXPIRED") + expired_upload_rejections += 1 + facts.update( + { + "foreign_upload_rejections": foreign_upload_rejections, + "expired_upload_rejections": expired_upload_rejections, + } + ) + (stack.staging / upload_id).unlink(missing_ok=True) + stack.sql_scalar( + "WITH removed AS (DELETE FROM uploads " + f"WHERE id='{upload_id}' RETURNING 1) SELECT COUNT(*) FROM removed" + ) + + facts["active_stage"] = "login-rate-limit" + limited_username = "s06-limit-" + secrets.token_hex(8) + login_url = stack.origin + "/sync/v1/auth/sessions" + login_body = { + "username": limited_username, + "password": "invalid-controlled-password", + "device_name": "S-06 limited", + } + prelimit = 0 + for index in range(10): + response = call("POST", login_url, body=login_body) + assert_error(response, 401, "LOGIN_FAILED") + prelimit += 1 + limited = retry_after = 0 + for index in range(ROUNDS): + facts["active_iteration"] = index + response = call("POST", login_url, body=login_body) + assert_error(response, 429, "RATE_LIMITED") + limited += 1 + retry_after += int(response[2].get("retry-after") == "60") + facts.update( + { + "prelimit_rejections": prelimit, + "rate_limited_responses": limited, + "retry_after_responses": retry_after, + } + ) + + facts["active_stage"] = "readiness-faults" + assert await_cluster_ready(stack.origin, 200) < 8000 + failure_latencies = [] + recovery_latencies = [] + + started = time.monotonic() + stack.stop_postgres() + failure_latencies.append( + int((time.monotonic() - started) * 1000) + await_cluster_ready(stack.origin, 503) + ) + started = time.monotonic() + stack.start_postgres() + recovery_latencies.append( + int((time.monotonic() - started) * 1000) + await_cluster_ready(stack.origin, 200) + ) + + started = time.monotonic() + stack.stop_minio() + failure_latencies.append( + int((time.monotonic() - started) * 1000) + await_cluster_ready(stack.origin, 503) + ) + started = time.monotonic() + stack.start_minio() + recovery_latencies.append( + int((time.monotonic() - started) * 1000) + await_cluster_ready(stack.origin, 200) + ) + + staging_available = stack.stack / "staging-available" + started = time.monotonic() + stack.staging.rename(staging_available) + stack.staging.write_text("unavailable", encoding="utf-8") + failure_latencies.append( + int((time.monotonic() - started) * 1000) + await_cluster_ready(stack.origin, 503) + ) + started = time.monotonic() + stack.staging.unlink() + staging_available.rename(stack.staging) + staging_available = None + recovery_latencies.append( + int((time.monotonic() - started) * 1000) + await_cluster_ready(stack.origin, 200) + ) + + facts.update( + { + "readiness_transitions": 6, + "ready_failure_max_ms": max(failure_latencies), + "ready_recovery_max_ms": max(recovery_latencies), + "readiness_dependencies": { + name: {"failure_ms": failure, "recovery_ms": recovery} + for name, failure, recovery in zip( + ("postgresql", "minio", "staging"), + failure_latencies, + recovery_latencies, + strict=True, + ) + }, + "worker_count": len(stack.worker_ids()), + "active_stage": "complete", + "active_iteration": ROUNDS, + } + ) + assert facts["worker_count"] == 2 + assert max(failure_latencies + recovery_latencies) < 8000 + assert call("GET", owner_base + "/objects/" + object_hash, headers=owner_auth)[1] == object_data + stack.assert_running() + status = "PASSED" + except BaseException as error: + reason = "S06_ORACLE_FAILED:" + type(error).__name__ + finally: + if stack is not None and staging_available is not None: + if stack.staging.is_file(): + stack.staging.unlink() + if staging_available.exists(): + staging_available.rename(stack.staging) + if stack is not None: + stack.stop() + payload = result( + case_id, + status if case_id == "S-06" else "FAILED", + reason or ("" if case_id == "S-06" else "CASE_ID_MISMATCH"), + facts, + ) + output.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + return 0 if payload["status"] == "PASSED" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/acceptance_cases/sync_production_stack.py b/scripts/acceptance_cases/sync_production_stack.py index 4277e0e..f5d26a7 100644 --- a/scripts/acceptance_cases/sync_production_stack.py +++ b/scripts/acceptance_cases/sync_production_stack.py @@ -92,6 +92,10 @@ class SyncProductionStack: self.minio = None self.sync = None self.handles = [] + self.pg_log = None + self.minio_log = None + self.sync_log = None + self.minio_env = {} initdb = Path(config["artifacts"]["postgres_initdb"]).resolve() self.pg_bin = initdb.parent @@ -146,10 +150,10 @@ class SyncProductionStack: self.stack.mkdir() self.minio_data.mkdir() self.staging.mkdir() - pg_log = (self.stack / "postgres.log").open("wb") - minio_log = (self.stack / "minio.log").open("wb") - sync_log = (self.stack / "sync.log").open("wb") - self.handles.extend([pg_log, minio_log, sync_log]) + self.pg_log = (self.stack / "postgres.log").open("wb") + self.minio_log = (self.stack / "minio.log").open("wb") + self.sync_log = (self.stack / "sync.log").open("wb") + self.handles.extend([self.pg_log, self.minio_log, self.sync_log]) subprocess.run( [ @@ -164,29 +168,12 @@ class SyncProductionStack: "-E", "UTF8", ], - stdout=pg_log, + stdout=self.pg_log, stderr=subprocess.STDOUT, check=True, timeout=120, ) - subprocess.run( - [ - str(self.pg_ctl), - "-D", - str(self.postgres_data), - "-l", - str(self.stack / "postgres-server.log"), - "-o", - f"-p {self.pg_port} -h 127.0.0.1", - "-w", - "start", - ], - stdout=pg_log, - stderr=subprocess.STDOUT, - check=True, - timeout=60, - ) - self.postgres_started = True + self.start_postgres() subprocess.run( [ str(self.createdb), @@ -198,30 +185,17 @@ class SyncProductionStack: "postgres", "opennexus", ], - stdout=pg_log, + stdout=self.pg_log, stderr=subprocess.STDOUT, check=True, timeout=30, ) - minio_env = os.environ.copy() - minio_env.update( + self.minio_env = os.environ.copy() + self.minio_env.update( {"MINIO_ROOT_USER": self.minio_user, "MINIO_ROOT_PASSWORD": self.minio_password} ) - self.minio = subprocess.Popen( - [ - str(self.minio_server), - "server", - str(self.minio_data), - "--address", - f"127.0.0.1:{self.minio_port}", - "--console-address", - f"127.0.0.1:{self.console_port}", - ], - stdout=minio_log, - stderr=subprocess.STDOUT, - env=minio_env, - ) + self.start_minio() wait_http(f"http://127.0.0.1:{self.minio_port}/minio/health/ready") self.service_env = os.environ.copy() @@ -251,7 +225,7 @@ class SyncProductionStack: ], cwd=SERVICE, env=self.service_env, - stdout=sync_log, + stdout=self.sync_log, stderr=subprocess.STDOUT, check=True, timeout=60, @@ -260,13 +234,13 @@ class SyncProductionStack: [str(self.server_python), "-m", "sync_server", "serve", "--workers", "2"], cwd=SERVICE, env=self.service_env, - stdout=sync_log, + stdout=self.sync_log, stderr=subprocess.STDOUT, ) wait_http(self.origin + "/ready", timeout=60) return self - def create_vault(self, label: str): + def login(self, username: str, password: str, label: str) -> dict: session = None for _ in range(5): try: @@ -274,8 +248,8 @@ class SyncProductionStack: "POST", self.origin + "/sync/v1/auth/sessions", body={ - "username": self.username, - "password": self.password, + "username": username, + "password": password, "device_name": label, }, timeout=10, @@ -288,7 +262,9 @@ class SyncProductionStack: time.sleep(0.2) if session is None: raise RuntimeError("ACCEPTANCE_LOGIN_FAILED") - auth = {"Authorization": "Bearer " + session["access_token"]} + return session + + def create_vault_for_auth(self, auth: dict, label: str): vault = None for _ in range(5): try: @@ -309,6 +285,89 @@ class SyncProductionStack: raise RuntimeError("ACCEPTANCE_VAULT_FAILED") return auth, self.origin + "/sync/v1/vaults/" + vault["vault_id"], vault["vault_id"] + def create_vault(self, label: str): + session = self.login(self.username, self.password, label) + auth = {"Authorization": "Bearer " + session["access_token"]} + return self.create_vault_for_auth(auth, label) + + def add_user(self, prefix: str) -> tuple[str, str]: + username = prefix + "-" + secrets.token_hex(8) + password = secrets.token_urlsafe(32) + environment = dict(self.service_env) + environment.update( + {"ACCEPTANCE_EXTRA_USERNAME": username, "ACCEPTANCE_EXTRA_PASSWORD": password} + ) + subprocess.run( + [ + str(self.server_python), + "-c", + "import os; from sync_server.database import Database; " + "Database(os.environ['SYNC_DATABASE_URL']).add_user(" + "os.environ['ACCEPTANCE_EXTRA_USERNAME'],os.environ['ACCEPTANCE_EXTRA_PASSWORD'])", + ], + cwd=SERVICE, + env=environment, + stdout=self.sync_log, + stderr=subprocess.STDOUT, + check=True, + timeout=30, + ) + return username, password + + def start_postgres(self) -> None: + subprocess.run( + [ + str(self.pg_ctl), + "-D", + str(self.postgres_data), + "-l", + str(self.stack / "postgres-server.log"), + "-o", + f"-p {self.pg_port} -h 127.0.0.1", + "-w", + "start", + ], + stdout=self.pg_log, + stderr=subprocess.STDOUT, + check=True, + timeout=60, + ) + self.postgres_started = True + + def stop_postgres(self) -> None: + if not self.postgres_started: + return + subprocess.run( + [str(self.pg_ctl), "-D", str(self.postgres_data), "-m", "fast", "-w", "stop"], + stdout=self.pg_log, + stderr=subprocess.STDOUT, + check=True, + timeout=30, + ) + self.postgres_started = False + + def start_minio(self) -> None: + if self.minio is not None and self.minio.poll() is None: + return + self.minio = subprocess.Popen( + [ + str(self.minio_server), + "server", + str(self.minio_data), + "--address", + f"127.0.0.1:{self.minio_port}", + "--console-address", + f"127.0.0.1:{self.console_port}", + ], + stdout=self.minio_log, + stderr=subprocess.STDOUT, + env=self.minio_env, + ) + + def stop_minio(self) -> None: + stop_tree(self.minio) + self.minio = None + def sql_scalar(self, statement: str) -> int: output = subprocess.check_output( [ @@ -361,22 +420,11 @@ class SyncProductionStack: def stop(self) -> None: stop_tree(self.sync) - stop_tree(self.minio) + self.stop_minio() if self.postgres_started: - subprocess.run( - [ - str(self.pg_ctl), - "-D", - str(self.postgres_data), - "-m", - "fast", - "-w", - "stop", - ], - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - check=False, - timeout=30, - ) + try: + self.stop_postgres() + except (OSError, subprocess.SubprocessError): + pass for handle in self.handles: handle.close() diff --git a/scripts/phase3_acceptance.py b/scripts/phase3_acceptance.py index 6e0c7a2..e016a11 100644 --- a/scripts/phase3_acceptance.py +++ b/scripts/phase3_acceptance.py @@ -95,6 +95,21 @@ CASE_DRIVERS: dict[str, dict[str, Any]] = { ), "required_artifacts": ("postgres_initdb", "minio_server"), }, + "S-06": { + "driver": "scripts/acceptance_cases/s06_sync_security.py", + "timeout_seconds": 900, + "required_metrics": ( + "hash_rejections", + "revoked_rejections", + "expired_token_rejections", + "upload_link_rejections", + "rate_limited_responses", + "ready_failure_max_ms", + "ready_recovery_max_ms", + "worker_count", + ), + "required_artifacts": ("postgres_initdb", "minio_server"), + }, } ENV_NAME = re.compile(r"[A-Z][A-Z0-9_]{2,127}") RUN_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{2,63}") diff --git a/server sync/sync_server/app.py b/server sync/sync_server/app.py index b790c39..50bdf0a 100644 --- a/server sync/sync_server/app.py +++ b/server sync/sync_server/app.py @@ -73,6 +73,8 @@ def create_app(db: Database, objects, staging: Path, *, quota=1024**3, clock=tim @app.exception_handler(SyncError) async def error(_request, exc): headers = {"Retry-After": "60"} if exc.status == 429 else {} + if exc.status == 503: + headers["X-OpenNexus-Worker"] = worker_id return JSONResponse({"error": {"code": exc.code, "details": exc.details}}, status_code=exc.status, headers=headers) @app.exception_handler(RequestValidationError) @@ -116,6 +118,7 @@ def create_app(db: Database, objects, staging: Path, *, quota=1024**3, clock=tim if row(conn, "SELECT version FROM schema_version")["version"] != 1: raise SyncError(503, "SCHEMA_INCOMPATIBLE") key = "health-probe/" + secrets.token_hex(16) + stored = False try: with tempfile.TemporaryFile(dir=staging) as local: local.write(b"opennexus-ready") @@ -125,17 +128,20 @@ def create_app(db: Database, objects, staging: Path, *, quota=1024**3, clock=tim if local.read() != b"opennexus-ready": raise OSError("STAGING_INTEGRITY") objects.put(key, b"opennexus-ready") + stored = True if objects.get(key) != b"opennexus-ready": raise OSError("STORAGE_INTEGRITY") finally: - objects.delete(key) + if stored: + objects.delete(key) - readiness = Readiness(readiness_probe) + readiness = Readiness(readiness_probe, cache_seconds=1) @app.get("/ready") - async def ready(): + async def ready(response: Response): if not await readiness.check(): raise SyncError(503, "DEPENDENCY_UNAVAILABLE") + response.headers["X-OpenNexus-Worker"] = worker_id return {"status": "ready", "schema": 1} @app.get("/sync/v1/handshake") diff --git a/server sync/tests/test_production_storage.py b/server sync/tests/test_production_storage.py index 7134b58..341677c 100644 --- a/server sync/tests/test_production_storage.py +++ b/server sync/tests/test_production_storage.py @@ -105,7 +105,9 @@ def test_readiness_fails_closed_when_object_storage_unavailable(env): def unavailable(*args): raise OSError("simulated outage") store.put = unavailable - assert client.get("/ready").status_code == 503 + unavailable = client.get("/ready") + assert unavailable.status_code == 503 + assert len(unavailable.headers["X-OpenNexus-Worker"]) == 16 assert client.get("/health").status_code == 200