From 834211ba480ae88fbbb21f1fda6fdf830c863b0d Mon Sep 17 00:00:00 2001 From: changshan Date: Mon, 20 Jul 2026 19:09:44 +0800 Subject: [PATCH] Align Go and Python prekey publish operation IDs with payload semantics Go derives the publish operation ID from the complete payload: the stable bundle ID plus a digest of the sorted OPK sidecar key IDs, so exact retries reuse the ID while a replenished OPK sidecar produces a new one instead of tripping the server idempotency check. Python now requires an explicit operation_id in PrekeyManager.publish_prekey_bundle, mirroring the Rust helper change in #54; internal callers pass op-publish-, which identifies the bundle-only payload. --- anp/direct_e2ee/client.py | 2 +- anp/direct_e2ee/prekey_manager.py | 14 ++-- anp/unittest/direct_e2ee/test_direct_e2ee.py | 25 +++++++ golang/direct_e2ee/direct_e2ee_test.go | 72 ++++++++++++++++++++ golang/direct_e2ee/prekey_manager.go | 23 ++++++- 5 files changed, 130 insertions(+), 6 deletions(-) diff --git a/anp/direct_e2ee/client.py b/anp/direct_e2ee/client.py index 18b9b7b..2077be8 100644 --- a/anp/direct_e2ee/client.py +++ b/anp/direct_e2ee/client.py @@ -65,7 +65,7 @@ def __init__( def publish_prekey_bundle(self) -> Dict[str, Any]: bundle = self._prekey_manager.ensure_fresh_prekey_bundle() - return self._prekey_manager.publish_prekey_bundle(bundle) + return self._prekey_manager.publish_prekey_bundle(bundle, f"op-publish-{bundle.bundle_id}") def ensure_fresh_prekey_bundle(self) -> PrekeyBundle: return self._prekey_manager.ensure_fresh_prekey_bundle() diff --git a/anp/direct_e2ee/prekey_manager.py b/anp/direct_e2ee/prekey_manager.py index a964a72..bbae959 100644 --- a/anp/direct_e2ee/prekey_manager.py +++ b/anp/direct_e2ee/prekey_manager.py @@ -104,7 +104,13 @@ def build_prekey_bundle( proof=signed_bundle["proof"], ) - def publish_prekey_bundle(self, bundle: PrekeyBundle) -> Dict[str, Any]: + def publish_prekey_bundle(self, bundle: PrekeyBundle, operation_id: str) -> Dict[str, Any]: + """Publish a signed prekey bundle over the RPC boundary. + + ``operation_id`` identifies the complete publish payload: reuse it + when retrying the same payload and provide a new one when the payload + changes. + """ if self._rpc_client is None: raise DirectE2eeError("RPC client is not configured", "rpc_unavailable") return self._rpc_client( @@ -115,7 +121,7 @@ def publish_prekey_bundle(self, bundle: PrekeyBundle) -> Dict[str, Any]: "profile": "anp.direct.e2ee.v1", "security_profile": "transport-protected", "sender_did": self._local_did, - "operation_id": f"op-publish-{bundle.bundle_id}", + "operation_id": operation_id, }, "body": { "prekey_bundle": bundle.to_dict(), @@ -132,12 +138,12 @@ def ensure_fresh_prekey_bundle(self) -> PrekeyBundle: ) bundle = self.build_prekey_bundle(signed_prekey) if self._rpc_client is not None: - self.publish_prekey_bundle(bundle) + self.publish_prekey_bundle(bundle, f"op-publish-{bundle.bundle_id}") return bundle _, metadata = latest bundle = self.build_prekey_bundle(metadata) if self._rpc_client is not None: - self.publish_prekey_bundle(bundle) + self.publish_prekey_bundle(bundle, f"op-publish-{bundle.bundle_id}") return bundle @staticmethod diff --git a/anp/unittest/direct_e2ee/test_direct_e2ee.py b/anp/unittest/direct_e2ee/test_direct_e2ee.py index d06ae99..76fd81a 100644 --- a/anp/unittest/direct_e2ee/test_direct_e2ee.py +++ b/anp/unittest/direct_e2ee/test_direct_e2ee.py @@ -95,6 +95,31 @@ def test_prekey_bundle_round_trip(tmp_path: Path) -> None: PrekeyManager.verify_prekey_bundle(bundle, bob_doc) +def test_publish_prekey_bundle_uses_explicit_operation_id(tmp_path: Path) -> None: + bob_doc, bob_keys = _build_identity("b.example", ["agents", "bob"]) + bob_did = bob_doc["id"] + bob_signing_key, _ = _load_identity_keys(bob_keys) + rpc = FakeRpcClient({}) + manager = PrekeyManager( + local_did=bob_did, + static_key_agreement_id=f"{bob_did}#key-3", + signing_private_key=bob_signing_key, + signing_verification_method=f"{bob_did}#key-1", + signed_prekey_store=FileSignedPrekeyStore(tmp_path / "spk"), + rpc_client=rpc, + ) + _, signed_prekey = manager.generate_signed_prekey("spk-bob-001", "2026-04-07T00:00:00Z") + bundle = manager.build_prekey_bundle(signed_prekey, bundle_id="bundle-bob-001") + + manager.publish_prekey_bundle(bundle, "op-publish-custom-001") + publish_params = rpc.calls[-1][1] + assert publish_params["meta"]["operation_id"] == "op-publish-custom-001" + + ensured = manager.ensure_fresh_prekey_bundle() + publish_params = rpc.calls[-1][1] + assert publish_params["meta"]["operation_id"] == f"op-publish-{ensured.bundle_id}" + + def test_session_init_and_follow_up_round_trip(tmp_path: Path) -> None: alice_doc, alice_keys = _build_identity("a.example", ["agents", "alice"]) bob_doc, bob_keys = _build_identity("b.example", ["agents", "bob"]) diff --git a/golang/direct_e2ee/direct_e2ee_test.go b/golang/direct_e2ee/direct_e2ee_test.go index 86427df..1abf123 100644 --- a/golang/direct_e2ee/direct_e2ee_test.go +++ b/golang/direct_e2ee/direct_e2ee_test.go @@ -131,6 +131,76 @@ func TestPublishPrekeyBundleDoesNotStripOneTimePrekeysOnFailure(t *testing.T) { } } +func TestPublishPrekeyBundleOperationIDIdentifiesCompletePayload(t *testing.T) { + bobBundle, err := authentication.CreateDidWBADocument("b.example", didOptions("b.example", "bob")) + if err != nil { + t.Fatalf("CreateDidWBADocument failed: %v", err) + } + bobDID := stringValue(bobBundle.DidDocument["id"]) + signingPrivateKey, err := anp.PrivateKeyFromPEM(bobBundle.Keys[authentication.VMKeyAuth].PrivateKeyPEM) + if err != nil { + t.Fatalf("PrivateKeyFromPEM failed: %v", err) + } + signedPrekeyStore, err := NewFileSignedPrekeyStore(filepath.Join(t.TempDir(), "spk")) + if err != nil { + t.Fatalf("NewFileSignedPrekeyStore failed: %v", err) + } + oneTimePrekeyStore, err := NewFileOneTimePrekeyStore(filepath.Join(t.TempDir(), "opk")) + if err != nil { + t.Fatalf("NewFileOneTimePrekeyStore failed: %v", err) + } + serviceDID, err := messageServiceDIDFromDocument(bobBundle.DidDocument) + if err != nil { + t.Fatalf("messageServiceDIDFromDocument failed: %v", err) + } + rpc := &fakeRPCClient{expectedServiceDID: serviceDID} + manager := NewPrekeyManager(bobDID, serviceDID, bobDID+"#"+authentication.VMKeyE2EEAgreement, signingPrivateKey, bobDID+"#"+authentication.VMKeyAuth, signedPrekeyStore, rpc.Call, oneTimePrekeyStore) + _, signedPrekey, err := manager.GenerateSignedPrekey("spk-bob-001", "2026-04-07T00:00:00Z") + if err != nil { + t.Fatalf("GenerateSignedPrekey failed: %v", err) + } + bundle, err := manager.BuildPrekeyBundle(signedPrekey, "bundle-bob-001", "") + if err != nil { + t.Fatalf("BuildPrekeyBundle failed: %v", err) + } + + // A bundle-only payload derives its operation ID from the bundle ID alone. + if _, err := manager.PublishPrekeyBundle(bundle); err != nil { + t.Fatalf("PublishPrekeyBundle failed: %v", err) + } + if got := rpc.publishOperationIDs[0]; got != "op-publish-bundle-bob-001" { + t.Fatalf("operation_id without OPKs = %q, want op-publish-bundle-bob-001", got) + } + + // With an OPK sidecar, exact retries of the same payload reuse the ID. + if _, err := manager.EnsureFreshOneTimePrekeys(2); err != nil { + t.Fatalf("EnsureFreshOneTimePrekeys failed: %v", err) + } + for i := 0; i < 2; i++ { + if _, err := manager.PublishPrekeyBundle(bundle); err != nil { + t.Fatalf("PublishPrekeyBundle with OPKs failed: %v", err) + } + } + withOPKs := rpc.publishOperationIDs[1] + if !strings.HasPrefix(withOPKs, "op-publish-bundle-bob-001-opks-") { + t.Fatalf("operation_id with OPKs = %q, want op-publish-bundle-bob-001-opks-*", withOPKs) + } + if rpc.publishOperationIDs[2] != withOPKs { + t.Fatalf("exact retry used %q, want reused operation ID %q", rpc.publishOperationIDs[2], withOPKs) + } + + // A replenished OPK sidecar changes the payload, so a new ID is required. + if _, _, err := manager.GenerateOneTimePrekey("opk-extra-001"); err != nil { + t.Fatalf("GenerateOneTimePrekey failed: %v", err) + } + if _, err := manager.PublishPrekeyBundle(bundle); err != nil { + t.Fatalf("PublishPrekeyBundle with replenished OPKs failed: %v", err) + } + if got := rpc.publishOperationIDs[3]; got == withOPKs { + t.Fatalf("replenished OPK sidecar reused operation ID %q, want a new one", got) + } +} + func TestSessionInitAndFollowUpRoundTrip(t *testing.T) { aliceDoc, bobDoc, aliceStatic, bobStatic, bobSPK, bundle := buildSessionFixtures(t) aliceDID := stringValue(aliceDoc["id"]) @@ -440,6 +510,7 @@ type fakeRPCClient struct { totalGetPrekeyCalls int failPublishWithOPK bool publishAttempts int + publishOperationIDs []string getPrekeyOperationIDs []string calls [][2]any } @@ -450,6 +521,7 @@ func (f *fakeRPCClient) Call(method string, params map[string]any) (map[string]a case "direct.e2ee.publish_prekey_bundle": f.publishAttempts++ meta := params["meta"].(map[string]any) + f.publishOperationIDs = append(f.publishOperationIDs, stringValue(meta["operation_id"])) target := meta["target"].(map[string]any) if target["kind"] != "service" || target["did"] != f.expectedServiceDID { return nil, invalidField("publish_prekey_bundle target") diff --git a/golang/direct_e2ee/prekey_manager.go b/golang/direct_e2ee/prekey_manager.go index 15591fe..6f517c3 100644 --- a/golang/direct_e2ee/prekey_manager.go +++ b/golang/direct_e2ee/prekey_manager.go @@ -2,8 +2,10 @@ package directe2ee import ( "crypto/ecdh" + "crypto/sha256" "fmt" "sort" + "strings" "time" anp "github.com/agent-network-protocol/anp/golang" @@ -84,6 +86,10 @@ func (m *PrekeyManager) BuildPrekeyBundle(signedPrekey SignedPrekey, bundleID st } // PublishPrekeyBundle publishes a prekey bundle over the RPC boundary. +// The publish operation ID identifies the complete publish payload: it is +// derived from the bundle ID plus a digest of the OPK sidecar key IDs, so an +// exact retry of the same payload reuses the ID while a replenished OPK +// sidecar produces a new one. func (m *PrekeyManager) PublishPrekeyBundle(bundle PrekeyBundle) (map[string]any, error) { if m.rpcClient == nil { return nil, &Error{Code: "rpc_unavailable", Message: "RPC client is not configured"} @@ -96,7 +102,22 @@ func (m *PrekeyManager) PublishPrekeyBundle(bundle PrekeyBundle) (map[string]any if len(oneTimePrekeys) > 0 { body["one_time_prekeys"] = oneTimePrekeysToAny(oneTimePrekeys) } - return m.rpcClient("direct.e2ee.publish_prekey_bundle", map[string]any{"meta": map[string]any{"anp_version": "1.0", "profile": "anp.direct.e2ee.v1", "security_profile": "transport-protected", "sender_did": m.localDID, "target": map[string]any{"kind": "service", "did": m.localServiceDID}, "operation_id": "op-publish-" + bundle.BundleID}, "body": body}) + return m.rpcClient("direct.e2ee.publish_prekey_bundle", map[string]any{"meta": map[string]any{"anp_version": "1.0", "profile": "anp.direct.e2ee.v1", "security_profile": "transport-protected", "sender_did": m.localDID, "target": map[string]any{"kind": "service", "did": m.localServiceDID}, "operation_id": publishOperationID(bundle.BundleID, oneTimePrekeys)}, "body": body}) +} + +// publishOperationID derives the publish operation ID from the complete +// publish payload rather than the stable bundle ID alone. oneTimePrekeys must +// be sorted by key ID (listOneTimePrekeys already guarantees this). +func publishOperationID(bundleID string, oneTimePrekeys []OneTimePrekey) string { + if len(oneTimePrekeys) == 0 { + return "op-publish-" + bundleID + } + keyIDs := make([]string, 0, len(oneTimePrekeys)) + for _, oneTimePrekey := range oneTimePrekeys { + keyIDs = append(keyIDs, oneTimePrekey.KeyID) + } + digest := sha256.Sum256([]byte(strings.Join(keyIDs, ","))) + return fmt.Sprintf("op-publish-%s-opks-%x", bundleID, digest[:4]) } // EnsureFreshPrekeyBundle returns the latest prekey bundle or creates one.