From 2b814f05a2f7cc74da576b76791bec5f8918b0a9 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 13 Sep 2026 20:43:38 +0800 Subject: [PATCH] fix(binance): release stale owner under bounded preconditions Co-Authored-By: Codex --- .github/workflows/main.yml | 4 +- scripts/migrate_daily_accounting_state.py | 156 ++++++++++++++- tests/test_daily_accounting_migration.py | 232 ++++++++++++++++++++++ tests/test_runtime_workflow_security.py | 4 +- 4 files changed, 390 insertions(+), 6 deletions(-) diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 470573f8..9e9f321d 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -52,7 +52,7 @@ on: type: choice required: false default: none - options: [none, inspect, quiesce, earn-forward-diagnose, audit, preview, scope-preview, cash-flow-preview, rebase-proposal, rebase-apply, prospective-rebase-apply, apply] + options: [none, inspect, quiesce, earn-forward-diagnose, audit, preview, scope-preview, cash-flow-preview, rebase-proposal, rebase-apply, prospective-rebase-apply, release-stale-owner, apply] proposal_recipient_certificate: description: "Public X.509 encryption certificate for rebase-proposal only; never provide a private key" type: string @@ -132,7 +132,7 @@ jobs: exit 1 fi ;; - inspect|quiesce|earn-forward-diagnose|audit|preview|scope-preview|cash-flow-preview|rebase-proposal|rebase-apply|prospective-rebase-apply|apply) + inspect|quiesce|earn-forward-diagnose|audit|preview|scope-preview|cash-flow-preview|rebase-proposal|rebase-apply|prospective-rebase-apply|release-stale-owner|apply) if [ "${RECOVERY_ACTION_INPUT:-none}" != "none" ] || [ "${GITHUB_REF:-}" != "refs/heads/main" ] || [ "${RUNTIME_TARGET_ENABLED:-}" != "false" ] || [ "$RECONCILE_ONLY_INPUT" != "true" ] || [ "$VALIDATE_ONLY_INPUT" = "true" ] || [ "$DIAGNOSE_BALANCES_INPUT" = "true" ] || [ "$RECONCILE_PERSIST_INPUT" = "true" ]; then echo "::error::Accounting migration requires disabled main runtime, reconcile_only, and no other recovery or diagnostic mode." exit 1 diff --git a/scripts/migrate_daily_accounting_state.py b/scripts/migrate_daily_accounting_state.py index b6436f18..7526cd5c 100644 --- a/scripts/migrate_daily_accounting_state.py +++ b/scripts/migrate_daily_accounting_state.py @@ -994,6 +994,141 @@ def _read_source(refs): return ledger, ledger_value, control_value +def _snapshot_marker(snapshot): + return { + "exists": bool(snapshot.exists), + "update_time": _timestamp(snapshot.update_time) if snapshot.exists else None, + "value": snapshot.to_dict() if snapshot.exists else None, + } + + +def _snapshot_matches(snapshot, marker): + return _snapshot_marker(snapshot) == marker + + +def _validate_stale_owner_control(target, control, *, expected): + if not isinstance(control, Mapping) or control.get("state") != "ACTIVE_LKG": + raise MigrationBlocked("recovery_control_not_active_lkg") + try: + from application.reconciliation_recovery import activated_target + + activated = activated_target(target, control, expected=expected) + if str(getattr(getattr(activated, "live_continuity", None), "state", "")).upper() != "ACTIVE_LKG": + raise MigrationBlocked("recovery_control_not_active_lkg") + except MigrationBlocked: + raise + except Exception: + raise MigrationBlocked("recovery_control_invalid") from None + + +def _release_stale_owner_transaction(transaction, *, refs, markers): + owner = refs["owner_ref"].get(transaction=transaction, retry=None) + ledger = refs["ledger_ref"].get(transaction=transaction, retry=None) + control = refs["control_ref"].get(transaction=transaction, retry=None) + if ( + not _snapshot_matches(owner, markers["owner"]) + or not _snapshot_matches(ledger, markers["ledger"]) + or not _snapshot_matches(control, markers["control"]) + ): + raise MigrationAtomicPrecondition("stale_owner_release_precondition_changed") + owner_value = owner.to_dict() + if not isinstance(owner_value, Mapping): + raise MigrationAtomicPrecondition("stale_owner_release_precondition_changed") + owner_id = owner_value.get("owner_id") + if not isinstance(owner_id, str) or not owner_id.strip(): + raise MigrationAtomicPrecondition("stale_owner_release_precondition_changed") + transaction.delete(refs["owner_ref"]) + return None + + +def release_stale_owner(refs, *, client, target, expected): + """Delete one confirmed stale owner document and nothing else.""" + owner_snapshot = refs["owner_ref"].get(retry=None) + if not owner_snapshot.exists: + return { + "status": "already_absent", + "stage": "stale_owner_release", + "owner_exists": False, + "no_order": True, + "write_performed": False, + } + owner_value = owner_snapshot.to_dict() + if not isinstance(owner_value, Mapping): + raise MigrationBlocked("owner_document_invalid") + owner_id = owner_value.get("owner_id") + if not isinstance(owner_id, str) or not owner_id.strip(): + raise MigrationBlocked("owner_id_invalid") + ledger_snapshot = refs["ledger_ref"].get(retry=None) + control_snapshot = refs["control_ref"].get(retry=None) + if not ledger_snapshot.exists or not control_snapshot.exists: + raise MigrationBlocked("stale_owner_release_source_missing") + ledger = ledger_snapshot.to_dict() + control = control_snapshot.to_dict() + if not isinstance(ledger, Mapping): + raise MigrationBlocked("ledger_unavailable") + _validate_safe_order_state(ledger) + _validate_stale_owner_control(target, control, expected=expected) + if client is None: + raise MigrationBlocked("account_scope_unverified") + try: + account = client.get_account() + except Exception: + raise MigrationBlocked("account_read_failed") from None + _private_spot_account( + account, expected_account_scope_sha256=expected["account_scope_sha256"] + ) + try: + open_orders = client.get_open_orders() + except Exception: + raise MigrationBlocked("open_orders_read_failed") from None + if not isinstance(open_orders, list): + raise MigrationBlocked("open_orders_unverified") + if open_orders: + raise MigrationBlocked("open_orders_present") + markers = { + "owner": _snapshot_marker(owner_snapshot), + "ledger": _snapshot_marker(ledger_snapshot), + "control": _snapshot_marker(control_snapshot), + } + from google.cloud import firestore + + @firestore.transactional + def apply(transaction): + return _release_stale_owner_transaction(transaction, refs=refs, markers=markers) + + try: + apply(get_firestore_client().transaction(max_attempts=1)) + except MigrationAtomicPrecondition: + raise + except Exception: + raise MigrationApplyUncertain("stale_owner_release_outcome_uncertain") from None + try: + owner_after = refs["owner_ref"].get(retry=None) + ledger_after = refs["ledger_ref"].get(retry=None) + control_after = refs["control_ref"].get(retry=None) + if ( + owner_after.exists + or not _snapshot_matches(ledger_after, markers["ledger"]) + or not _snapshot_matches(control_after, markers["control"]) + ): + raise MigrationApplyUncertain("stale_owner_release_readback_uncertain") + except MigrationApplyUncertain: + raise + except Exception: + raise MigrationApplyUncertain("stale_owner_release_readback_uncertain") from None + return { + "status": "released", + "stage": "stale_owner_release", + "owner_exists": False, + "owner_released": True, + "ledger_unchanged": True, + "control_unchanged": True, + "no_order": True, + "write_performed": True, + "execution_authority_granted": False, + } + + def _read_earn_diagnosis_source(refs): """Read the three accounting documents without treating an owner as a blocker.""" owner = refs["owner_ref"].get(retry=None) @@ -2331,6 +2466,16 @@ def run(action: str, *, expected_digest: str = "", now: datetime | None = None): if not expected: raise MigrationBlocked("expected_account_scope_missing") refs = _refs() + if action == "release-stale-owner": + owner_snapshot = refs["owner_ref"].get(retry=None) + client = None + if owner_snapshot.exists: + client = connect_client( + os.environ["BINANCE_API_KEY"], os.environ["BINANCE_API_SECRET"], timeout=30 + ) + return release_stale_owner( + refs, client=client, target=target, expected=expected + ) ledger_snapshot, ledger, control = _read_source(refs) client = connect_client( os.environ["BINANCE_API_KEY"], os.environ["BINANCE_API_SECRET"], timeout=30 @@ -2463,6 +2608,7 @@ def main(argv=None) -> int: "rebase-proposal", "rebase-apply", "prospective-rebase-apply", + "release-stale-owner", "apply", ), ) @@ -2481,9 +2627,13 @@ def main(argv=None) -> int: json.dumps( { "status": "uncertain", - "stage": "private_scope_publication" - if args.action == "scope-preview" - else "accounting_migration_apply", + "stage": ( + "private_scope_publication" + if args.action == "scope-preview" + else "stale_owner_release" + if args.action == "release-stale-owner" + else "accounting_migration_apply" + ), "reason_code": str(exc), "no_retry": True, "no_order": True, diff --git a/tests/test_daily_accounting_migration.py b/tests/test_daily_accounting_migration.py index 860bbd55..978e74db 100644 --- a/tests/test_daily_accounting_migration.py +++ b/tests/test_daily_accounting_migration.py @@ -193,6 +193,11 @@ def __init__(self): def update(self, ref, patch): self.writes.append((ref, copy.deepcopy(patch))) + def delete(self, ref): + self.writes.append(("delete", ref)) + ref.snapshot.value = None + ref.snapshot.exists = False + def test_atomic_apply_updates_only_allowlisted_fields(): from scripts.migrate_daily_accounting_state import compare_and_apply @@ -226,6 +231,233 @@ def test_atomic_apply_updates_only_allowlisted_fields(): assert "order_submission" not in tx.writes[0][1] +def _stale_owner_refs(*, owner=None, ledger=None, control=None): + return { + "owner_ref": Ref(Snapshot(owner if owner is not None else {"owner_id": "a" * 32})), + "ledger_ref": Ref(Snapshot(ledger or _ledger())), + "control_ref": Ref(Snapshot(control or {"state": "ACTIVE_LKG"})), + } + + +def test_release_stale_owner_transaction_deletes_only_owner(): + from scripts.migrate_daily_accounting_state import ( + _release_stale_owner_transaction, + _snapshot_marker, + ) + + refs = _stale_owner_refs() + markers = {name: _snapshot_marker(refs[f"{name}_ref"].snapshot) + for name in ("owner", "ledger", "control")} + tx = Transaction() + + result = _release_stale_owner_transaction(tx, refs=refs, markers=markers) + + assert result is None + assert tx.writes == [("delete", refs["owner_ref"])] + + +@pytest.mark.parametrize("changed", ["owner", "ledger", "control"]) +def test_release_stale_owner_transaction_rejects_concurrent_document_change(changed): + from scripts.migrate_daily_accounting_state import ( + MigrationAtomicPrecondition, + _release_stale_owner_transaction, + _snapshot_marker, + ) + + refs = _stale_owner_refs() + markers = {name: _snapshot_marker(refs[f"{name}_ref"].snapshot) + for name in ("owner", "ledger", "control")} + ref = refs[f"{changed}_ref"] + ref.snapshot.value = {**(ref.snapshot.value or {}), "changed": True} + tx = Transaction() + + with pytest.raises(MigrationAtomicPrecondition, match="stale_owner_release_precondition_changed"): + _release_stale_owner_transaction(tx, refs=refs, markers=markers) + assert tx.writes == [] + + +def test_release_stale_owner_returns_already_absent_without_broker_or_delete(monkeypatch): + from scripts import migrate_daily_accounting_state as migration + + refs = _stale_owner_refs(owner=None) + refs["owner_ref"] = Ref(Snapshot(None)) + client = SimpleNamespace( + get_account=lambda: (_ for _ in ()).throw(AssertionError("broker must not be read")), + get_open_orders=lambda: (_ for _ in ()).throw(AssertionError("broker must not be read")), + ) + + result = migration.release_stale_owner( + refs, client=client, target=SimpleNamespace(), expected={"account_scope_sha256": "a" * 64} + ) + + assert result == { + "status": "already_absent", + "stage": "stale_owner_release", + "owner_exists": False, + "no_order": True, + "write_performed": False, + } + + +@pytest.mark.parametrize( + "ledger,reason", + [ + (_ledger(order_submission={"state": "SUBMISSION_UNKNOWN"}), "unsafe_order_state"), + (_ledger(order_submission={"state": "TERMINAL", "funding_receipt": {}}), "unsafe_order_state"), + ], +) +def test_release_stale_owner_rejects_unknown_or_funding_receipt(ledger, reason, monkeypatch): + from scripts import migrate_daily_accounting_state as migration + + refs = _stale_owner_refs(ledger=ledger) + monkeypatch.setattr(migration, "_validate_stale_owner_control", lambda *args, **kwargs: None) + client = SimpleNamespace( + get_account=lambda: {"uid": "synthetic", "balances": []}, + get_open_orders=lambda: [], + ) + + with pytest.raises(migration.MigrationBlocked, match=reason): + migration.release_stale_owner( + refs, client=client, target=SimpleNamespace(), expected={"account_scope_sha256": "a" * 64} + ) + + +@pytest.mark.parametrize("orders", [[{"symbol": "BTCUSDT"}], None]) +def test_release_stale_owner_rejects_nonempty_or_invalid_open_orders(orders, monkeypatch): + from scripts import migrate_daily_accounting_state as migration + + refs = _stale_owner_refs() + monkeypatch.setattr(migration, "_validate_stale_owner_control", lambda *args, **kwargs: None) + monkeypatch.setattr(migration, "_private_spot_account", lambda *args, **kwargs: ("uid", ())) + client = SimpleNamespace( + get_account=lambda: {"uid": "synthetic", "balances": []}, + get_open_orders=lambda: orders, + ) + + with pytest.raises(migration.MigrationBlocked, match="open_orders_"): + migration.release_stale_owner( + refs, client=client, target=SimpleNamespace(), expected={"account_scope_sha256": "a" * 64} + ) + + +def test_release_stale_owner_rejects_account_scope_mismatch(monkeypatch): + from scripts import migrate_daily_accounting_state as migration + + refs = _stale_owner_refs() + monkeypatch.setattr(migration, "_validate_stale_owner_control", lambda *args, **kwargs: None) + def mismatch(*args, **kwargs): + raise migration.MigrationBlocked("account_scope_unverified") + monkeypatch.setattr(migration, "_private_spot_account", mismatch) + client = SimpleNamespace( + get_account=lambda: {"uid": "synthetic", "balances": []}, + get_open_orders=lambda: [], + ) + + with pytest.raises(migration.MigrationBlocked, match="account_scope_unverified"): + migration.release_stale_owner( + refs, client=client, target=SimpleNamespace(), expected={"account_scope_sha256": "a" * 64} + ) + + +def _configure_stale_owner_release(monkeypatch, migration): + import google.cloud.firestore + + monkeypatch.setattr(migration, "_validate_stale_owner_control", lambda *args, **kwargs: None) + monkeypatch.setattr(migration, "_private_spot_account", lambda *args, **kwargs: ("uid", ())) + monkeypatch.setattr(google.cloud.firestore, "transactional", lambda fn: fn, raising=False) + + +def test_release_stale_owner_success_deletes_only_owner_and_reads_back(monkeypatch): + from scripts import migrate_daily_accounting_state as migration + + refs = _stale_owner_refs() + _configure_stale_owner_release(monkeypatch, migration) + tx = Transaction() + firestore_client = SimpleNamespace(transaction=lambda *, max_attempts: tx) + monkeypatch.setattr(migration, "get_firestore_client", lambda: firestore_client) + client = SimpleNamespace( + get_account=lambda: {"uid": "synthetic", "balances": []}, + get_open_orders=lambda: [], + ) + + result = migration.release_stale_owner( + refs, client=client, target=SimpleNamespace(), expected={"account_scope_sha256": "a" * 64} + ) + + assert result["status"] == "released" + assert result["ledger_unchanged"] is True + assert result["control_unchanged"] is True + assert result["execution_authority_granted"] is False + assert tx.writes == [("delete", refs["owner_ref"])] + + +def test_release_stale_owner_transaction_outcome_is_uncertain_without_retry(monkeypatch): + from scripts import migrate_daily_accounting_state as migration + + refs = _stale_owner_refs() + _configure_stale_owner_release(monkeypatch, migration) + calls = [] + def transaction(*, max_attempts): + calls.append(max_attempts) + raise TimeoutError("synthetic transaction timeout") + monkeypatch.setattr(migration, "get_firestore_client", lambda: SimpleNamespace(transaction=transaction)) + client = SimpleNamespace( + get_account=lambda: {"uid": "synthetic", "balances": []}, + get_open_orders=lambda: [], + ) + + with pytest.raises(migration.MigrationApplyUncertain, match="stale_owner_release_outcome_uncertain"): + migration.release_stale_owner( + refs, client=client, target=SimpleNamespace(), expected={"account_scope_sha256": "a" * 64} + ) + assert calls == [1] + + +def test_release_stale_owner_readback_uncertain_does_not_retry(monkeypatch): + from scripts import migrate_daily_accounting_state as migration + + class NoReadbackDeleteTransaction(Transaction): + def delete(self, ref): + self.writes.append(("delete", ref)) + + refs = _stale_owner_refs() + _configure_stale_owner_release(monkeypatch, migration) + tx = NoReadbackDeleteTransaction() + monkeypatch.setattr(migration, "get_firestore_client", lambda: SimpleNamespace( + transaction=lambda *, max_attempts: tx + )) + client = SimpleNamespace( + get_account=lambda: {"uid": "synthetic", "balances": []}, + get_open_orders=lambda: [], + ) + + with pytest.raises(migration.MigrationApplyUncertain, match="stale_owner_release_readback_uncertain"): + migration.release_stale_owner( + refs, client=client, target=SimpleNamespace(), expected={"account_scope_sha256": "a" * 64} + ) + assert tx.writes == [("delete", refs["owner_ref"])] + + +def test_release_stale_owner_action_is_wired_without_generic_source_read(monkeypatch, capsys): + from scripts import migrate_daily_accounting_state as migration + + refs = _stale_owner_refs(owner=None) + refs["owner_ref"] = Ref(Snapshot(None)) + monkeypatch.setattr(migration, "require_runtime_context", lambda: None) + monkeypatch.setattr( + migration, + "resolve_runtime_target_from_env", + lambda **kwargs: SimpleNamespace(live_continuity=SimpleNamespace(state="RECONCILE_ONLY")), + ) + monkeypatch.setattr(migration, "_expected_digests", lambda: {"account_scope_sha256": "a" * 64}) + monkeypatch.setattr(migration, "_refs", lambda: refs) + monkeypatch.setattr(migration, "connect_client", lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("broker must not be connected"))) + monkeypatch.setattr(migration.os, "environ", {"GITHUB_SHA": "f" * 40}) + + assert migration.main(["release-stale-owner"]) == 0 + assert json.loads(capsys.readouterr().out)["status"] == "already_absent" + + def test_apply_readback_confirms_patch_and_preserved_fields(): from scripts.migrate_daily_accounting_state import _verify_applied diff --git a/tests/test_runtime_workflow_security.py b/tests/test_runtime_workflow_security.py index 6574920f..91ce0232 100644 --- a/tests/test_runtime_workflow_security.py +++ b/tests/test_runtime_workflow_security.py @@ -333,7 +333,7 @@ def test_accounting_migration_has_explicit_preview_and_apply_inputs(): inputs = workflow[workflow.index(" inputs:") : workflow.index("permissions:")] assert "accounting_migration_action:" in inputs - assert "options: [none, inspect, quiesce, earn-forward-diagnose, audit, preview, scope-preview, cash-flow-preview, rebase-proposal, rebase-apply, prospective-rebase-apply, apply]" in inputs + assert "options: [none, inspect, quiesce, earn-forward-diagnose, audit, preview, scope-preview, cash-flow-preview, rebase-proposal, rebase-apply, prospective-rebase-apply, release-stale-owner, apply]" in inputs assert "accounting_migration_preview_run_id:" in inputs assert "accounting_migration_expected_digest:" in inputs @@ -363,6 +363,8 @@ def test_accounting_migration_has_explicit_preview_and_apply_inputs(): ("rebase-apply", "", "", "false", "false", "refs/heads/main", False), ("rebase-apply", "12345", "", "true", "false", "refs/heads/main", False), ("quiesce", "", "", "true", "false", "refs/heads/main", True), + ("release-stale-owner", "", "", "true", "false", "refs/heads/main", True), + ("release-stale-owner", "", "", "false", "false", "refs/heads/main", False), ("quiesce", "12345", "", "true", "false", "refs/heads/main", False), ("quiesce", "", "", "false", "false", "refs/heads/main", False), ("quiesce", "", "", "true", "true", "refs/heads/main", False),