diff --git a/acapy_agent/anoncreds/default/legacy_indy/recover.py b/acapy_agent/anoncreds/default/legacy_indy/recover.py index e5a00d4898..5ae46a743c 100644 --- a/acapy_agent/anoncreds/default/legacy_indy/recover.py +++ b/acapy_agent/anoncreds/default/legacy_indy/recover.py @@ -1,15 +1,48 @@ """Recover a revocation registry.""" +import asyncio import hashlib import logging import time +from typing import Optional, Sequence, Tuple import aiohttp import base58 import indy_vdr -from anoncreds import RevocationRegistry, RevocationRegistryDefinition +from anoncreds import ( + RevocationRegistry, + RevocationRegistryDefinition, +) +from uuid_utils import uuid4 -from ...models.revocation import RevList +from ....cache.base import BaseCache +from ....connections.models.conn_record import ConnRecord +from ....core.profile import Profile +from ....ledger.base import BaseLedger +from ....ledger.error import ( + LedgerError, +) +from ....ledger.multiple_ledger.ledger_requests_executor import ( + GET_REVOC_REG_DELTA, + IndyLedgerRequestsExecutor, +) +from ....messaging.responder import BaseResponder +from ....multitenant.base import BaseMultitenantManager +from ....protocols.endorse_transaction.v1_0.manager import ( + TransactionManager, + TransactionManagerError, +) +from ....protocols.endorse_transaction.v1_0.util import get_endorser_connection_id +from ....storage.error import StorageError +from ...constants import ( + CATEGORY_REV_LIST, + CATEGORY_REV_REG_DEF, +) +from ...models.issuer_cred_rev_record import IssuerCredRevRecord +from ...models.revocation import ( + RevList, +) +from ...revocation.manager import RevocationManagerError LOGGER = logging.getLogger(__name__) @@ -89,14 +122,19 @@ async def fetch_txns( return registry_from_ledger, revoked -async def generate_ledger_rrrecovery_txn(genesis_txns: str, rev_list: RevList) -> dict: - """Generate a new ledger accum entry, using the wallet value if revocations ahead of ledger.""" # noqa: E501 - registry_from_ledger, prev_revoked = await fetch_txns( +async def generate_ledger_rrrecovery_txn(genesis_txns: str, rev_list: RevList): + """Generate a new ledger accum entry, based on wallet vs ledger revocation state.""" + new_delta = None + + ledger_data = await fetch_txns( genesis_txns, rev_list.rev_reg_def_id, rev_list.issuer_id ) + if not ledger_data: + return new_delta + registry_from_ledger, prev_revoked = ledger_data set_revoked = { - index for index, value in enumerate(rev_list.revocation_list) if value == 1 + i for i, revoked in enumerate(rev_list.revocation_list) if revoked == 1 } mismatch = prev_revoked - set_revoked if mismatch: @@ -120,4 +158,254 @@ async def generate_ledger_rrrecovery_txn(genesis_txns: str, rev_list: RevList) - registry["value"]["accum"] = rev_list.current_accumulator registry["value"]["issued"] = [] registry["value"]["revoked"] = list(updates) + return registry + + +async def _get_endorser_info( + profile: Profile, +) -> Tuple[Optional[str], Optional[ConnRecord]]: + connection_id = await get_endorser_connection_id(profile) + + endorser_did = None + async with profile.session() as session: + connection_record = await ConnRecord.retrieve_by_id(session, connection_id) + endorser_info = await connection_record.metadata_get(session, "endorser_info") + endorser_did = endorser_info.get("endorser_did") + + return endorser_did, connection_record + + +async def fix_and_publish_from_invalid_accum_err(profile: Profile, err_msg: str): + """Fix and publish revocation registry entries from invalid accumulator error.""" + cache = profile.inject_or(BaseCache) + + async def check_retry(accum): + """Used to manage retries for fixing revocation registry entries.""" + if cache is None: + LOGGER.warning( + "No cache backend configured; skipping retry tracking for %s", + accum, + ) + return + + retry_value = await cache.get(accum) + if not retry_value: + await cache.set(accum, 5) + else: + if retry_value > 0: + await cache.set(accum, retry_value - 1) + else: + LOGGER.error( + "Revocation registry entry transaction failed for %s", + accum, + ) + + def get_genesis_transactions(): + """Get the genesis transactions needed for fixing broken accum.""" + genesis_transactions = profile.context.settings.get("ledger.genesis_transactions") + if not genesis_transactions: + write_ledger = profile.context.injector.inject(BaseLedger) + pool = write_ledger.pool + genesis_transactions = pool.genesis_txns + return genesis_transactions + + async def create_and_send_endorser_txn(): + """Create and send the endorser transaction again.""" + async with ledger: + # Create the revocation registry entry + (rev_reg_def_id, requested_txn) = await ledger.send_revoc_reg_entry( + rev_list.rev_reg_def_id, + "CL_ACCUM", + recovery_txn, + rev_list.issuer_id, + write_ledger=False, + endorser_did=endorser_did, + ) + + job_id = uuid4().hex + meta_data = { + "context": { + "job_id": job_id, + "rev_reg_def_id": rev_reg_def_id, + "rev_list": rev_list.serialize(), + "options": { + "endorser_connection_id": connection.connection_id, + "create_transaction_for_endorser": True, + }, + } + } + + # Send the transaction to the endorser again with recovery txn + transaction_manager = TransactionManager(profile) + try: + revo_transaction = await transaction_manager.create_record( + messages_attach=requested_txn["signed_txn"], + connection_id=connection.connection_id, + meta_data=meta_data, + ) + ( + revo_transaction, + revo_transaction_request, + ) = await transaction_manager.create_request(transaction=revo_transaction) + except (StorageError, TransactionManagerError) as err: + raise RevocationManagerError(err.roll_up) from err + + responder = profile.inject_or(BaseResponder) + if not responder: + raise RevocationManagerError( + "No responder found. Unable to send transaction request" + ) + await responder.send( + revo_transaction_request, + connection_id=connection.connection_id, + ) + + async with profile.session() as session: + rev_reg_records = await session.handle.fetch_all( + CATEGORY_REV_REG_DEF, + {}, + ) + # Cycle through all rev_rev_def records to find the offending accumulator + for rev_reg_entry in rev_reg_records: + ledger = session.inject_or(BaseLedger) + async with ledger: + # Get the value from the ledger + (accum_response, _) = await ledger.get_revoc_reg_delta(rev_reg_entry.name) + accum = accum_response.get("value", {}).get("accum") + + # If the accum from the ledger matches the error message, fix it + if accum and accum in err_msg: + # if accum and accum in err_msg: + await check_retry(accum) + + # Get the genesis transactions needed for fix + genesis_transactions = get_genesis_transactions() + + # We know this needs endorsement + endorser_did, connection = await _get_endorser_info(profile) + rev_list_entry = await session.handle.fetch( + CATEGORY_REV_LIST, rev_reg_entry.name + ) + + rev_list = RevList.deserialize(rev_list_entry.value_json["rev_list"]) + + ( + rev_reg_delta, + recovery_txn, + applied_txn, + ) = await fix_ledger_entry( + profile, rev_list, False, genesis_transactions, False, endorser_did + ) + if recovery_txn.get("value"): + await create_and_send_endorser_txn() + + # Some time in between re-tries + await asyncio.sleep(1) + + +def _get_revoked_discrepancies( + 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( + profile: Profile, + rev_list: RevList, + apply_ledger_update: bool, + genesis_transactions: str, + write_ledger: bool = True, + endorser_did: Optional[str] = None, +) -> Tuple[dict, dict, dict]: + """Fix the ledger entry to match wallet-recorded credentials.""" + applied_txn = {} + recovery_txn = {} + + LOGGER.debug("Fixing ledger entry for revocation list...") + + multitenant_mgr = profile.inject_or(BaseMultitenantManager) + if multitenant_mgr: + ledger_exec_inst = IndyLedgerRequestsExecutor(profile) + else: + ledger_exec_inst = profile.inject(IndyLedgerRequestsExecutor) + _, ledger = await ledger_exec_inst.get_ledger_for_identifier( + rev_list.rev_reg_def_id, + txn_record_type=GET_REVOC_REG_DELTA, + ) + + if not ledger: + reason = "No ledger available for revocation registry entry fix" + if not profile.context.settings.get_value("wallet.type"): + reason += ": missing wallet-type?" + raise LedgerError(reason=reason) + + async with ledger: + (rev_reg_delta, _) = await ledger.get_revoc_reg_delta(rev_list.rev_reg_def_id) + + async with profile.session() as session: + # get rev reg records from wallet (revocations and status) + recs = await IssuerCredRevRecord.query_by_ids( + session, rev_reg_id=rev_list.rev_reg_def_id + ) + + revoked_ids, rec_count = _get_revoked_discrepancies(recs, rev_reg_delta) + + LOGGER.debug(f"Fixed entry recs count = {rec_count}") + LOGGER.debug(f"Fixed entry recs revoked ids = {revoked_ids}") + + # No update required if no discrepancies + if rec_count == 0: + return (rev_reg_delta, {}, {}) + + # We have revocation discrepancies, generate the recovery txn + recovery_txn = await generate_ledger_rrrecovery_txn( + genesis_transactions, rev_list + ) + + # If no recovery transaction was generated, skip ledger update + if not recovery_txn: + LOGGER.debug( + "No recovery transaction generated for revocation list %s; " + "skipping ledger update", + rev_list.rev_reg_def_id, + ) + return (rev_reg_delta, recovery_txn, applied_txn) + + if apply_ledger_update: + ledger_response = await ledger.send_revoc_reg_entry( + rev_list.rev_reg_def_id, + "CL_ACCUM", + recovery_txn, + rev_list.issuer_id, + write_ledger=write_ledger, + endorser_did=endorser_did, + ) + + if isinstance(ledger_response, dict) and "result" in ledger_response: + applied_txn = ledger_response["result"] + rev_list_value_json = rev_list.value_json + rev_list_value_json["rev_list"]["currentAccumulator"] = applied_txn[ + "txn" + ]["data"]["value"]["accum"] + rev_list.current_accumulator = applied_txn["txn"]["data"]["value"][ + "accum" + ] + await session.handle.replace( + CATEGORY_REV_LIST, + rev_list.rev_reg_def_id, + rev_list_value_json, + rev_list.tags, + ) + return (rev_reg_delta, recovery_txn, applied_txn) + + # Ledger update not applied, return without applied_txn + return (rev_reg_delta, recovery_txn, {}) diff --git a/acapy_agent/anoncreds/default/legacy_indy/registry.py b/acapy_agent/anoncreds/default/legacy_indy/registry.py index 734841547c..625e9d6946 100644 --- a/acapy_agent/anoncreds/default/legacy_indy/registry.py +++ b/acapy_agent/anoncreds/default/legacy_indy/registry.py @@ -1,16 +1,10 @@ """Legacy Indy Registry.""" -import asyncio import logging import re from asyncio import shield from typing import List, Optional, Pattern, Sequence, Tuple -from anoncreds import ( - CredentialDefinition, - RevocationRegistryDefinition, - RevocationRegistryDefinitionPrivate, -) from base58 import alphabet from uuid_utils import uuid4 @@ -18,7 +12,7 @@ from ....cache.base import BaseCache from ....config.injection_context import InjectionContext from ....core.event_bus import EventBus -from ....core.profile import Profile, ProfileSession +from ....core.profile import Profile from ....ledger.base import BaseLedger from ....ledger.error import ( LedgerError, @@ -26,7 +20,6 @@ LedgerTransactionError, ) from ....ledger.merkel_validation.constants import ( - GET_REVOC_REG_DELTA, GET_REVOC_REG_ENTRY, GET_SCHEMA, ) @@ -53,13 +46,8 @@ BaseAnonCredsRegistrar, BaseAnonCredsResolver, ) -from ...constants import ( - CATEGORY_REV_LIST, - CATEGORY_REV_REG_DEF, - CATEGORY_REV_REG_DEF_PRIVATE, -) from ...events import RevListFinishedEvent -from ...issuer import CATEGORY_CRED_DEF, AnonCredsIssuer, AnonCredsIssuerError +from ...issuer import AnonCredsIssuer, AnonCredsIssuerError from ...models.credential_definition import ( CredDef, CredDefResult, @@ -67,7 +55,6 @@ CredDefValue, GetCredDefResult, ) -from ...models.issuer_cred_rev_record import IssuerCredRevRecord from ...models.revocation import ( GetRevListResult, GetRevRegDefResult, @@ -81,7 +68,7 @@ ) from ...models.schema import AnonCredsSchema, GetSchemaResult, SchemaResult, SchemaState from ...models.schema_info import AnonCredsSchemaInfo -from .recover import generate_ledger_rrrecovery_txn +from .recover import fix_ledger_entry LOGGER = logging.getLogger(__name__) @@ -846,13 +833,13 @@ async def _revoc_reg_entry_with_fix( # In this scenario we try to post a correction LOGGER.warning("Retry ledger update/fix due to error") LOGGER.warning(err) - (_, _, rev_entry_res) = await self.fix_ledger_entry( + (_, _, rev_entry_res) = await fix_ledger_entry( profile, rev_list, True, ledger.pool.genesis_txns, - write_ledger, - endorser_did, + write_ledger=write_ledger, + endorser_did=endorser_did, ) LOGGER.warning("Ledger update/fix applied") elif "InvalidClientTaaAcceptanceError" in err.roll_up: @@ -1080,140 +1067,6 @@ async def update_revocation_list( revocation_list_metadata={}, ) - async def fix_ledger_entry( - self, - profile: Profile, - rev_list: RevList, - apply_ledger_update: bool, - genesis_transactions: str, - write_ledger: bool = True, - endorser_did: Optional[str] = None, - ) -> Tuple[dict, dict, dict]: - """Fix the ledger entry to match wallet-recorded credentials.""" - - def _wallet_accumalator_matches_ledger_list( - rev_list: RevList, rev_reg_delta: dict - ) -> bool: - return ( - rev_reg_delta.get("value") - and rev_list.current_accumulator == rev_reg_delta["value"]["accum"] - ) - - applied_txn = {} - recovery_txn = {} - - LOGGER.debug("Fixing ledger entry for revocation list...") - - multitenant_mgr = profile.inject_or(BaseMultitenantManager) - if multitenant_mgr: - ledger_exec_inst = IndyLedgerRequestsExecutor(profile) - else: - ledger_exec_inst = profile.inject(IndyLedgerRequestsExecutor) - _, ledger = await ledger_exec_inst.get_ledger_for_identifier( - rev_list.rev_reg_def_id, - txn_record_type=GET_REVOC_REG_DELTA, - ) - - async with ledger: - (rev_reg_delta, _) = await ledger.get_revoc_reg_delta(rev_list.rev_reg_def_id) - - async with profile.session() as session: - LOGGER.debug(f"revocation_list = {rev_list.revocation_list}") - LOGGER.debug(f"rev_reg_delta = {rev_reg_delta.get('value')}") - - rev_list = await self._sync_wallet_rev_list_with_issuer_cred_rev_records( - session, rev_list - ) - - if not _wallet_accumalator_matches_ledger_list(rev_list, rev_reg_delta): - recovery_txn = await generate_ledger_rrrecovery_txn( - genesis_transactions, rev_list - ) - - if apply_ledger_update and recovery_txn: - ledger = session.inject_or(BaseLedger) - if not ledger: - reason = NO_LEDGER_AVAILABLE_MSG - if not session.context.settings.get_value(WALLET_TYPE): - reason += MISSING_WALLET_TYPE_MSG - raise LedgerError(reason=reason) - - async with ledger: - applied_txn = await ledger.send_revoc_reg_entry( - rev_list.rev_reg_def_id, - "CL_ACCUM", - recovery_txn, - rev_list.issuer_id, - write_ledger, - endorser_did, - ) - - return (rev_reg_delta, recovery_txn, applied_txn) - - async def _sync_wallet_rev_list_with_issuer_cred_rev_records( - self, session: ProfileSession, rev_list: RevList - ) -> RevList: - """Sync the wallet revocation list with the issuer cred rev records.""" - - async def _revoked_issuer_cred_rev_record_ids() -> List[int]: - cred_rev_records = await IssuerCredRevRecord.query_by_ids( - session, rev_reg_id=rev_list.rev_reg_def_id - ) - return [ - int(rec.cred_rev_id) for rec in cred_rev_records if rec.state == "revoked" - ] - - def _revocation_list_to_array_of_indexes( - revocation_list: List[int], - ) -> List[int]: - return [index for index, value in enumerate(revocation_list) if value == 1] - - revoked = await _revoked_issuer_cred_rev_record_ids() - if revoked == _revocation_list_to_array_of_indexes(rev_list.revocation_list): - return rev_list - - # The revocation list is out of sync with the issuer cred rev records - # Recreate the revocation list with the issuer cred rev records - revoc_reg_def_entry = await session.handle.fetch( - CATEGORY_REV_REG_DEF, rev_list.rev_reg_def_id - ) - cred_def_entry = await session.handle.fetch( - CATEGORY_CRED_DEF, - RevRegDef.deserialize(revoc_reg_def_entry.value_json).cred_def_id, - ) - revoc_reg_def_private_entry = await session.handle.fetch( - CATEGORY_REV_REG_DEF_PRIVATE, rev_list.rev_reg_def_id - ) - updated_list = await asyncio.get_event_loop().run_in_executor( - None, - lambda: rev_list.to_native().update( - cred_def=CredentialDefinition.load(cred_def_entry.value_json), - rev_reg_def=RevocationRegistryDefinition.load( - revoc_reg_def_entry.value_json - ), - rev_reg_def_private=RevocationRegistryDefinitionPrivate.load( - revoc_reg_def_private_entry.raw_value - ), - issued=None, - revoked=revoked, - timestamp=None, - ), - ) - rev_list_entry_update = await session.handle.fetch( - CATEGORY_REV_LIST, rev_list.rev_reg_def_id, for_update=True - ) - tags = rev_list_entry_update.tags - rev_list_entry_update = rev_list_entry_update.value_json - rev_list_entry_update["rev_list"] = updated_list.to_dict() - - await session.handle.replace( - CATEGORY_REV_LIST, - rev_list.rev_reg_def_id, - value_json=rev_list_entry_update, - tags=tags, - ) - return RevList.deserialize(updated_list.to_json()) - async def txn_submit( self, ledger: BaseLedger, diff --git a/acapy_agent/anoncreds/default/legacy_indy/routes.py b/acapy_agent/anoncreds/default/legacy_indy/routes.py index 2a77ed48c0..f9b7495138 100644 --- a/acapy_agent/anoncreds/default/legacy_indy/routes.py +++ b/acapy_agent/anoncreds/default/legacy_indy/routes.py @@ -1 +1,35 @@ """Routes for Legacy Indy Registry.""" + +import logging +import re + +from ....core.event_bus import EventBus, EventWithMetadata +from ....core.profile import Profile +from ...events import REV_LIST_ENDORSED_UPDATE_FAILED_EVENT +from .recover import fix_and_publish_from_invalid_accum_err + +LOGGER = logging.getLogger(__name__) + + +def register_events(event_bus: EventBus): + """Subscribe to any events we need to support.""" + # If revocation list requires endorsement and fails to update, this event is emitted + # to trigger retry logic and notify of failure + event_bus.subscribe( + re.compile(REV_LIST_ENDORSED_UPDATE_FAILED_EVENT), + notify_issuer_about_update_failure_due_to_endorsement, + ) + + +async def notify_issuer_about_update_failure_due_to_endorsement( + profile: Profile, + event: EventWithMetadata, +) -> None: + """Notify issuer about a failure that couldn't be automatically recovered. + + Args: + profile (Profile): The profile context + event (EventWithMetadata): Failure message describing the endorsement failure + + """ + await fix_and_publish_from_invalid_accum_err(profile, event.payload["msg"]) diff --git a/acapy_agent/anoncreds/default/legacy_indy/tests/test_recover.py b/acapy_agent/anoncreds/default/legacy_indy/tests/test_recover.py index b504690c5e..4e900d1611 100644 --- a/acapy_agent/anoncreds/default/legacy_indy/tests/test_recover.py +++ b/acapy_agent/anoncreds/default/legacy_indy/tests/test_recover.py @@ -131,7 +131,7 @@ async def test_generate_ledger_rrrecovery_txn(self): ) assert result == {} - # Logs waring when ledger has revoked indexes not in wallet + # Logs warning when ledger has revoked indexes not in wallet with mock.patch( "acapy_agent.anoncreds.default.legacy_indy.recover.LOGGER" ) as mock_logger: diff --git a/acapy_agent/anoncreds/default/legacy_indy/tests/test_registry.py b/acapy_agent/anoncreds/default/legacy_indy/tests/test_registry.py index b11aa13aa7..a765f7cf5e 100644 --- a/acapy_agent/anoncreds/default/legacy_indy/tests/test_registry.py +++ b/acapy_agent/anoncreds/default/legacy_indy/tests/test_registry.py @@ -6,9 +6,6 @@ import pytest from anoncreds import ( - CredentialDefinition, - RevocationRegistryDefinition, - RevocationRegistryDefinitionPrivate, Schema, ) from base58 import alphabet @@ -47,6 +44,7 @@ RevRegDefValue, ) from ....models.schema import AnonCredsSchema, GetSchemaResult, SchemaResult +from ..recover import fix_ledger_entry B58 = alphabet if isinstance(alphabet, str) else alphabet.decode("ascii") INDY_DID = rf"^(did:sov:)?[{B58}]{{21,22}}$" @@ -1075,26 +1073,19 @@ async def test_register_revocation_list_with_create_transaction_option_and_auto_ ), ), ) - @mock.patch.object( - test_module.LegacyIndyRegistry, - "_sync_wallet_rev_list_with_issuer_cred_rev_records", - mock.CoroutineMock( - return_value=RevList( - issuer_id="CsQY9MGeD3CQP4EyuVFo5m", - current_accumulator="2 124C594B6B20E41B681E92B2C43FD165EA9E68BC3C9D63A82C8893124983CAE94 21 124C5341937827427B0A3A32113BD5E64FB7AB39BD3E5ABDD7970874501CA4897 6 5438CB6F442E2F807812FD9DC0C39AFF4A86B1E6766DBB5359E86A4D70401B0F 4 39D1CA5C4716FFC4FE0853C4FF7F081DFD8DF8D2C2CA79705211680AC77BF3A1 6 70504A5493F89C97C225B68310811A41AD9CD889301F238E93C95AD085E84191 4 39582252194D756D5D86D0EED02BF1B95CE12AED2FA5CD3C53260747D891993C", - revocation_list=[1, 0, 1, 0], - timestamp=1669640864487, - rev_reg_def_id="4xE68b6S5VRFrKMMG1U95M:4:4xE68b6S5VRFrKMMG1U95M:3:CL:59232:default:CL_ACCUM:4ae1cc6c-f6bd-486c-8057-88f2ce74e960", - ) - ), - ) @mock.patch( - "acapy_agent.anoncreds.default.legacy_indy.registry.generate_ledger_rrrecovery_txn", + "acapy_agent.anoncreds.default.legacy_indy.recover.generate_ledger_rrrecovery_txn", mock.CoroutineMock(return_value=MockTxn()), ) async def test_fix_ledger_entry(self, *_): mock_base_ledger = mock.MagicMock(BaseLedger, autospec=True) - mock_base_ledger.send_revoc_reg_entry = mock.CoroutineMock(return_value={}) + mock_base_ledger.send_revoc_reg_entry = mock.CoroutineMock( + return_value={ + "rev_reg_def_id": "rev_reg_def_id", + "signed_txn": "txn", + } + ) + mock_base_ledger.get_revoc_reg_delta = mock.CoroutineMock(return_value=({}, 123)) self.profile.context.injector.bind_instance( BaseLedger, mock_base_ledger, @@ -1119,7 +1110,7 @@ async def test_fix_ledger_entry(self, *_): ) self.profile.inject_or = mock.MagicMock() - result = await self.registry.fix_ledger_entry( + result = await fix_ledger_entry( self.profile, RevList( issuer_id="CsQY9MGeD3CQP4EyuVFo5m", @@ -1130,86 +1121,10 @@ async def test_fix_ledger_entry(self, *_): ), True, '{"reqSignature":{},"txn":{"data":{"data":{"alias":"Node1","blskey":"4N8aUNHSgjQVgkpm8nhNEfDf6txHznoYREg9kirmJrkivgL4oSEimFF6nsQ6M41QvhM2Z33nves5vfSn9n1UwNFJBYtWVnHYMATn76vLuL3zU88KyeAYcHfsih3He6UHcXDxcaecHVz6jhCYz1P2UZn2bDVruL5wXpehgBfBaLKm3Ba","blskey_pop":"RahHYiCvoNCtPTrVtP7nMC5eTYrsUA8WjXbdhNc8debh1agE9bGiJxWBXYNFbnJXoXhWFMvyqhqhRoq737YQemH5ik9oL7R4NTTCz2LEZhkgLJzB3QRQqJyBNyv7acbdHrAT8nQ9UkLbaVL9NBpnWXBTw4LEMePaSHEw66RzPNdAX1","client_ip":"172.17.0.2","client_port":9702,"node_ip":"172.17.0.2","node_port":9701,"services":["VALIDATOR"]},"dest":"Gw6pDLhcBcoQesN72qfotTgFa7cbuqZpkX3Xo6pLhPhv"},"metadata":{"from":"Th7MpTaRZVRYnPiabds81Y"},"type":"0"},"txnMetadata":{"seqNo":1,"txnId":"fea82e10e894419fe2bea7d96296a6d46f50f93f9eeda954ec461b2ed2950b62"},"ver":"1"}\n{"reqSignature":{},"txn":{"data":{"data":{"alias":"Node2","blskey":"37rAPpXVoxzKhz7d9gkUe52XuXryuLXoM6P6LbWDB7LSbG62Lsb33sfG7zqS8TK1MXwuCHj1FKNzVpsnafmqLG1vXN88rt38mNFs9TENzm4QHdBzsvCuoBnPH7rpYYDo9DZNJePaDvRvqJKByCabubJz3XXKbEeshzpz4Ma5QYpJqjk","blskey_pop":"Qr658mWZ2YC8JXGXwMDQTzuZCWF7NK9EwxphGmcBvCh6ybUuLxbG65nsX4JvD4SPNtkJ2w9ug1yLTj6fgmuDg41TgECXjLCij3RMsV8CwewBVgVN67wsA45DFWvqvLtu4rjNnE9JbdFTc1Z4WCPA3Xan44K1HoHAq9EVeaRYs8zoF5","client_ip":"172.17.0.2","client_port":9704,"node_ip":"172.17.0.2","node_port":9703,"services":["VALIDATOR"]},"dest":"8ECVSk179mjsjKRLWiQtssMLgp6EPhWXtaYyStWPSGAb"},"metadata":{"from":"EbP4aYNeTHL6q385GuVpRV"},"type":"0"},"txnMetadata":{"seqNo":2,"txnId":"1ac8aece2a18ced660fef8694b61aac3af08ba875ce3026a160acbc3a3af35fc"},"ver":"1"}\n{"reqSignature":{},"txn":{"data":{"data":{"alias":"Node3","blskey":"3WFpdbg7C5cnLYZwFZevJqhubkFALBfCBBok15GdrKMUhUjGsk3jV6QKj6MZgEubF7oqCafxNdkm7eswgA4sdKTRc82tLGzZBd6vNqU8dupzup6uYUf32KTHTPQbuUM8Yk4QFXjEf2Usu2TJcNkdgpyeUSX42u5LqdDDpNSWUK5deC5","blskey_pop":"QwDeb2CkNSx6r8QC8vGQK3GRv7Yndn84TGNijX8YXHPiagXajyfTjoR87rXUu4G4QLk2cF8NNyqWiYMus1623dELWwx57rLCFqGh7N4ZRbGDRP4fnVcaKg1BcUxQ866Ven4gw8y4N56S5HzxXNBZtLYmhGHvDtk6PFkFwCvxYrNYjh","client_ip":"172.17.0.2","client_port":9706,"node_ip":"172.17.0.2","node_port":9705,"services":["VALIDATOR"]},"dest":"DKVxG2fXXTU8yT5N7hGEbXB3dfdAnYv1JczDUHpmDxya"},"metadata":{"from":"4cU41vWW82ArfxJxHkzXPG"},"type":"0"},"txnMetadata":{"seqNo":3,"txnId":"7e9f355dffa78ed24668f0e0e369fd8c224076571c51e2ea8be5f26479edebe4"},"ver":"1"}\n{"reqSignature":{},"txn":{"data":{"data":{"alias":"Node4","blskey":"2zN3bHM1m4rLz54MJHYSwvqzPchYp8jkHswveCLAEJVcX6Mm1wHQD1SkPYMzUDTZvWvhuE6VNAkK3KxVeEmsanSmvjVkReDeBEMxeDaayjcZjFGPydyey1qxBHmTvAnBKoPydvuTAqx5f7YNNRAdeLmUi99gERUU7TD8KfAa6MpQ9bw","blskey_pop":"RPLagxaR5xdimFzwmzYnz4ZhWtYQEj8iR5ZU53T2gitPCyCHQneUn2Huc4oeLd2B2HzkGnjAff4hWTJT6C7qHYB1Mv2wU5iHHGFWkhnTX9WsEAbunJCV2qcaXScKj4tTfvdDKfLiVuU2av6hbsMztirRze7LvYBkRHV3tGwyCptsrP","client_ip":"172.17.0.2","client_port":9708,"node_ip":"172.17.0.2","node_port":9707,"services":["VALIDATOR"]},"dest":"4PS3EDQ3dW1tci1Bp6543CfuuebjFrg36kLAUcskGfaA"},"metadata":{"from":"TWwCRQRZ2ZHMJFn9TzLp7W"},"type":"0"},"txnMetadata":{"seqNo":4,"txnId":"aa5e817d7cc626170eca175822029339a444eb0ee8f0bd20d3b0b76e566fb008"},"ver":"1"}', - True, - "endorser_did", ) assert isinstance(result, tuple) - @mock.patch.object(CredentialDefinition, "load") - @mock.patch.object(RevocationRegistryDefinition, "load") - @mock.patch.object(RevocationRegistryDefinitionPrivate, "load") - @mock.patch.object( - IssuerCredRevRecord, - "query_by_ids", - return_value=[ - IssuerCredRevRecord( - state=IssuerCredRevRecord.STATE_REVOKED, - cred_ex_id="cred_ex_id", - rev_reg_id="4xE68b6S5VRFrKMMG1U95M:4:4xE68b6S5VRFrKMMG1U95M:3:CL:59232:default:CL_ACCUM:4ae1cc6c-f6bd-486c-8057-88f2ce74e960", - cred_rev_id="1", - ), - IssuerCredRevRecord( - state=IssuerCredRevRecord.STATE_REVOKED, - cred_ex_id="cred_ex_id", - rev_reg_id="4xE68b6S5VRFrKMMG1U95M:5:4xE68b6S5VRFrKMMG1U95M:3:CL:59232:default:CL_ACCUM:4ae1cc6c-f6bd-486c-8057-88f2ce74e960", - cred_rev_id="2", - ), - ], - ) - @mock.patch.object( - RevList, - "to_native", - return_value=mock.MagicMock( - update=mock.MagicMock(return_value=MockRevListEntry()) - ), - ) - @mock.patch.object(AskarAnonCredsProfileSession, "handle") - async def test_sync_wallet_rev_list_with_issuer_cred_rev_records( - self, mock_handle, *_ - ): - async with self.profile.session() as session: - # Matching revocations and rev_list - mock_handle.fetch = mock.CoroutineMock( - side_effect=[ - MockRevRegDefEntry(), - MockCredDefEntry(), - mock.CoroutineMock(return_value=None), - ] - ) - result = await self.registry._sync_wallet_rev_list_with_issuer_cred_rev_records( - session, - RevList( - issuer_id="CsQY9MGeD3CQP4EyuVFo5m", - current_accumulator="21 124C594B6B20E41B681E92B2C43FD165EA9E68BC3C9D63A82C8893124983CAE94 21 124C5341937827427B0A3A32113BD5E64FB7AB39BD3E5ABDD7970874501CA4897 6 5438CB6F442E2F807812FD9DC0C39AFF4A86B1E6766DBB5359E86A4D70401B0F 4 39D1CA5C4716FFC4FE0853C4FF7F081DFD8DF8D2C2CA79705211680AC77BF3A1 6 70504A5493F89C97C225B68310811A41AD9CD889301F238E93C95AD085E84191 4 39582252194D756D5D86D0EED02BF1B95CE12AED2FA5CD3C53260747D891993C", - revocation_list=[0, 1, 1, 0], - timestamp=1669640864487, - rev_reg_def_id="4xE68b6S5VRFrKMMG1U95M:4:4xE68b6S5VRFrKMMG1U95M:3:CL:59232:default:CL_ACCUM:4ae1cc6c-f6bd-486c-8057-88f2ce74e960", - ), - ) - assert isinstance(result, RevList) - # Non-matching revocations and rev_list - mock_handle.fetch = mock.CoroutineMock( - side_effect=[ - MockRevRegDefEntry(), - MockCredDefEntry(), - mock.CoroutineMock(return_value=None), - MockRevListEntry(), - ] - ) - mock_handle.replace = mock.CoroutineMock(return_value=None) - result = await self.registry._sync_wallet_rev_list_with_issuer_cred_rev_records( - session, - RevList( - issuer_id="CsQY9MGeD3CQP4EyuVFo5m", - current_accumulator="21 124C594B6B20E41B681E92B2C43FD165EA9E68BC3C9D63A82C8893124983CAE94 21 124C5341937827427B0A3A32113BD5E64FB7AB39BD3E5ABDD7970874501CA4897 6 5438CB6F442E2F807812FD9DC0C39AFF4A86B1E6766DBB5359E86A4D70401B0F 4 39D1CA5C4716FFC4FE0853C4FF7F081DFD8DF8D2C2CA79705211680AC77BF3A1 6 70504A5493F89C97C225B68310811A41AD9CD889301F238E93C95AD085E84191 4 39582252194D756D5D86D0EED02BF1B95CE12AED2FA5CD3C53260747D891993C", - revocation_list=[0, 1, 0, 0], - timestamp=1669640864487, - rev_reg_def_id="4xE68b6S5VRFrKMMG1U95M:4:4xE68b6S5VRFrKMMG1U95M:3:CL:59232:default:CL_ACCUM:4ae1cc6c-f6bd-486c-8057-88f2ce74e960", - ), - ) - assert isinstance(result, RevList) - async def test_get_schem_info(self): result = await self.registry.get_schema_info_by_id( self.profile, diff --git a/acapy_agent/anoncreds/events.py b/acapy_agent/anoncreds/events.py index fd48a2ccd9..82ecbba340 100644 --- a/acapy_agent/anoncreds/events.py +++ b/acapy_agent/anoncreds/events.py @@ -69,6 +69,12 @@ # If retries continue to fail, this will notify the issuer that intervention is required INTERVENTION_REQUIRED_EVENT = "anoncreds::revocation-registry::intervention-required" +# If revocation list requires endorsement and fails to update, this event is emitted to +# trigger retry logic and notify of failure +REV_LIST_ENDORSED_UPDATE_FAILED_EVENT = ( + "anoncreds::revocation-list::endorsed-update-failed" +) + class BaseEventPayload(Protocol): """Base event payload.""" diff --git a/acapy_agent/anoncreds/revocation/__init__.py b/acapy_agent/anoncreds/revocation/__init__.py index 65b3ab7043..444397e5d4 100644 --- a/acapy_agent/anoncreds/revocation/__init__.py +++ b/acapy_agent/anoncreds/revocation/__init__.py @@ -5,7 +5,7 @@ """ from .manager import RevocationManager, RevocationManagerError -from .recover import RevocRecoveryException, fetch_txns, generate_ledger_rrrecovery_txn +from .recover import RevocRecoveryException from .revocation import ( AnonCredsRevocation, AnonCredsRevocationError, @@ -21,6 +21,4 @@ "RevocRecoveryException", "RevocationManager", "RevocationManagerError", - "fetch_txns", - "generate_ledger_rrrecovery_txn", ] diff --git a/acapy_agent/anoncreds/revocation/manager.py b/acapy_agent/anoncreds/revocation/manager.py index 4562b9609c..3108e3d381 100644 --- a/acapy_agent/anoncreds/revocation/manager.py +++ b/acapy_agent/anoncreds/revocation/manager.py @@ -2,7 +2,7 @@ import logging from collections.abc import Mapping, Sequence -from typing import TYPE_CHECKING, Optional, Tuple +from typing import Optional, Tuple from ...core.error import BaseError from ...core.profile import Profile @@ -14,9 +14,6 @@ from ..models.issuer_cred_rev_record import IssuerCredRevRecord from .revocation import AnonCredsRevocation -if TYPE_CHECKING: - from ..default.legacy_indy.registry import LegacyIndyRegistry - class RevocationManagerError(BaseError): """Revocation manager error.""" @@ -203,16 +200,6 @@ async def update_rev_reg_revoked_state( f"No revocation list found for revocation registry id {rev_reg_def_id}" ) - indy_registry = LegacyIndyRegistry() - - if await indy_registry.supports(rev_reg_def_id): - return await indy_registry.fix_ledger_entry( - self._profile, - rev_list, - apply_ledger_update, - genesis_transactions, - ) - raise RevocationManagerError( "Indy registry does not support revocation registry " f"identified by {rev_reg_def_id}" diff --git a/acapy_agent/anoncreds/revocation/recover.py b/acapy_agent/anoncreds/revocation/recover.py index a664c3e0ac..5557fcd902 100644 --- a/acapy_agent/anoncreds/revocation/recover.py +++ b/acapy_agent/anoncreds/revocation/recover.py @@ -1,120 +1,9 @@ -"""Recover a revocation registry.""" +"""Recover a revocation registry. -import hashlib -import importlib -import logging -import tempfile -import time - -import aiohttp -import base58 - -LOGGER = logging.getLogger(__name__) - - -""" -This module calculates a new ledger accumulator, based on the revocation status -on the ledger vs revocations recorded in the wallet. -The calculated transaction can be written to the ledger to get the ledger back -in sync with the wallet. -This function can be used if there were previous revocation errors (i.e. the -credential revocation was successfully written to the wallet but the ledger write -failed.) +This module contains general exceptions or helper functions related to revocation +registry recovery that are not specific to any one implementation. """ -# TODO This should probably be moved to an Indy plugin - class RevocRecoveryException(Exception): - """Raise exception generating the recovery transaction.""" - - -async def fetch_txns(genesis_txns, registry_id): - """Fetch tails file and revocation registry information.""" - try: - vdr_module = importlib.import_module("indy_vdr") - credx_module = importlib.import_module("indy_credx") - except Exception as e: - raise RevocRecoveryException(f"Failed to import library {e}") from e - - pool = await vdr_module.open_pool(transactions=genesis_txns) - LOGGER.debug("Connected to pool") - - LOGGER.debug("Fetch registry: %s", registry_id) - fetch = vdr_module.ledger.build_get_revoc_reg_def_request(None, registry_id) - result = await pool.submit_request(fetch) - if not result["data"]: - raise RevocRecoveryException(f"Registry definition not found for {registry_id}") - data = result["data"] - data["ver"] = "1.0" - defn = credx_module.RevocationRegistryDefinition.load(data) - LOGGER.debug("Tails URL: %s", defn.tails_location) - - async with aiohttp.ClientSession() as session: - data = await session.get(defn.tails_location) - tails_data = await data.read() - tails_hash = base58.b58encode(hashlib.sha256(tails_data).digest()).decode("utf-8") - if tails_hash != defn.tails_hash: - raise RevocRecoveryException( - f"Tails hash mismatch {tails_hash} {defn.tails_hash}" - ) - else: - LOGGER.debug("Checked tails hash: %s", tails_hash) - tails_temp = tempfile.NamedTemporaryFile(delete=False) - tails_temp.write(tails_data) - tails_temp.close() - - to_timestamp = int(time.time()) - fetch = vdr_module.ledger.build_get_revoc_reg_delta_request( - None, registry_id, None, to_timestamp - ) - result = await pool.submit_request(fetch) - if not result["data"]: - raise RevocRecoveryException("Error fetching delta from ledger") - - accum_to = result["data"]["value"]["accum_to"] - accum_to["ver"] = "1.0" - delta = credx_module.RevocationRegistryDelta.load(accum_to) - registry = credx_module.RevocationRegistry.load(accum_to) - LOGGER.debug("Ledger registry state: %s", registry.to_json()) - revoked = set(result["data"]["value"]["revoked"]) - LOGGER.debug("Ledger revoked indexes: %s", revoked) - - return defn, registry, delta, revoked, tails_temp - - -async def generate_ledger_rrrecovery_txn( - genesis_txns, registry_id, set_revoked, cred_def, rev_reg_def_private -): - """Generate a new ledger accum entry, based on wallet vs ledger revocation state.""" - new_delta = None - - ledger_data = await fetch_txns(genesis_txns, registry_id) - if not ledger_data: - return new_delta - defn, registry, _delta, prev_revoked, tails_temp = ledger_data - - set_revoked = set(set_revoked) - mismatch = prev_revoked - set_revoked - if mismatch: - LOGGER.warning( - "Credential index(es) revoked on the ledger, but not in wallet: %s", - mismatch, - ) - - updates = set_revoked - prev_revoked - if not updates: - LOGGER.debug("No updates to perform") - else: - LOGGER.debug("New revoked indexes: %s", updates) - - LOGGER.debug("tails_temp: %s", tails_temp.name) - update_registry = registry.copy() - new_delta = update_registry.update( - cred_def, defn, rev_reg_def_private, [], updates - ) - - LOGGER.debug("New delta:") - LOGGER.debug(new_delta.to_json()) - - return new_delta + """Raise exception performing recovery.""" diff --git a/acapy_agent/protocols/endorse_transaction/v1_0/handlers/endorsed_transaction_response_handler.py b/acapy_agent/protocols/endorse_transaction/v1_0/handlers/endorsed_transaction_response_handler.py index 42762c34cb..dabffb25d6 100644 --- a/acapy_agent/protocols/endorse_transaction/v1_0/handlers/endorsed_transaction_response_handler.py +++ b/acapy_agent/protocols/endorse_transaction/v1_0/handlers/endorsed_transaction_response_handler.py @@ -1,5 +1,7 @@ """Endorsed transaction response handler.""" +from .....anoncreds.base import AnonCredsRegistrationError +from .....anoncreds.events import REV_LIST_ENDORSED_UPDATE_FAILED_EVENT from .....messaging.base_handler import ( BaseHandler, BaseResponder, @@ -32,12 +34,21 @@ async def handle(self, context: RequestContext, responder: BaseResponder): raise HandlerException("No connection established") async def send_failed_transaction_event(err_msg: str): - await notify_rev_reg_entry_txn_failed(context.profile, err_msg) + is_anoncreds = context.profile.settings.get("wallet.type") in ( + "askar-anoncreds", + "kanon-anoncreds", + ) + if is_anoncreds: + await context.profile.notify( + REV_LIST_ENDORSED_UPDATE_FAILED_EVENT, {"msg": err_msg} + ) + else: + await notify_rev_reg_entry_txn_failed(context.profile, err_msg) mgr = TransactionManager(context.profile) try: transaction = await mgr.receive_endorse_response(context.message) - except TransactionManagerError as err: + except (TransactionManagerError, AnonCredsRegistrationError) as err: self._logger.exception("Error receiving endorsed transaction response") await send_failed_transaction_event(str(err)) raise HandlerException(str(err)) @@ -54,7 +65,11 @@ async def send_failed_transaction_event(err_msg: str): transaction_acknowledgement_message, connection_id=transaction.connection_id, ) - except (StorageError, TransactionManagerError) as err: + except ( + StorageError, + TransactionManagerError, + AnonCredsRegistrationError, + ) as err: self._logger.exception(err) await send_failed_transaction_event(str(err)) raise HandlerException(str(err)) diff --git a/acapy_agent/revocation_anoncreds/__init__.py b/acapy_agent/revocation_anoncreds/__init__.py index d7d94c7165..3fc8140d24 100644 --- a/acapy_agent/revocation_anoncreds/__init__.py +++ b/acapy_agent/revocation_anoncreds/__init__.py @@ -11,11 +11,7 @@ IssuerCredRevRecordSchemaAnonCreds, ) from ..anoncreds.revocation.manager import RevocationManager, RevocationManagerError -from ..anoncreds.revocation.recover import ( - RevocRecoveryException, - fetch_txns, - generate_ledger_rrrecovery_txn, -) +from ..anoncreds.revocation.recover import RevocRecoveryException from ..anoncreds.revocation.revocation import ( AnonCredsRevocation, AnonCredsRevocationError, @@ -39,6 +35,4 @@ "RevocRecoveryException", "RevocationManager", "RevocationManagerError", - "fetch_txns", - "generate_ledger_rrrecovery_txn", ]