Rehydrate escrowed signatures into kwa after threshold is satisfied

Author: ryan-hansen

src/keri/core/kraming.py

@@ -806,6 +806,114 @@ def _remNonAuthAttachments(self, key):
         self.db.kramBSSS.rem(key)
         self.db.kramTMQS.rem(key)
 
+    @staticmethod
+    def _dedupeAttachmentItems(items):
+        """Return items in first-seen order with duplicates removed.
+
+        Items are CESR tuples or bytes; identity uses a nested qb64/raw key.
+        """
+        seen = set()
+        out = []
+        for item in items:
+            key = Kramer._attachmentItemKey(item)
+            if key in seen:
+                continue
+            seen.add(key)
+            out.append(item)
+        return out
+
+    @staticmethod
+    def _attachmentItemKey(item):
+        if isinstance(item, (bytes, memoryview)):
+            return ('b', bytes(item))
+        if isinstance(item, (list, tuple)):
+            return tuple(Kramer._attachmentItemKey(x) for x in item)
+        if hasattr(item, 'qb64'):
+            return item.qb64
+        return repr(item)
+
+    def _rehydrateKwaFromEscrow(self, partialKey, _senderId, kwa):
+        """Merge escrowed partial multisig state into ``kwa`` for downstream dispatch.
+
+        After threshold is satisfied on a later delivery, ``kwa`` reflects only
+        that parse; ``kramPMKS`` and non-auth attachment DBs hold the union of
+        state accumulated across deliveries.
+
+        Parameters:
+            partialKey (tuple): ``(AID, MID)`` escrow key
+            _senderId (str): message sender AID (qb64); reserved for filtering
+            kwa (dict): parser attachment dict; mutated in place
+        """
+        escrow_sigs = self.db.kramPMKS.get(partialKey)
+        if escrow_sigs:
+            kwa['sigers'] = sorted(escrow_sigs, key=lambda s: s.index)
+
+        trqs_esc = list(self.db.kramTRQS.get(partialKey) or [])
+        kwa['trqs'] = self._dedupeAttachmentItems(
+            trqs_esc + kwa.get('trqs', []))
+
+        flat_tsgs = list(self.db.kramTSGS.get(partialKey) or [])
+        groups = {}
+        for prefixer, number, diger, siger in flat_tsgs:
+            gk = (prefixer.qb64, number.sn, diger.qb64)
+            if gk not in groups:
+                groups[gk] = [prefixer, number, diger, []]
+            sigers = groups[gk][3]
+            if not any(s.qb64 == siger.qb64 for s in sigers):
+                sigers.append(siger)
+
+        merged_tsgs = {}
+        for quad in groups.values():
+            prefixer, number, diger, sigers = quad
+            gk = (prefixer.qb64, number.sn, diger.qb64)
+            merged_tsgs[gk] = (prefixer, number, diger, list(sigers))
+
+        for prefixer, number, diger, sigers in kwa.get('tsgs', []):
+            gk = (prefixer.qb64, number.sn, diger.qb64)
+            if gk not in merged_tsgs:
+                merged_tsgs[gk] = (prefixer, number, diger, [])
+            bucket = merged_tsgs[gk][3]
+            for s in sigers:
+                if not any(x.qb64 == s.qb64 for x in bucket):
+                    bucket.append(s)
+
+        if merged_tsgs:
+            kwa['tsgs'] = list(merged_tsgs.values())
+        else:
+            kwa.pop('tsgs', None)
+
+        ssts_esc = list(self.db.kramSSTS.get(partialKey) or [])
+        kwa['ssts'] = self._dedupeAttachmentItems(
+            ssts_esc + kwa.get('ssts', []))
+
+        frcs_esc = list(self.db.kramFRCS.get(partialKey) or [])
+        kwa['frcs'] = self._dedupeAttachmentItems(
+            frcs_esc + kwa.get('frcs', []))
+
+        tdcs_esc = list(self.db.kramTDCS.get(partialKey) or [])
+        kwa['tdcs'] = self._dedupeAttachmentItems(
+            tdcs_esc + kwa.get('tdcs', []))
+
+        ptds_esc = list(self.db.kramPTDS.get(partialKey) or [])
+        kwa['ptds'] = self._dedupeAttachmentItems(
+            ptds_esc + kwa.get('ptds', []))
+
+        bsqs_esc = list(self.db.kramBSQS.get(partialKey) or [])
+        kwa['bsqs'] = self._dedupeAttachmentItems(
+            bsqs_esc + kwa.get('bsqs', []))
+
+        bsss_esc = list(self.db.kramBSSS.get(partialKey) or [])
+        kwa['bsss'] = self._dedupeAttachmentItems(
+            bsss_esc + kwa.get('bsss', []))
+
+        tmqs_esc = list(self.db.kramTMQS.get(partialKey) or [])
+        kwa['tmqs'] = self._dedupeAttachmentItems(
+            tmqs_esc + kwa.get('tmqs', []))
+
+        for name in ('trqs', 'ssts', 'frcs', 'tdcs', 'ptds', 'bsqs', 'bsss',
+                     'tmqs'):
+            if not kwa.get(name):
+                kwa.pop(name, None)
 
     def intake(self, serder, kwa=None):
         """Process message through KRAM denial and cache logic.
@@ -981,6 +1089,7 @@ def kramit(self, msg, kwa=None):
                     sigIndices = [sig.index for sig in allSigs]
 
                     if kever.tholder.satisfy(indices=sigIndices):
+                        self._rehydrateKwaFromEscrow(key, senderId, kwa)
                         return msg
 
                 # Threshold not satisfied, message remains pending
@@ -1206,6 +1315,7 @@ def kramit(self, msg, kwa=None):
                     sigIndices = [sig.index for sig in allSigs]
 
                     if kever.tholder.satisfy(indices=sigIndices):
+                        self._rehydrateKwaFromEscrow(partialKey, senderId, kwa)
                         return msg
 
                 # Threshold not satisfied, message remains pending

tests/core/test_kraming.py

@@ -1916,6 +1916,102 @@ def test_non_auth_attachments_stored(mockHelpingNowUTC):
     """Done Test"""
 
 
+def test_multisig_kwa_rehydration_after_threshold(mockHelpingNowUTC):
+    """Threshold on a later delivery merges escrow PMKS + non-auth into kwa.
+
+    The second parse may omit non-auth attachments that were stored on the
+    first partial; :meth:`Kramer._rehydrateKwaFromEscrow` restores them.
+    """
+
+    salt1 = Salter(raw=b'0123456789abcdef').qb64
+    salt2 = Salter(raw=b'0123456789abcdeg').qb64
+
+    with (openHby(name="rehSender", base="test", salt=salt1) as senderHby,
+          openHby(name="rehReceiver", base="test", salt=salt2) as receiverHby):
+
+        senderHab = senderHby.makeHab(name="rehSender", isith='2', icount=3,
+                                      transferable=True)
+        receiverHab = receiverHby.makeHab(name="rehReceiver", isith='1', icount=1,
+                                          transferable=True)
+
+        crossKvy = Kevery(db=receiverHby.db, lax=False, local=False)
+        senderIcp = senderHab.makeOwnEvent(sn=0)
+        Parser(version=Vrsn_1_0).parse(ims=bytearray(senderIcp), kvy=crossKvy)
+
+        with openCF(name="rehKram", base="test") as cf:
+            cf.put(KRAM_INTEGRATION_CONFIG)
+            kramer = Kramer(db=receiverHby.db, cf=cf)
+
+            stamp = helping.nowIso8601()
+            prefixer = Prefixer(qb64=senderHab.pre)
+
+            msg = query(pre=senderHab.pre,
+                        route="ksn",
+                        query=dict(i=senderHab.pre, src=senderHab.pre),
+                        stamp=stamp,
+                        pvrsn=Vrsn_2_0)
+
+            allSigers = senderHab.mgr.sign(ser=msg.raw,
+                                           verfers=senderHab.kever.verfers,
+                                           indexed=True)
+
+            partialKey = (senderHab.pre, msg.said)
+
+            senderKever = receiverHby.db.kevers[senderHab.pre]
+            seqner = Seqner(sn=senderKever.sner.num)
+            saider = Saider(qb64=senderKever.serder.said)
+            diger = Diger(ser=msg.raw)
+            otherPrefixer = Prefixer(qb64=receiverHab.pre)
+
+            trqs = [(prefixer, seqner, saider, allSigers[0])]
+            tsgs = [(otherPrefixer, seqner, saider, [allSigers[0]])]
+            ssts = [(otherPrefixer, seqner, saider)]
+            firner = Seqner(sn=0)
+            dater = Dater(dts=stamp)
+            frcs = [(firner, dater)]
+            verser = Verser(pvrsn=Vrsn_2_0)
+            tdcs = [(verser, diger)]
+            ptds = [b'\x00\x01\x02\x03']
+            noncer0 = Noncer()
+            noncer1 = Noncer()
+            labeler = Labeler(label='test')
+            bsqs = [(diger, noncer0, noncer1, labeler)]
+            number = Number(num=1)
+            noncer2 = Noncer()
+            bsss = [(diger, noncer0, noncer1, labeler, number, noncer2)]
+            texter = Texter(text='application/json')
+            tmqs = [(diger, noncer0, labeler, texter)]
+
+            kwa = dict(ssgs=[(prefixer, [allSigers[0]])],
+                       trqs=trqs, tsgs=tsgs, ssts=ssts,
+                       frcs=frcs, tdcs=tdcs, ptds=ptds,
+                       bsqs=bsqs, bsss=bsss, tmqs=tmqs)
+
+            r1 = kramer.kramit(msg, kwa)
+            assert r1 is None
+            assert len(kramer.db.kramPMKS.get(keys=partialKey)) == 1
+
+            kwa2 = dict(ssgs=[(prefixer, [allSigers[2]])])
+            r2 = kramer.kramit(msg, kwa2)
+            assert r2 is not None
+
+            assert {s.index for s in kwa2['sigers']} == {0, 2}
+            assert len(kwa2['trqs']) == 1
+            assert len(kwa2['tsgs']) == 1
+            assert len(kwa2['ssts']) == 1
+            assert len(kwa2['frcs']) == 1
+            assert len(kwa2['tdcs']) == 1
+            assert len(kwa2['ptds']) == 1
+            assert len(kwa2['bsqs']) == 1
+            assert len(kwa2['bsss']) == 1
+            assert len(kwa2['tmqs']) == 1
+
+            escrow = kramer.db.kramPMKS.get(keys=partialKey)
+            assert {s.index for s in escrow} == {0, 2}
+
+    """Done Test"""
+
+
 def test_non_auth_attachments_empty_kwa(mockHelpingNowUTC):
     """Test that _storeNonAuthAttachments is a no-op when kwa contains no
     non-auth attachment keys. Partial dbs remain empty.