Update accumulator value in wallet on repair (openwallet-foundation#3299)

* Update wallet reg entry on repair

Signed-off-by: Jamie Hale <[email protected]>

* Refactoring

Signed-off-by: Jamie Hale <[email protected]>

* await prepare notification function

Signed-off-by: Jamie Hale <[email protected]>

* refactor / fix test

Signed-off-by: Jamie Hale <[email protected]>

* Add a couple unit tests

Signed-off-by: Jamie Hale <[email protected]>

---------

Signed-off-by: Jamie Hale <[email protected]>

Author: jamshale

acapy_agent/revocation/manager.py

@@ -2,7 +2,7 @@
 
 import json
 import logging
-from typing import Mapping, Optional, Sequence, Text, Tuple
+from typing import Mapping, NamedTuple, Optional, Sequence, Text, Tuple
 
 from ..connections.models.conn_record import ConnRecord
 from ..core.error import BaseError
@@ -26,6 +26,17 @@ class RevocationManagerError(BaseError):
     """Revocation manager error."""
 
 
+class RevocationNotificationInfo(NamedTuple):
+    """Revocation notification information."""
+
+    rev_reg_id: str
+    cred_rev_id: str
+    thread_id: Optional[str]
+    connection_id: Optional[str]
+    comment: Optional[str]
+    notify_version: Optional[str]
+
+
 class RevocationManager:
     """Class for managing revocation operations."""
 
@@ -107,6 +118,46 @@ async def revoke_credential_by_cred_ex_id(
             write_ledger=write_ledger,
         )
 
+    async def _prepare_revocation_notification(
+        self,
+        revoc_notif_info: RevocationNotificationInfo,
+    ):
+        """Saves the revocation notification record, and thread_id if not provided."""
+        thread_id = (
+            revoc_notif_info.thread_id
+            or f"indy::{revoc_notif_info.rev_reg_id}::{revoc_notif_info.cred_rev_id}"
+        )
+        rev_notify_rec = RevNotificationRecord(
+            rev_reg_id=revoc_notif_info.rev_reg_id,
+            cred_rev_id=revoc_notif_info.cred_rev_id,
+            thread_id=thread_id,
+            connection_id=revoc_notif_info.connection_id,
+            comment=revoc_notif_info.comment,
+            version=revoc_notif_info.notify_version,
+        )
+        async with self._profile.session() as session:
+            await rev_notify_rec.save(session, reason="New revocation notification")
+
+    async def _get_endorsement_txn_for_revocation(
+        self, endorser_conn_id: str, issuer_rr_upd: IssuerRevRegRecord
+    ):
+        async with self._profile.session() as session:
+            try:
+                connection_record = await ConnRecord.retrieve_by_id(
+                    session, endorser_conn_id
+                )
+            except StorageNotFoundError:
+                raise RevocationManagerError(
+                    "No endorser connection record found " f"for id: {endorser_conn_id}"
+                )
+            endorser_info = await connection_record.metadata_get(session, "endorser_info")
+        endorser_did = endorser_info["endorser_did"]
+        return await issuer_rr_upd.send_entry(
+            self._profile,
+            write_ledger=False,
+            endorser_did=endorser_did,
+        )
+
     async def revoke_credential(
         self,
         rev_reg_id: str,
@@ -150,81 +201,72 @@ async def revoke_credential(
                 write_ledger is True, otherwise None.
         """
         issuer = self._profile.inject(IndyIssuer)
-
         revoc = IndyRevocation(self._profile)
+
         issuer_rr_rec = await revoc.get_issuer_rev_reg_record(rev_reg_id)
         if not issuer_rr_rec:
             raise RevocationManagerError(
                 f"No revocation registry record found for id: {rev_reg_id}"
             )
 
         if notify:
-            thread_id = thread_id or f"indy::{rev_reg_id}::{cred_rev_id}"
-            rev_notify_rec = RevNotificationRecord(
-                rev_reg_id=rev_reg_id,
-                cred_rev_id=cred_rev_id,
-                thread_id=thread_id,
-                connection_id=connection_id,
-                comment=comment,
-                version=notify_version,
-            )
-            async with self._profile.session() as session:
-                await rev_notify_rec.save(session, reason="New revocation notification")
-
-        if publish:
-            rev_reg = await revoc.get_ledger_registry(rev_reg_id)
-            await rev_reg.get_or_fetch_local_tails_path()
-            # pick up pending revocations on input revocation registry
-            crids = (issuer_rr_rec.pending_pub or []) + [cred_rev_id]
-            (delta_json, _) = await issuer.revoke_credentials(
-                issuer_rr_rec.cred_def_id,
-                issuer_rr_rec.revoc_reg_id,
-                issuer_rr_rec.tails_local_path,
-                crids,
+            await self._prepare_revocation_notification(
+                RevocationNotificationInfo(
+                    rev_reg_id=rev_reg_id,
+                    cred_rev_id=cred_rev_id,
+                    thread_id=thread_id,
+                    connection_id=connection_id,
+                    comment=comment,
+                    notify_version=notify_version,
+                ),
             )
-            async with self._profile.transaction() as txn:
-                issuer_rr_upd = await IssuerRevRegRecord.retrieve_by_id(
-                    txn, issuer_rr_rec.record_id, for_update=True
-                )
-                if delta_json:
-                    issuer_rr_upd.revoc_reg_entry = json.loads(delta_json)
-                await issuer_rr_upd.clear_pending(txn, crids)
-                await txn.commit()
-            await self.set_cred_revoked_state(rev_reg_id, crids)
-            if delta_json:
-                if write_ledger:
-                    rev_entry_resp = await issuer_rr_upd.send_entry(self._profile)
-                    await notify_revocation_published_event(
-                        self._profile, rev_reg_id, [cred_rev_id]
-                    )
-                    return rev_entry_resp
-                else:
-                    async with self._profile.session() as session:
-                        try:
-                            connection_record = await ConnRecord.retrieve_by_id(
-                                session, endorser_conn_id
-                            )
-                        except StorageNotFoundError:
-                            raise RevocationManagerError(
-                                "No endorser connection record found "
-                                f"for id: {endorser_conn_id}"
-                            )
-                        endorser_info = await connection_record.metadata_get(
-                            session, "endorser_info"
-                        )
-                    endorser_did = endorser_info["endorser_did"]
-                    rev_entry_resp = await issuer_rr_upd.send_entry(
-                        self._profile,
-                        write_ledger=write_ledger,
-                        endorser_did=endorser_did,
-                    )
-                    return rev_entry_resp
-        else:
+
+        if not publish:
+            # If not publishing, just mark the revocation as pending.
             async with self._profile.transaction() as txn:
                 await issuer_rr_rec.mark_pending(txn, cred_rev_id)
                 await txn.commit()
             return None
 
+        rev_reg = await revoc.get_ledger_registry(rev_reg_id)
+        await rev_reg.get_or_fetch_local_tails_path()
+        # pick up pending revocations on input revocation registry
+        crids = (issuer_rr_rec.pending_pub or []) + [cred_rev_id]
+        (delta_json, _) = await issuer.revoke_credentials(
+            issuer_rr_rec.cred_def_id,
+            issuer_rr_rec.revoc_reg_id,
+            issuer_rr_rec.tails_local_path,
+            crids,
+        )
+
+        # Update the revocation registry record with the new delta
+        # and clear pending revocations
+        async with self._profile.transaction() as txn:
+            issuer_rr_upd = await IssuerRevRegRecord.retrieve_by_id(
+                txn, issuer_rr_rec.record_id, for_update=True
+            )
+            if delta_json:
+                issuer_rr_upd.revoc_reg_entry = json.loads(delta_json)
+            await issuer_rr_upd.clear_pending(txn, crids)
+            await txn.commit()
+
+        await self.set_cred_revoked_state(rev_reg_id, crids)
+
+        # Revocation list needs to be updated on ledger
+        if delta_json:
+            # Can write to ledger directly
+            if write_ledger:
+                rev_entry_resp = await issuer_rr_upd.send_entry(self._profile)
+                await notify_revocation_published_event(
+                    self._profile, rev_reg_id, [cred_rev_id]
+                )
+                return rev_entry_resp
+            # Author --> Need endorsed transaction for revocation
+            else:
+                return await self._get_endorsement_txn_for_revocation(
+                    endorser_conn_id, issuer_rr_upd
+                )
+
     async def update_rev_reg_revoked_state(
         self,
         apply_ledger_update: bool,

acapy_agent/revocation/models/issuer_rev_reg_record.py

@@ -14,7 +14,11 @@
 from uuid_utils import uuid4
 
 from ...core.profile import Profile, ProfileSession
-from ...indy.credx.issuer import CATEGORY_CRED_DEF, CATEGORY_REV_REG_DEF_PRIVATE
+from ...indy.credx.issuer import (
+    CATEGORY_CRED_DEF,
+    CATEGORY_REV_REG,
+    CATEGORY_REV_REG_DEF_PRIVATE,
+)
 from ...indy.issuer import IndyIssuer, IndyIssuerError
 from ...indy.models.revocation import (
     IndyRevRegDef,
@@ -358,91 +362,108 @@ async def send_entry(
 
         return rev_entry_res
 
+    def _get_revoked_discrepancies(
+        self, recs: Sequence[IssuerCredRevRecord], rev_reg_delta: dict
+    ) -> Tuple[list, int]:
+        revoked_ids = []
+        rec_count = 0
+        for rec in recs:
+            if rec.state == IssuerCredRevRecord.STATE_REVOKED:
+                revoked_ids.append(int(rec.cred_rev_id))
+                if int(rec.cred_rev_id) not in rev_reg_delta["value"]["revoked"]:
+                    rec_count += 1
+
+        return revoked_ids, rec_count
+
     async def fix_ledger_entry(
         self,
         profile: Profile,
         apply_ledger_update: bool,
         genesis_transactions: str,
     ) -> Tuple[dict, dict, dict]:
         """Fix the ledger entry to match wallet-recorded credentials."""
+        recovery_txn = {}
+        applied_txn = {}
+
         # get rev reg delta (revocations published to ledger)
         ledger = profile.inject(BaseLedger)
         async with ledger:
             (rev_reg_delta, _) = await ledger.get_revoc_reg_delta(self.revoc_reg_id)
 
         # get rev reg records from wallet (revocations and status)
-        recs = []
-        rec_count = 0
-        accum_count = 0
-        recovery_txn = {}
-        applied_txn = {}
         async with profile.session() as session:
             recs = await IssuerCredRevRecord.query_by_ids(
                 session, rev_reg_id=self.revoc_reg_id
             )
 
-        revoked_ids = []
-        for rec in recs:
-            if rec.state == IssuerCredRevRecord.STATE_REVOKED:
-                revoked_ids.append(int(rec.cred_rev_id))
-                if int(rec.cred_rev_id) not in rev_reg_delta["value"]["revoked"]:
-                    # await rec.set_state(session, IssuerCredRevRecord.STATE_ISSUED)
-                    rec_count += 1
+        revoked_ids, rec_count = self._get_revoked_discrepancies(recs, rev_reg_delta)
+
+        LOGGER.debug(f"Fixed entry recs count = {rec_count}")
+        LOGGER.debug(f"Rev reg entry value: {self.revoc_reg_entry.value}")
+        LOGGER.debug(f'Rev reg delta: {rev_reg_delta.get("value")}')
+
+        # No update required if no discrepancies
+        if rec_count == 0:
+            return (rev_reg_delta, {}, {})
+
+        # We have revocation discrepancies, generate the recovery txn
+        async with profile.session() as session:
+            # We need the cred_def and rev_reg_def_private to generate the recovery txn
+            issuer_rev_reg_record = await IssuerRevRegRecord.retrieve_by_revoc_reg_id(
+                session, self.revoc_reg_id
+            )
+            cred_def_id = issuer_rev_reg_record.cred_def_id
+            cred_def = await session.handle.fetch(CATEGORY_CRED_DEF, cred_def_id)
+            rev_reg_def_private = await session.handle.fetch(
+                CATEGORY_REV_REG_DEF_PRIVATE, self.revoc_reg_id
+            )
 
-        LOGGER.debug(">>> fixed entry recs count = %s", rec_count)
-        LOGGER.debug(
-            ">>> rev_reg_record.revoc_reg_entry.value: %s",
-            self.revoc_reg_entry.value,
+        credx_module = importlib.import_module("indy_credx")
+        cred_defn = credx_module.CredentialDefinition.load(cred_def.value_json)
+        rev_reg_defn_private = credx_module.RevocationRegistryDefinitionPrivate.load(
+            rev_reg_def_private.value_json
         )
-        LOGGER.debug('>>> rev_reg_delta.get("value"): %s', rev_reg_delta.get("value"))
-
-        # if we had any revocation discrepancies, check the accumulator value
-        if rec_count > 0:
-            if (self.revoc_reg_entry.value and rev_reg_delta.get("value")) and not (
-                self.revoc_reg_entry.value.accum == rev_reg_delta["value"]["accum"]
-            ):
-                # self.revoc_reg_entry = rev_reg_delta["value"]
-                # await self.save(session)
-                accum_count += 1
+        calculated_txn = await generate_ledger_rrrecovery_txn(
+            genesis_transactions,
+            self.revoc_reg_id,
+            revoked_ids,
+            cred_defn,
+            rev_reg_defn_private,
+        )
+        recovery_txn = json.loads(calculated_txn.to_json())
+
+        LOGGER.debug(f"Applying ledger update: {apply_ledger_update}")
+        if apply_ledger_update:
             async with profile.session() as session:
-                issuer_rev_reg_record = await IssuerRevRegRecord.retrieve_by_revoc_reg_id(
-                    session, self.revoc_reg_id
+                ledger = session.inject_or(BaseLedger)
+                if not ledger:
+                    reason = "No ledger available"
+                    if not session.context.settings.get_value("wallet.type"):
+                        reason += ": missing wallet-type?"
+                    raise LedgerError(reason=reason)
+
+                async with ledger:
+                    ledger_response = await ledger.send_revoc_reg_entry(
+                        self.revoc_reg_id, "CL_ACCUM", recovery_txn
+                    )
+
+            applied_txn = ledger_response["result"]
+
+            # Update the local wallets rev reg entry with the new accumulator value
+            async with profile.session() as session:
+                rev_reg = await session.handle.fetch(
+                    CATEGORY_REV_REG, self.revoc_reg_id, for_update=True
                 )
-                cred_def_id = issuer_rev_reg_record.cred_def_id
-                _cred_def = await session.handle.fetch(CATEGORY_CRED_DEF, cred_def_id)
-                _rev_reg_def_private = await session.handle.fetch(
-                    CATEGORY_REV_REG_DEF_PRIVATE, self.revoc_reg_id
+                new_value_json = rev_reg.value_json
+                new_value_json["value"]["accum"] = applied_txn["txn"]["data"]["value"][
+                    "accum"
+                ]
+                await session.handle.replace(
+                    CATEGORY_REV_REG,
+                    rev_reg.name,
+                    json.dumps(new_value_json),
+                    rev_reg.tags,
                 )
-            credx_module = importlib.import_module("indy_credx")
-            cred_defn = credx_module.CredentialDefinition.load(_cred_def.value_json)
-            rev_reg_defn_private = credx_module.RevocationRegistryDefinitionPrivate.load(
-                _rev_reg_def_private.value_json
-            )
-            calculated_txn = await generate_ledger_rrrecovery_txn(
-                genesis_transactions,
-                self.revoc_reg_id,
-                revoked_ids,
-                cred_defn,
-                rev_reg_defn_private,
-            )
-            recovery_txn = json.loads(calculated_txn.to_json())
-
-            LOGGER.debug(">>> apply_ledger_update = %s", apply_ledger_update)
-            if apply_ledger_update:
-                async with profile.session() as session:
-                    ledger = session.inject_or(BaseLedger)
-                    if not ledger:
-                        reason = "No ledger available"
-                        if not session.context.settings.get_value("wallet.type"):
-                            reason += ": missing wallet-type?"
-                        raise LedgerError(reason=reason)
-
-                    async with ledger:
-                        ledger_response = await ledger.send_revoc_reg_entry(
-                            self.revoc_reg_id, "CL_ACCUM", recovery_txn
-                        )
-
-                applied_txn = ledger_response["result"]
 
         return (rev_reg_delta, recovery_txn, applied_txn)