AnonCreds module¶

This module provides Hedera AnonCreds Registry API.

Please note that only AnonCreds Verifiable Data Registry (VDR) capabilities are provided by SDK, so it's intended to be used along with actual AnonCreds implementations (such as anoncreds-rs Python wrapper).

AnonCreds Registry¶

`HederaAnonCredsRegistry` ¶

Anoncreds objects registry (resolver + registrar) implementation that leverage Hedera HCS as VDR.

Parameters:

Name	Type	Description	Default
`client`	`Client`	Hedera Client	required
`cache_instance`	`Cache[str, object] \| None`	Custom cache instance. If not provided, in-memory cache is used	`None`

Source code in hiero_did_sdk_python/anoncreds/hedera_anoncreds_registry.py

class HederaAnonCredsRegistry:
    """Anoncreds objects registry (resolver + registrar) implementation that leverage Hedera HCS as VDR.

    Args:
        client: Hedera Client
        cache_instance: Custom cache instance. If not provided, in-memory cache is used
    """

    def __init__(
        self,
        client: Client,
        cache_instance: Cache[str, object] | None = None,
    ):
        self._client = client
        self._hcs_file_service = HcsFileService(client)
        self._hcs_topic_service = HcsTopicService(client)

        cache_instance = cache_instance or MemoryCache[str, object]()

        self._schema_cache: Cache[str, AnonCredsSchema] = cast(Cache[str, AnonCredsSchema], cache_instance)
        self._cred_def_cache: Cache[str, AnonCredsCredDef] = cast(Cache[str, AnonCredsCredDef], cache_instance)
        self._rev_reg_def_cache: Cache[str, RevRegDefWithHcsMetadata] = cast(
            Cache[str, RevRegDefWithHcsMetadata], cache_instance
        )
        self._rev_reg_entries_messages_cache = cast(Cache[str, list[HcsMessageWithResponseMetadata]], cache_instance)

    async def get_schema(self, schema_id: str) -> GetSchemaResult:
        """Get a schema from the registry.

        Args:
            schema_id: Schema ID to resolver

        Returns:
            object: Schema resolution result
        """
        try:
            parsed_identifier = parse_anoncreds_identifier(schema_id)

            if parsed_identifier.object_type != AnonCredsObjectType.SCHEMA:
                return GetSchemaResult(
                    schema_id=schema_id,
                    resolution_metadata={
                        "error": "notFound",
                        "message": f"AnonCreds Schema id '{schema_id}' is invalid",
                    },
                    schema_metadata={},
                )

            schema_topic_id = parsed_identifier.topic_id

            cached_schema = self._schema_cache.get(schema_topic_id)

            if cached_schema:
                schema = cached_schema
            else:
                schema_payload = await self._hcs_file_service.resolve_file(parsed_identifier.topic_id)
                schema = AnonCredsSchema.from_json(schema_payload.decode()) if schema_payload else None

                if not schema:
                    return GetSchemaResult(
                        schema_id=schema_id,
                        resolution_metadata={
                            "error": "notFound",
                            "message": f"AnonCreds schema with id '{schema_id}' not found",
                        },
                        schema_metadata={},
                    )

                self._schema_cache.set(schema_topic_id, schema)

            return GetSchemaResult(schema=schema, schema_id=schema_id, resolution_metadata={}, schema_metadata={})
        except Exception as error:
            LOGGER.error(f"Error on retrieving AnonCreds Schema: {error!s}")
            return GetSchemaResult(
                schema_id=schema_id,
                resolution_metadata={
                    "error": "otherError",
                    "message": f"unable to resolve schema: ${error!s}",
                },
                schema_metadata={},
            )

    async def register_schema(self, schema: AnonCredsSchema, issuer_key_der: str) -> RegisterSchemaResult:
        """Register Schema.

        Args:
            schema: Schema object to register
            issuer_key_der: Issuer private key encoded in DER format

        Returns:
            object: Schema registration result
        """
        try:
            hcs_file_payload = schema.to_json().encode()
            schema_topic_id = await self._hcs_file_service.submit_file(hcs_file_payload, issuer_key_der)

            return RegisterSchemaResult(
                schema_state=SchemaState(
                    state="finished",
                    schema=schema,
                    schema_id=build_anoncreds_identifier(schema.issuer_id, schema_topic_id, AnonCredsObjectType.SCHEMA),
                ),
                schema_metadata={},
                registration_metadata={},
            )
        except Exception as error:
            LOGGER.error(f"Error on registering Anoncreds Schema: {error!s}")
            return RegisterSchemaResult(
                schema_state=SchemaState(state="failed", schema=schema, reason=f"unknownError: ${error!s}"),
                schema_metadata={},
                registration_metadata={},
            )

    async def get_cred_def(self, cred_def_id: str) -> GetCredDefResult:
        """Get a credential definition from the registry.

        Args:
            cred_def_id: Credential definition ID to resolve

        Returns:
            object: Credential definition resolution result
        """
        try:
            parsed_identifier = parse_anoncreds_identifier(cred_def_id)

            if parsed_identifier.object_type != AnonCredsObjectType.PUBLIC_CRED_DEF:
                return GetCredDefResult(
                    credential_definition_id=cred_def_id,
                    resolution_metadata={
                        "error": "notFound",
                        "message": f"Credential definition id '{cred_def_id}' is invalid",
                    },
                    credential_definition_metadata={},
                )

            cred_def_topic_id = parsed_identifier.topic_id

            cached_cred_def = self._cred_def_cache.get(cred_def_topic_id)

            if cached_cred_def:
                cred_def = cached_cred_def
            else:
                cred_def_payload = await self._hcs_file_service.resolve_file(cred_def_topic_id)
                cred_def = AnonCredsCredDef.from_json(cred_def_payload.decode()) if cred_def_payload else None

                if not cred_def:
                    return GetCredDefResult(
                        credential_definition_id=cred_def_id,
                        resolution_metadata={
                            "error": "notFound",
                            "message": f"AnonCreds credential definition with id '{cred_def_id}' not found",
                        },
                        credential_definition_metadata={},
                    )

                self._cred_def_cache.set(cred_def_topic_id, cred_def)

            return GetCredDefResult(
                credential_definition=cred_def,
                credential_definition_id=cred_def_id,
                resolution_metadata={},
                credential_definition_metadata={},
            )
        except Exception as error:
            LOGGER.error(f"Error on retrieving AnonCreds credential definition: {error!s}")
            return GetCredDefResult(
                credential_definition_id=cred_def_id,
                resolution_metadata={
                    "error": "otherError",
                    "message": f"unable to resolve credential definition: ${error!s}",
                },
                credential_definition_metadata={},
            )

    async def register_cred_def(self, cred_def: AnonCredsCredDef, issuer_key_der: str) -> RegisterCredDefResult:
        """Register Credential Definition.

        Args:
            cred_def: Credential definition object to register
            issuer_key_der: Issuer private key encoded in DER

        Returns:
            object: Credential definition registration result
        """
        try:
            hcs_file_payload = cred_def.to_json().encode()
            cred_def_topic_id = await self._hcs_file_service.submit_file(hcs_file_payload, issuer_key_der)

            return RegisterCredDefResult(
                credential_definition_state=CredDefState(
                    state="finished",
                    credential_definition=cred_def,
                    credential_definition_id=build_anoncreds_identifier(
                        cred_def.issuer_id, cred_def_topic_id, AnonCredsObjectType.PUBLIC_CRED_DEF
                    ),
                ),
                registration_metadata={},
                credential_definition_metadata={},
            )
        except Exception as error:
            LOGGER.error(f"Error on registering Anoncreds Cred Def: {error!s}")
            return RegisterCredDefResult(
                credential_definition_state=CredDefState(
                    state="failed", credential_definition=cred_def, reason=f"unknownError: ${error!s}"
                ),
                registration_metadata={},
                credential_definition_metadata={},
            )

    async def get_rev_reg_def(self, revocation_registry_definition_id: str) -> GetRevRegDefResult:
        """Get a revocation registry definition from the registry.

        Args:
            revocation_registry_definition_id: Revocation registry definition ID to resolve

        Returns:
            object: Revocation registry definition resolution result
        """
        try:
            parsed_identifier = parse_anoncreds_identifier(revocation_registry_definition_id)

            if parsed_identifier.object_type != AnonCredsObjectType.REV_REG:
                return GetRevRegDefResult(
                    revocation_registry_definition_id=revocation_registry_definition_id,
                    resolution_metadata={
                        "error": "notFound",
                        "message": f"Revocation registry id '{revocation_registry_definition_id}' is invalid",
                    },
                    revocation_registry_definition_metadata={},
                )

            rev_reg_def_topic_id = parsed_identifier.topic_id

            cached_rev_reg_def_with_metadata = self._rev_reg_def_cache.get(rev_reg_def_topic_id)

            if cached_rev_reg_def_with_metadata:
                rev_reg_def_with_metadata = cached_rev_reg_def_with_metadata
            else:
                rev_reg_def_payload = await self._hcs_file_service.resolve_file(rev_reg_def_topic_id)
                rev_reg_def_with_metadata = (
                    RevRegDefWithHcsMetadata.from_json(rev_reg_def_payload.decode()) if rev_reg_def_payload else None
                )

                if not rev_reg_def_with_metadata:
                    return GetRevRegDefResult(
                        revocation_registry_definition_id=revocation_registry_definition_id,
                        resolution_metadata={
                            "error": "notFound",
                            "message": f"AnonCreds revocation registry with id '{revocation_registry_definition_id}' not found",
                        },
                        revocation_registry_definition_metadata={},
                    )

                self._rev_reg_def_cache.set(rev_reg_def_topic_id, rev_reg_def_with_metadata)

            return GetRevRegDefResult(
                revocation_registry_definition=rev_reg_def_with_metadata.rev_reg_def,
                revocation_registry_definition_id=revocation_registry_definition_id,
                resolution_metadata={},
                revocation_registry_definition_metadata={**rev_reg_def_with_metadata.hcs_metadata},
            )
        except Exception as error:
            LOGGER.error(f"Error on retrieving AnonCreds revocation registry definition: {error!s}")
            return GetRevRegDefResult(
                revocation_registry_definition_id=revocation_registry_definition_id,
                resolution_metadata={
                    "error": "otherError",
                    "message": f"unable to resolve revocation registry definition: ${error!s}",
                },
                revocation_registry_definition_metadata={},
            )

    async def register_rev_reg_def(
        self, rev_reg_def: AnonCredsRevRegDef, issuer_key_der: str
    ) -> RegisterRevRegDefResult:
        """Register Revocation registry definition.

        Args:
            rev_reg_def: Revocation registry definition object to register
            issuer_key_der: Issuer private key encoded in DER

        Returns:
            object: Revocation registry definition registration result
        """
        try:
            issuer_key = PrivateKey.from_string(issuer_key_der)

            entries_topic_options = HcsTopicOptions(submit_key=issuer_key.public_key())
            entries_topic_id = await self._hcs_topic_service.create_topic(entries_topic_options, [issuer_key])

            rev_reg_def_with_metadata = RevRegDefWithHcsMetadata(
                rev_reg_def=rev_reg_def, hcs_metadata={"entries_topic_id": entries_topic_id}
            )

            hcs_file_payload = rev_reg_def_with_metadata.to_json().encode()
            rev_reg_def_topic_id = await self._hcs_file_service.submit_file(hcs_file_payload, issuer_key_der)

            # We want to cache registry definition right away
            # Helps to avoid potential cases where issuer pushes rev entries immediately but registry definition data (HCS-1 messages) is not propagated to mirror nodes yet
            self._rev_reg_def_cache.set(rev_reg_def_topic_id, rev_reg_def_with_metadata)

            return RegisterRevRegDefResult(
                revocation_registry_definition_state=RevRegDefState(
                    state="finished",
                    revocation_registry_definition=rev_reg_def,
                    revocation_registry_definition_id=build_anoncreds_identifier(
                        rev_reg_def.issuer_id, rev_reg_def_topic_id, AnonCredsObjectType.REV_REG
                    ),
                ),
                registration_metadata={},
                revocation_registry_definition_metadata={**rev_reg_def_with_metadata.hcs_metadata},
            )
        except Exception as error:
            LOGGER.error(f"Error on registering Anoncreds Revocation registry definition: {error!s}")
            return RegisterRevRegDefResult(
                revocation_registry_definition_state=RevRegDefState(
                    state="failed", revocation_registry_definition=rev_reg_def, reason=f"unknownError: ${error!s}"
                ),
                registration_metadata={},
                revocation_registry_definition_metadata={},
            )

    async def get_rev_list(self, rev_reg_id: str, timestamp: int) -> GetRevListResult:
        """Get a revocation list from the registry.

        Args:
            rev_reg_id: Revocation registry ID
            timestamp: Timestamp to resolve revocation list for

        Returns:
            object: Revocation list resolution result
        """
        try:
            rev_reg_def_result = await self.get_rev_reg_def(rev_reg_id)

            if not rev_reg_def_result.revocation_registry_definition:
                return GetRevListResult(
                    revocation_registry_id=rev_reg_id,
                    resolution_metadata={
                        "error": "notFound",
                        "message": f"AnonCreds revocation registry with id '{rev_reg_id}' not found",
                    },
                    revocation_list_metadata={},
                )

            rev_reg_def = rev_reg_def_result.revocation_registry_definition
            entries_topic_id = rev_reg_def_result.revocation_registry_definition_metadata.get("entries_topic_id")

            if not entries_topic_id:
                return GetRevListResult(
                    revocation_registry_id=rev_reg_id,
                    resolution_metadata={
                        "error": "notFound",
                        "message": "Entries topic ID is missing from revocation registry metadata",
                    },
                    revocation_list_metadata={},
                )

            cached_messages = self._rev_reg_entries_messages_cache.get(entries_topic_id)
            if cached_messages:
                last_cached_message_timestamp = cached_messages[-1].consensus_timestamp

                if last_cached_message_timestamp.seconds >= timestamp:
                    borderline_timestamp = Timestamp(seconds=timestamp, nanos=0)
                    entries_messages = filter(
                        lambda message: message.consensus_timestamp.seconds < timestamp
                        or message.consensus_timestamp == borderline_timestamp,
                        cached_messages,
                    )
                    entries = [cast(AnonCredsRevRegEntry, message.message) for message in entries_messages]

                    # This means that requested timestamp is before the actual registration of rev list
                    # In such case, we want to return initial state for the list (by adding first message to entries)
                    if len(entries) == 0:
                        entries.append(cast(AnonCredsRevRegEntry, cached_messages[0].message))

                    return GetRevListResult(
                        revocation_registry_id=rev_reg_id,
                        revocation_list=AnonCredsRevList.from_rev_reg_entries(
                            entries, rev_reg_id, rev_reg_def, timestamp
                        ),
                        resolution_metadata={},
                        revocation_list_metadata={},
                    )
                else:
                    new_messages = await HcsMessageResolver(
                        topic_id=entries_topic_id,
                        message_type=HcsRevRegEntryMessage,
                        timestamp_from=last_cached_message_timestamp,
                        timestamp_to=Timestamp(seconds=timestamp, nanos=0),
                        include_response_metadata=True,
                    ).execute(self._client)

                    # Note: 'chain' function is used instead of lists sum due to significantly better performance on large lists
                    # See: https://docs.python.org/3/library/itertools.html, https://stackoverflow.com/a/41772165
                    entries_messages = (
                        list(chain(cached_messages, cast(list[HcsMessageWithResponseMetadata], new_messages)))
                        if len(new_messages) > 0
                        else cached_messages
                    )

                    self._rev_reg_entries_messages_cache.set(entries_topic_id, entries_messages)

                    entries = [cast(AnonCredsRevRegEntry, message.message) for message in entries_messages]

                    return GetRevListResult(
                        revocation_registry_id=rev_reg_id,
                        revocation_list=AnonCredsRevList.from_rev_reg_entries(
                            entries, rev_reg_id, rev_reg_def, timestamp
                        ),
                        resolution_metadata={},
                        revocation_list_metadata={},
                    )

            entries_messages = await HcsMessageResolver(
                topic_id=entries_topic_id,
                message_type=HcsRevRegEntryMessage,
                timestamp_to=Timestamp(seconds=timestamp, nanos=0),
                include_response_metadata=True,
            ).execute(self._client)

            if len(entries_messages) == 0:
                # If returned entries list is empty, we need to fetch the first message and check if list is registered
                # It's possible that requested timestamp is before the actual registration of rev list -> we want to return initial state for the list (by adding first message to entries)

                # The second request looks redundant here, but it should be the rare case that will be subsequently handled by cache
                entries_messages = await HcsMessageResolver(
                    topic_id=entries_topic_id,
                    message_type=HcsRevRegEntryMessage,
                    limit=1,
                    include_response_metadata=True,
                ).execute(self._client)

                if len(entries_messages) == 0:
                    return GetRevListResult(
                        revocation_registry_id=rev_reg_id,
                        resolution_metadata={
                            "error": "notFound",
                            "message": f"Registered revocation list for registry id '{rev_reg_id}' is not found",
                        },
                        revocation_list_metadata={},
                    )

            entries_messages = cast(list[HcsMessageWithResponseMetadata], entries_messages)
            self._rev_reg_entries_messages_cache.set(entries_topic_id, entries_messages)

            entries = [cast(AnonCredsRevRegEntry, message.message) for message in entries_messages]

            return GetRevListResult(
                revocation_registry_id=rev_reg_id,
                revocation_list=AnonCredsRevList.from_rev_reg_entries(entries, rev_reg_id, rev_reg_def, timestamp),
                resolution_metadata={},
                revocation_list_metadata={},
            )

        except Exception as error:
            LOGGER.error(f"Error on retrieving AnonCreds revocation list: {error!s}")
            return GetRevListResult(
                revocation_registry_id=rev_reg_id,
                resolution_metadata={
                    "error": "otherError",
                    "message": f"Unable to resolve revocation list: ${error!s}",
                },
                revocation_list_metadata={},
            )

    async def register_rev_list(self, rev_list: AnonCredsRevList, issuer_key_der: str) -> RegisterRevListResult:
        """Register Revocation list.

        Args:
            rev_list: Revocation list object to register
            issuer_key_der: Issuer private key encoded in DER

        Returns: Revocation list registration result
        """
        try:
            return await self._submit_rev_list_entry(rev_list, issuer_key_der)
        except Exception as error:
            LOGGER.error(f"Error on registering Anoncreds revocation list: {error!s}")
            return RegisterRevListResult(
                revocation_list_state=RevListState(
                    state="failed", revocation_list=rev_list, reason=f"unknownError: ${error!s}"
                ),
                registration_metadata={},
                revocation_list_metadata={},
            )

    async def update_rev_list(
        self, prev_list: AnonCredsRevList, curr_list: AnonCredsRevList, revoked: Sequence[int], issuer_key_der: str
    ) -> RegisterRevListResult:
        """Update Revocation list.

        Args:
            prev_list: Previous Revocation list object
            curr_list: Current Revocation list object
            revoked: Revoked credential indexes
            issuer_key_der: Issuer private key encoded in DER

        Returns: Revocation list update result
        """
        try:
            return await self._submit_rev_list_entry(curr_list, issuer_key_der, prev_list, revoked)
        except Exception as error:
            LOGGER.error(f"Error on updating Anoncreds revocation list: {error!s}")
            return RegisterRevListResult(
                revocation_list_state=RevListState(
                    state="failed", revocation_list=curr_list, reason=f"unknownError: ${error!s}"
                ),
                registration_metadata={},
                revocation_list_metadata={},
            )

    async def _submit_rev_list_entry(
        self,
        rev_list: AnonCredsRevList,
        issuer_key_der: str,
        prev_list: AnonCredsRevList | None = None,
        revoked: Sequence[int] | None = None,
    ) -> RegisterRevListResult:
        if prev_list and prev_list.rev_reg_def_id != rev_list.rev_reg_def_id:
            return RegisterRevListResult(
                revocation_list_state=RevListState(
                    state="failed",
                    revocation_list=rev_list,
                    reason=f"Revocation registry ids do not match for previous and current list: '{prev_list.rev_reg_def_id}' != '{rev_list.rev_reg_def_id}'",
                ),
                registration_metadata={},
                revocation_list_metadata={},
            )

        rev_reg_def_result = await self.get_rev_reg_def(rev_list.rev_reg_def_id)

        if not rev_reg_def_result.revocation_registry_definition:
            return RegisterRevListResult(
                revocation_list_state=RevListState(
                    state="failed",
                    revocation_list=rev_list,
                    reason=f"AnonCreds revocation registry with id '{rev_list.rev_reg_def_id}' not found",
                ),
                registration_metadata={},
                revocation_list_metadata={},
            )

        entries_topic_id = rev_reg_def_result.revocation_registry_definition_metadata.get("entries_topic_id")

        if not entries_topic_id:
            return RegisterRevListResult(
                revocation_list_state=RevListState(
                    state="failed",
                    revocation_list=rev_list,
                    reason="notFound: Entries topic ID is missing from revocation registry metadata",
                ),
                registration_metadata={},
                revocation_list_metadata={},
            )

        entry_message = HcsRevRegEntryMessage(
            value=RevRegEntryValue(
                prev_accum=prev_list.current_accumulator if prev_list else None,
                accum=rev_list.current_accumulator,
                revoked=list(revoked) if revoked else None,
            )
        )

        def build_message_submit_transaction(
            message_submit_transaction: TopicMessageSubmitTransaction,
        ) -> Transaction:
            message_submit_transaction.transaction_fee = MAX_TRANSACTION_FEE.to_tinybars()  # pyright: ignore [reportAttributeAccessIssue]
            return message_submit_transaction.freeze_with(self._client).sign(PrivateKey.from_string(issuer_key_der))

        await HcsMessageTransaction(entries_topic_id, entry_message, build_message_submit_transaction).execute(
            self._client
        )

        return RegisterRevListResult(
            revocation_list_state=RevListState(state="finished", revocation_list=rev_list),
            registration_metadata={},
            revocation_list_metadata={},
        )

`get_cred_def(cred_def_id)` `async` ¶

Get a credential definition from the registry.

Parameters:

Name	Type	Description	Default
`cred_def_id`	`str`	Credential definition ID to resolve	required

Returns:

Name	Type	Description
`object`	`GetCredDefResult`	Credential definition resolution result

Source code in hiero_did_sdk_python/anoncreds/hedera_anoncreds_registry.py

async def get_cred_def(self, cred_def_id: str) -> GetCredDefResult:
    """Get a credential definition from the registry.

    Args:
        cred_def_id: Credential definition ID to resolve

    Returns:
        object: Credential definition resolution result
    """
    try:
        parsed_identifier = parse_anoncreds_identifier(cred_def_id)

        if parsed_identifier.object_type != AnonCredsObjectType.PUBLIC_CRED_DEF:
            return GetCredDefResult(
                credential_definition_id=cred_def_id,
                resolution_metadata={
                    "error": "notFound",
                    "message": f"Credential definition id '{cred_def_id}' is invalid",
                },
                credential_definition_metadata={},
            )

        cred_def_topic_id = parsed_identifier.topic_id

        cached_cred_def = self._cred_def_cache.get(cred_def_topic_id)

        if cached_cred_def:
            cred_def = cached_cred_def
        else:
            cred_def_payload = await self._hcs_file_service.resolve_file(cred_def_topic_id)
            cred_def = AnonCredsCredDef.from_json(cred_def_payload.decode()) if cred_def_payload else None

            if not cred_def:
                return GetCredDefResult(
                    credential_definition_id=cred_def_id,
                    resolution_metadata={
                        "error": "notFound",
                        "message": f"AnonCreds credential definition with id '{cred_def_id}' not found",
                    },
                    credential_definition_metadata={},
                )

            self._cred_def_cache.set(cred_def_topic_id, cred_def)

        return GetCredDefResult(
            credential_definition=cred_def,
            credential_definition_id=cred_def_id,
            resolution_metadata={},
            credential_definition_metadata={},
        )
    except Exception as error:
        LOGGER.error(f"Error on retrieving AnonCreds credential definition: {error!s}")
        return GetCredDefResult(
            credential_definition_id=cred_def_id,
            resolution_metadata={
                "error": "otherError",
                "message": f"unable to resolve credential definition: ${error!s}",
            },
            credential_definition_metadata={},
        )

`get_rev_list(rev_reg_id, timestamp)` `async` ¶

Get a revocation list from the registry.

Parameters:

Name	Type	Description	Default
`rev_reg_id`	`str`	Revocation registry ID	required
`timestamp`	`int`	Timestamp to resolve revocation list for	required

Returns:

Name	Type	Description
`object`	`GetRevListResult`	Revocation list resolution result

Source code in hiero_did_sdk_python/anoncreds/hedera_anoncreds_registry.py

async def get_rev_list(self, rev_reg_id: str, timestamp: int) -> GetRevListResult:
    """Get a revocation list from the registry.

    Args:
        rev_reg_id: Revocation registry ID
        timestamp: Timestamp to resolve revocation list for

    Returns:
        object: Revocation list resolution result
    """
    try:
        rev_reg_def_result = await self.get_rev_reg_def(rev_reg_id)

        if not rev_reg_def_result.revocation_registry_definition:
            return GetRevListResult(
                revocation_registry_id=rev_reg_id,
                resolution_metadata={
                    "error": "notFound",
                    "message": f"AnonCreds revocation registry with id '{rev_reg_id}' not found",
                },
                revocation_list_metadata={},
            )

        rev_reg_def = rev_reg_def_result.revocation_registry_definition
        entries_topic_id = rev_reg_def_result.revocation_registry_definition_metadata.get("entries_topic_id")

        if not entries_topic_id:
            return GetRevListResult(
                revocation_registry_id=rev_reg_id,
                resolution_metadata={
                    "error": "notFound",
                    "message": "Entries topic ID is missing from revocation registry metadata",
                },
                revocation_list_metadata={},
            )

        cached_messages = self._rev_reg_entries_messages_cache.get(entries_topic_id)
        if cached_messages:
            last_cached_message_timestamp = cached_messages[-1].consensus_timestamp

            if last_cached_message_timestamp.seconds >= timestamp:
                borderline_timestamp = Timestamp(seconds=timestamp, nanos=0)
                entries_messages = filter(
                    lambda message: message.consensus_timestamp.seconds < timestamp
                    or message.consensus_timestamp == borderline_timestamp,
                    cached_messages,
                )
                entries = [cast(AnonCredsRevRegEntry, message.message) for message in entries_messages]

                # This means that requested timestamp is before the actual registration of rev list
                # In such case, we want to return initial state for the list (by adding first message to entries)
                if len(entries) == 0:
                    entries.append(cast(AnonCredsRevRegEntry, cached_messages[0].message))

                return GetRevListResult(
                    revocation_registry_id=rev_reg_id,
                    revocation_list=AnonCredsRevList.from_rev_reg_entries(
                        entries, rev_reg_id, rev_reg_def, timestamp
                    ),
                    resolution_metadata={},
                    revocation_list_metadata={},
                )
            else:
                new_messages = await HcsMessageResolver(
                    topic_id=entries_topic_id,
                    message_type=HcsRevRegEntryMessage,
                    timestamp_from=last_cached_message_timestamp,
                    timestamp_to=Timestamp(seconds=timestamp, nanos=0),
                    include_response_metadata=True,
                ).execute(self._client)

                # Note: 'chain' function is used instead of lists sum due to significantly better performance on large lists
                # See: https://docs.python.org/3/library/itertools.html, https://stackoverflow.com/a/41772165
                entries_messages = (
                    list(chain(cached_messages, cast(list[HcsMessageWithResponseMetadata], new_messages)))
                    if len(new_messages) > 0
                    else cached_messages
                )

                self._rev_reg_entries_messages_cache.set(entries_topic_id, entries_messages)

                entries = [cast(AnonCredsRevRegEntry, message.message) for message in entries_messages]

                return GetRevListResult(
                    revocation_registry_id=rev_reg_id,
                    revocation_list=AnonCredsRevList.from_rev_reg_entries(
                        entries, rev_reg_id, rev_reg_def, timestamp
                    ),
                    resolution_metadata={},
                    revocation_list_metadata={},
                )

        entries_messages = await HcsMessageResolver(
            topic_id=entries_topic_id,
            message_type=HcsRevRegEntryMessage,
            timestamp_to=Timestamp(seconds=timestamp, nanos=0),
            include_response_metadata=True,
        ).execute(self._client)

        if len(entries_messages) == 0:
            # If returned entries list is empty, we need to fetch the first message and check if list is registered
            # It's possible that requested timestamp is before the actual registration of rev list -> we want to return initial state for the list (by adding first message to entries)

            # The second request looks redundant here, but it should be the rare case that will be subsequently handled by cache
            entries_messages = await HcsMessageResolver(
                topic_id=entries_topic_id,
                message_type=HcsRevRegEntryMessage,
                limit=1,
                include_response_metadata=True,
            ).execute(self._client)

            if len(entries_messages) == 0:
                return GetRevListResult(
                    revocation_registry_id=rev_reg_id,
                    resolution_metadata={
                        "error": "notFound",
                        "message": f"Registered revocation list for registry id '{rev_reg_id}' is not found",
                    },
                    revocation_list_metadata={},
                )

        entries_messages = cast(list[HcsMessageWithResponseMetadata], entries_messages)
        self._rev_reg_entries_messages_cache.set(entries_topic_id, entries_messages)

        entries = [cast(AnonCredsRevRegEntry, message.message) for message in entries_messages]

        return GetRevListResult(
            revocation_registry_id=rev_reg_id,
            revocation_list=AnonCredsRevList.from_rev_reg_entries(entries, rev_reg_id, rev_reg_def, timestamp),
            resolution_metadata={},
            revocation_list_metadata={},
        )

    except Exception as error:
        LOGGER.error(f"Error on retrieving AnonCreds revocation list: {error!s}")
        return GetRevListResult(
            revocation_registry_id=rev_reg_id,
            resolution_metadata={
                "error": "otherError",
                "message": f"Unable to resolve revocation list: ${error!s}",
            },
            revocation_list_metadata={},
        )

`get_rev_reg_def(revocation_registry_definition_id)` `async` ¶

Get a revocation registry definition from the registry.

Parameters:

Name	Type	Description	Default
`revocation_registry_definition_id`	`str`	Revocation registry definition ID to resolve	required

Returns:

Name	Type	Description
`object`	`GetRevRegDefResult`	Revocation registry definition resolution result

Source code in hiero_did_sdk_python/anoncreds/hedera_anoncreds_registry.py

async def get_rev_reg_def(self, revocation_registry_definition_id: str) -> GetRevRegDefResult:
    """Get a revocation registry definition from the registry.

    Args:
        revocation_registry_definition_id: Revocation registry definition ID to resolve

    Returns:
        object: Revocation registry definition resolution result
    """
    try:
        parsed_identifier = parse_anoncreds_identifier(revocation_registry_definition_id)

        if parsed_identifier.object_type != AnonCredsObjectType.REV_REG:
            return GetRevRegDefResult(
                revocation_registry_definition_id=revocation_registry_definition_id,
                resolution_metadata={
                    "error": "notFound",
                    "message": f"Revocation registry id '{revocation_registry_definition_id}' is invalid",
                },
                revocation_registry_definition_metadata={},
            )

        rev_reg_def_topic_id = parsed_identifier.topic_id

        cached_rev_reg_def_with_metadata = self._rev_reg_def_cache.get(rev_reg_def_topic_id)

        if cached_rev_reg_def_with_metadata:
            rev_reg_def_with_metadata = cached_rev_reg_def_with_metadata
        else:
            rev_reg_def_payload = await self._hcs_file_service.resolve_file(rev_reg_def_topic_id)
            rev_reg_def_with_metadata = (
                RevRegDefWithHcsMetadata.from_json(rev_reg_def_payload.decode()) if rev_reg_def_payload else None
            )

            if not rev_reg_def_with_metadata:
                return GetRevRegDefResult(
                    revocation_registry_definition_id=revocation_registry_definition_id,
                    resolution_metadata={
                        "error": "notFound",
                        "message": f"AnonCreds revocation registry with id '{revocation_registry_definition_id}' not found",
                    },
                    revocation_registry_definition_metadata={},
                )

            self._rev_reg_def_cache.set(rev_reg_def_topic_id, rev_reg_def_with_metadata)

        return GetRevRegDefResult(
            revocation_registry_definition=rev_reg_def_with_metadata.rev_reg_def,
            revocation_registry_definition_id=revocation_registry_definition_id,
            resolution_metadata={},
            revocation_registry_definition_metadata={**rev_reg_def_with_metadata.hcs_metadata},
        )
    except Exception as error:
        LOGGER.error(f"Error on retrieving AnonCreds revocation registry definition: {error!s}")
        return GetRevRegDefResult(
            revocation_registry_definition_id=revocation_registry_definition_id,
            resolution_metadata={
                "error": "otherError",
                "message": f"unable to resolve revocation registry definition: ${error!s}",
            },
            revocation_registry_definition_metadata={},
        )

`get_schema(schema_id)` `async` ¶

Get a schema from the registry.

Parameters:

Name	Type	Description	Default
`schema_id`	`str`	Schema ID to resolver	required

Returns:

Name	Type	Description
`object`	`GetSchemaResult`	Schema resolution result

Source code in hiero_did_sdk_python/anoncreds/hedera_anoncreds_registry.py

async def get_schema(self, schema_id: str) -> GetSchemaResult:
    """Get a schema from the registry.

    Args:
        schema_id: Schema ID to resolver

    Returns:
        object: Schema resolution result
    """
    try:
        parsed_identifier = parse_anoncreds_identifier(schema_id)

        if parsed_identifier.object_type != AnonCredsObjectType.SCHEMA:
            return GetSchemaResult(
                schema_id=schema_id,
                resolution_metadata={
                    "error": "notFound",
                    "message": f"AnonCreds Schema id '{schema_id}' is invalid",
                },
                schema_metadata={},
            )

        schema_topic_id = parsed_identifier.topic_id

        cached_schema = self._schema_cache.get(schema_topic_id)

        if cached_schema:
            schema = cached_schema
        else:
            schema_payload = await self._hcs_file_service.resolve_file(parsed_identifier.topic_id)
            schema = AnonCredsSchema.from_json(schema_payload.decode()) if schema_payload else None

            if not schema:
                return GetSchemaResult(
                    schema_id=schema_id,
                    resolution_metadata={
                        "error": "notFound",
                        "message": f"AnonCreds schema with id '{schema_id}' not found",
                    },
                    schema_metadata={},
                )

            self._schema_cache.set(schema_topic_id, schema)

        return GetSchemaResult(schema=schema, schema_id=schema_id, resolution_metadata={}, schema_metadata={})
    except Exception as error:
        LOGGER.error(f"Error on retrieving AnonCreds Schema: {error!s}")
        return GetSchemaResult(
            schema_id=schema_id,
            resolution_metadata={
                "error": "otherError",
                "message": f"unable to resolve schema: ${error!s}",
            },
            schema_metadata={},
        )

`register_cred_def(cred_def, issuer_key_der)` `async` ¶

Register Credential Definition.

Parameters:

Name	Type	Description	Default
`cred_def`	`AnonCredsCredDef`	Credential definition object to register	required
`issuer_key_der`	`str`	Issuer private key encoded in DER	required

Returns:

Name	Type	Description
`object`	`RegisterCredDefResult`	Credential definition registration result

Source code in hiero_did_sdk_python/anoncreds/hedera_anoncreds_registry.py

async def register_cred_def(self, cred_def: AnonCredsCredDef, issuer_key_der: str) -> RegisterCredDefResult:
    """Register Credential Definition.

    Args:
        cred_def: Credential definition object to register
        issuer_key_der: Issuer private key encoded in DER

    Returns:
        object: Credential definition registration result
    """
    try:
        hcs_file_payload = cred_def.to_json().encode()
        cred_def_topic_id = await self._hcs_file_service.submit_file(hcs_file_payload, issuer_key_der)

        return RegisterCredDefResult(
            credential_definition_state=CredDefState(
                state="finished",
                credential_definition=cred_def,
                credential_definition_id=build_anoncreds_identifier(
                    cred_def.issuer_id, cred_def_topic_id, AnonCredsObjectType.PUBLIC_CRED_DEF
                ),
            ),
            registration_metadata={},
            credential_definition_metadata={},
        )
    except Exception as error:
        LOGGER.error(f"Error on registering Anoncreds Cred Def: {error!s}")
        return RegisterCredDefResult(
            credential_definition_state=CredDefState(
                state="failed", credential_definition=cred_def, reason=f"unknownError: ${error!s}"
            ),
            registration_metadata={},
            credential_definition_metadata={},
        )

`register_rev_list(rev_list, issuer_key_der)` `async` ¶

Register Revocation list.

Parameters:

Name	Type	Description	Default
`rev_list`	`AnonCredsRevList`	Revocation list object to register	required
`issuer_key_der`	`str`	Issuer private key encoded in DER	required

Returns: Revocation list registration result

Source code in hiero_did_sdk_python/anoncreds/hedera_anoncreds_registry.py

async def register_rev_list(self, rev_list: AnonCredsRevList, issuer_key_der: str) -> RegisterRevListResult:
    """Register Revocation list.

    Args:
        rev_list: Revocation list object to register
        issuer_key_der: Issuer private key encoded in DER

    Returns: Revocation list registration result
    """
    try:
        return await self._submit_rev_list_entry(rev_list, issuer_key_der)
    except Exception as error:
        LOGGER.error(f"Error on registering Anoncreds revocation list: {error!s}")
        return RegisterRevListResult(
            revocation_list_state=RevListState(
                state="failed", revocation_list=rev_list, reason=f"unknownError: ${error!s}"
            ),
            registration_metadata={},
            revocation_list_metadata={},
        )

`register_rev_reg_def(rev_reg_def, issuer_key_der)` `async` ¶

Register Revocation registry definition.

Parameters:

Name	Type	Description	Default
`rev_reg_def`	`AnonCredsRevRegDef`	Revocation registry definition object to register	required
`issuer_key_der`	`str`	Issuer private key encoded in DER	required

Returns:

Name	Type	Description
`object`	`RegisterRevRegDefResult`	Revocation registry definition registration result

Source code in hiero_did_sdk_python/anoncreds/hedera_anoncreds_registry.py

async def register_rev_reg_def(
    self, rev_reg_def: AnonCredsRevRegDef, issuer_key_der: str
) -> RegisterRevRegDefResult:
    """Register Revocation registry definition.

    Args:
        rev_reg_def: Revocation registry definition object to register
        issuer_key_der: Issuer private key encoded in DER

    Returns:
        object: Revocation registry definition registration result
    """
    try:
        issuer_key = PrivateKey.from_string(issuer_key_der)

        entries_topic_options = HcsTopicOptions(submit_key=issuer_key.public_key())
        entries_topic_id = await self._hcs_topic_service.create_topic(entries_topic_options, [issuer_key])

        rev_reg_def_with_metadata = RevRegDefWithHcsMetadata(
            rev_reg_def=rev_reg_def, hcs_metadata={"entries_topic_id": entries_topic_id}
        )

        hcs_file_payload = rev_reg_def_with_metadata.to_json().encode()
        rev_reg_def_topic_id = await self._hcs_file_service.submit_file(hcs_file_payload, issuer_key_der)

        # We want to cache registry definition right away
        # Helps to avoid potential cases where issuer pushes rev entries immediately but registry definition data (HCS-1 messages) is not propagated to mirror nodes yet
        self._rev_reg_def_cache.set(rev_reg_def_topic_id, rev_reg_def_with_metadata)

        return RegisterRevRegDefResult(
            revocation_registry_definition_state=RevRegDefState(
                state="finished",
                revocation_registry_definition=rev_reg_def,
                revocation_registry_definition_id=build_anoncreds_identifier(
                    rev_reg_def.issuer_id, rev_reg_def_topic_id, AnonCredsObjectType.REV_REG
                ),
            ),
            registration_metadata={},
            revocation_registry_definition_metadata={**rev_reg_def_with_metadata.hcs_metadata},
        )
    except Exception as error:
        LOGGER.error(f"Error on registering Anoncreds Revocation registry definition: {error!s}")
        return RegisterRevRegDefResult(
            revocation_registry_definition_state=RevRegDefState(
                state="failed", revocation_registry_definition=rev_reg_def, reason=f"unknownError: ${error!s}"
            ),
            registration_metadata={},
            revocation_registry_definition_metadata={},
        )

`register_schema(schema, issuer_key_der)` `async` ¶

Register Schema.

Parameters:

Name	Type	Description	Default
`schema`	`AnonCredsSchema`	Schema object to register	required
`issuer_key_der`	`str`	Issuer private key encoded in DER format	required

Returns:

Name	Type	Description
`object`	`RegisterSchemaResult`	Schema registration result

Source code in hiero_did_sdk_python/anoncreds/hedera_anoncreds_registry.py

async def register_schema(self, schema: AnonCredsSchema, issuer_key_der: str) -> RegisterSchemaResult:
    """Register Schema.

    Args:
        schema: Schema object to register
        issuer_key_der: Issuer private key encoded in DER format

    Returns:
        object: Schema registration result
    """
    try:
        hcs_file_payload = schema.to_json().encode()
        schema_topic_id = await self._hcs_file_service.submit_file(hcs_file_payload, issuer_key_der)

        return RegisterSchemaResult(
            schema_state=SchemaState(
                state="finished",
                schema=schema,
                schema_id=build_anoncreds_identifier(schema.issuer_id, schema_topic_id, AnonCredsObjectType.SCHEMA),
            ),
            schema_metadata={},
            registration_metadata={},
        )
    except Exception as error:
        LOGGER.error(f"Error on registering Anoncreds Schema: {error!s}")
        return RegisterSchemaResult(
            schema_state=SchemaState(state="failed", schema=schema, reason=f"unknownError: ${error!s}"),
            schema_metadata={},
            registration_metadata={},
        )

`update_rev_list(prev_list, curr_list, revoked, issuer_key_der)` `async` ¶

Update Revocation list.

Parameters:

Name	Type	Description	Default
`prev_list`	`AnonCredsRevList`	Previous Revocation list object	required
`curr_list`	`AnonCredsRevList`	Current Revocation list object	required
`revoked`	`Sequence[int]`	Revoked credential indexes	required
`issuer_key_der`	`str`	Issuer private key encoded in DER	required

Returns: Revocation list update result

Source code in hiero_did_sdk_python/anoncreds/hedera_anoncreds_registry.py

async def update_rev_list(
    self, prev_list: AnonCredsRevList, curr_list: AnonCredsRevList, revoked: Sequence[int], issuer_key_der: str
) -> RegisterRevListResult:
    """Update Revocation list.

    Args:
        prev_list: Previous Revocation list object
        curr_list: Current Revocation list object
        revoked: Revoked credential indexes
        issuer_key_der: Issuer private key encoded in DER

    Returns: Revocation list update result
    """
    try:
        return await self._submit_rev_list_entry(curr_list, issuer_key_der, prev_list, revoked)
    except Exception as error:
        LOGGER.error(f"Error on updating Anoncreds revocation list: {error!s}")
        return RegisterRevListResult(
            revocation_list_state=RevListState(
                state="failed", revocation_list=curr_list, reason=f"unknownError: ${error!s}"
            ),
            registration_metadata={},
            revocation_list_metadata={},
        )

Models and types¶

`AnonCredsCredDef` `dataclass` ¶

Bases: Serializable

Model representing AnonCreds credential definition.

Attributes:

Name	Type	Description
`issuer_id`	`str`	Credential definition issuer DID
`schema_id`	`str`	Schema ID
`type_`	`str`	Credential definition type. Only "CL" type is currently supported
`tag`	`str`	Credential definition tag
`value`	`CredDefValue`	Definition value object

Source code in hiero_did_sdk_python/anoncreds/models/credential_definition.py

@dataclass
class AnonCredsCredDef(Serializable):
    """Model representing AnonCreds credential definition.

    Attributes:
        issuer_id: Credential definition issuer DID
        schema_id: Schema ID
        type_: Credential definition type. Only "CL" type is currently supported
        tag: Credential definition tag
        value: Definition value object
    """

    issuer_id: str
    schema_id: str
    type_: ClassVar[str] = CREDENTIAL_DEFINITION_TYPE
    tag: str
    value: CredDefValue

    @classmethod
    def from_json_payload(cls, payload: dict):
        match payload:
            case {"issuerId": issuer_id, "schemaId": schema_id, "type": type_, "tag": tag, "value": value}:
                if type_ != CREDENTIAL_DEFINITION_TYPE:
                    raise Exception(f"Unsupported Anoncreds Cred Def type: {type_}")
                return cls(
                    issuer_id=issuer_id, schema_id=schema_id, tag=tag, value=CredDefValue.from_json_payload(value)
                )
            case _:
                raise Exception(f"{cls.__name__} JSON parsing failed: Invalid JSON structure")

    def get_json_payload(self):
        return {
            "issuerId": self.issuer_id,
            "schemaId": self.schema_id,
            "type": self.type_,
            "tag": self.tag,
            "value": self.value.get_json_payload(),
        }

`AnonCredsRevList` `dataclass` ¶

Bases: Serializable

Model representing AnonCreds revocation list object.

Attributes:

Name	Type	Description
`issuer_id`	`str`	Revocation registry issuer DID
`rev_reg_def_id`	`str`	Revocation registry definition ID
`revocation_list`	`list[int]`	Revocation list that represents state (revoked/non-revoked) of each credential in registry, list size correspond to number of credentials that revocation registry can hold
`current_accumulator`	`str`	Current value of CL accumulator
`timestamp`	`int \| None`	Timestamp associated with revocation list instance

Source code in hiero_did_sdk_python/anoncreds/models/revocation/revocation_list.py

@dataclass
class AnonCredsRevList(Serializable):
    """Model representing AnonCreds revocation list object.

    Attributes:
        issuer_id: Revocation registry issuer DID
        rev_reg_def_id: Revocation registry definition ID
        revocation_list: Revocation list that represents state (revoked/non-revoked) of each credential in registry, list size correspond to number of credentials that revocation registry can hold
        current_accumulator: Current value of CL accumulator
        timestamp: Timestamp associated with revocation list instance
    """

    issuer_id: str
    rev_reg_def_id: str
    revocation_list: list[int]
    current_accumulator: str
    timestamp: int | None = None

    @classmethod
    def from_rev_reg_entries(
        cls,
        entries: list[AnonCredsRevRegEntry],
        rev_reg_id: str,
        rev_reg_def: AnonCredsRevRegDef,
        timestamp: int | None = None,
    ):
        """Build revocation list object from corresponding revocation registry entries.

        Args:
            entries: List of revocation registry entries to build state from
            rev_reg_id: Revocation registry ID
            rev_reg_def: Revocation registry definition object
            timestamp: Requested timestamp to associate revocation list with
        """
        revoked_indexes = []

        for entry in entries:
            if entry.value.revoked:
                revoked_indexes += entry.value.revoked

        rev_list_bit_array = _indexes_to_bit_array(revoked_indexes, rev_reg_def.value.max_cred_num)
        accum = entries[-1].value.accum

        return cls(
            issuer_id=rev_reg_def.issuer_id,
            rev_reg_def_id=rev_reg_id,
            revocation_list=rev_list_bit_array,
            current_accumulator=accum,
            timestamp=timestamp,
        )

    @classmethod
    def from_json_payload(cls, payload: dict):
        match payload:
            case {
                "issuerId": issuer_id,
                "revRegDefId": rev_reg_def_id,
                "revocationList": revocation_list,
                "currentAccumulator": current_accumulator,
                **rest,
            }:
                return cls(
                    issuer_id=issuer_id,
                    rev_reg_def_id=rev_reg_def_id,
                    revocation_list=revocation_list,
                    current_accumulator=current_accumulator,
                    timestamp=rest.get("timestamp"),
                )
            case _:
                raise Exception(f"{cls.__name__} JSON parsing failed: Invalid JSON structure")

    def get_json_payload(self):
        payload = {
            "issuerId": self.issuer_id,
            "revRegDefId": self.rev_reg_def_id,
            "revocationList": self.revocation_list,
            "currentAccumulator": self.current_accumulator,
        }

        if self.timestamp:
            payload["timestamp"] = self.timestamp

        return payload

`from_rev_reg_entries(entries, rev_reg_id, rev_reg_def, timestamp=None)` `classmethod` ¶

Build revocation list object from corresponding revocation registry entries.

Parameters:

Name	Type	Description	Default
`entries`	`list[AnonCredsRevRegEntry]`	List of revocation registry entries to build state from	required
`rev_reg_id`	`str`	Revocation registry ID	required
`rev_reg_def`	`AnonCredsRevRegDef`	Revocation registry definition object	required
`timestamp`	`int \| None`	Requested timestamp to associate revocation list with	`None`

Source code in hiero_did_sdk_python/anoncreds/models/revocation/revocation_list.py

@classmethod
def from_rev_reg_entries(
    cls,
    entries: list[AnonCredsRevRegEntry],
    rev_reg_id: str,
    rev_reg_def: AnonCredsRevRegDef,
    timestamp: int | None = None,
):
    """Build revocation list object from corresponding revocation registry entries.

    Args:
        entries: List of revocation registry entries to build state from
        rev_reg_id: Revocation registry ID
        rev_reg_def: Revocation registry definition object
        timestamp: Requested timestamp to associate revocation list with
    """
    revoked_indexes = []

    for entry in entries:
        if entry.value.revoked:
            revoked_indexes += entry.value.revoked

    rev_list_bit_array = _indexes_to_bit_array(revoked_indexes, rev_reg_def.value.max_cred_num)
    accum = entries[-1].value.accum

    return cls(
        issuer_id=rev_reg_def.issuer_id,
        rev_reg_def_id=rev_reg_id,
        revocation_list=rev_list_bit_array,
        current_accumulator=accum,
        timestamp=timestamp,
    )

`AnonCredsRevRegDef` `dataclass` ¶

Bases: Serializable

Model representing AnonCreds revocation registry definition.

Attributes:

Name	Type	Description
`issuer_id`	`str`	Revocation registry issuer DID
`type_`	`str`	Revocation registry type. Only "CL_ACCUM" type is currently supported
`cred_def_id`	`str`	Credential definition ID
`tag`	`str`	Revocation registry tag
`value`	`RevRegDefValue`	Definition value object

Source code in hiero_did_sdk_python/anoncreds/models/revocation/revocation_registry_definition.py

@dataclass
class AnonCredsRevRegDef(Serializable):
    """Model representing AnonCreds revocation registry definition.

    Attributes:
        issuer_id: Revocation registry issuer DID
        type_: Revocation registry type. Only "CL_ACCUM" type is currently supported
        cred_def_id: Credential definition ID
        tag: Revocation registry tag
        value: Definition value object
    """

    issuer_id: str
    type_: ClassVar[str] = REVOCATION_REGISTRY_TYPE
    cred_def_id: str
    tag: str
    value: RevRegDefValue

    @classmethod
    def from_json_payload(cls, payload: dict):
        match payload:
            case {"issuerId": issuer_id, "type": type_, "credDefId": cred_def_id, "tag": tag, "value": value}:
                if type_ != REVOCATION_REGISTRY_TYPE:
                    raise Exception(f"Unsupported Anoncreds Revocation Registry type: {type_}")
                return cls(
                    issuer_id=issuer_id, cred_def_id=cred_def_id, tag=tag, value=RevRegDefValue.from_json_payload(value)
                )
            case _:
                raise Exception(f"{cls.__name__} JSON parsing failed: Invalid JSON structure")

    def get_json_payload(self):
        return {
            "issuerId": self.issuer_id,
            "type": self.type_,
            "credDefId": self.cred_def_id,
            "tag": self.tag,
            "value": self.value.get_json_payload(),
        }

`AnonCredsRevRegEntry` `dataclass` ¶

Bases: Serializable

Model representing AnonCreds revocation registry entry.

Attributes:

Name	Type	Description
`value`	`RevRegEntryValue`	Entry value object
`ver`	`str`	Entry version

Source code in hiero_did_sdk_python/anoncreds/models/revocation/revocation_registry_entry.py

@dataclass
class AnonCredsRevRegEntry(Serializable):
    """Model representing AnonCreds revocation registry entry.

    Attributes:
        value: Entry value object
        ver: Entry version
    """

    value: RevRegEntryValue
    ver: str = DEFAULT_REV_REG_ENTRY_VERSION

    @classmethod
    def from_json_payload(cls, payload: dict):
        match payload:
            case {"payload": compressed_str}:
                compressed_bytes = b64_to_bytes(compressed_str)
                entry_params = json.loads(ZstdDecompressor().decompress(compressed_bytes))
                return cls._from_json_payload_raw(entry_params)
            case _:
                raise Exception(f"{cls.__name__} JSON parsing failed: Invalid JSON structure")

    def get_json_payload(self):
        payload_str = json.dumps(self._get_json_payload_raw()).encode()
        compressed_payload = ZstdCompressor().compress(payload_str)
        return {"payload": bytes_to_b64(compressed_payload)}

    @classmethod
    def _from_json_payload_raw(cls, payload: dict):
        match payload:
            case {"ver": ver, "value": value}:
                return cls(ver=ver, value=RevRegEntryValue.from_json_payload(value))
            case _:
                raise Exception(f"{cls.__name__} JSON parsing failed: Invalid JSON structure")

    def _get_json_payload_raw(self):
        return {"ver": self.ver, "value": self.value.get_json_payload()}

`AnonCredsSchema` `dataclass` ¶

Bases: Serializable

Model representing AnonCreds schema.

Attributes:

Name	Type	Description
`name`	`str`	Schema name
`issuer_id`	`str`	Schema issuer DID
`attr_names`	`list[str]`	List of schema attribute names
`version`	`str`	Schema version

Source code in hiero_did_sdk_python/anoncreds/models/schema.py

@dataclass
class AnonCredsSchema(Serializable):
    """Model representing AnonCreds schema.

    Attributes:
        name: Schema name
        issuer_id: Schema issuer DID
        attr_names: List of schema attribute names
        version: Schema version
    """

    name: str
    issuer_id: str
    attr_names: list[str]
    version: str

    @classmethod
    def from_json_payload(cls, payload: dict):
        match payload:
            case {"name": name, "issuerId": issuer_id, "attrNames": attr_names, "version": version}:
                return cls(name=name, issuer_id=issuer_id, attr_names=attr_names, version=version)
            case _:
                raise Exception(f"{cls.__name__} JSON parsing failed: Invalid JSON structure")

    def get_json_payload(self):
        return {"name": self.name, "issuerId": self.issuer_id, "attrNames": self.attr_names, "version": self.version}

`CredDefValue` `dataclass` ¶

Bases: Serializable

Model representing AnonCreds credential definition value.

Attributes:

Name	Type	Description
`primary`	`CredDefValuePrimary`	Credential definition primary value
`revocation`	`CredDefValueRevocation \| None`	Credential definition revocation-specific value

Source code in hiero_did_sdk_python/anoncreds/models/credential_definition.py

@dataclass
class CredDefValue(Serializable):
    """Model representing AnonCreds credential definition value.

    Attributes:
        primary: Credential definition primary value
        revocation: Credential definition revocation-specific value
    """

    primary: CredDefValuePrimary
    revocation: CredDefValueRevocation | None

    @classmethod
    def from_json_payload(cls, payload: dict):
        match payload:
            case {"primary": primary, "revocation": revocation}:
                return cls(
                    CredDefValuePrimary.from_json_payload(primary), CredDefValueRevocation.from_json_payload(revocation)
                )
            case {"primary": primary}:
                return cls(CredDefValuePrimary.from_json_payload(primary), None)
            case _:
                raise Exception(f"{cls.__name__} JSON parsing failed: Invalid JSON structure")

    def get_json_payload(self):
        payload = {"primary": self.primary.get_json_payload()}

        if self.revocation:
            payload["revocation"] = self.revocation.get_json_payload()

        return payload

`CredDefValuePrimary` `dataclass` ¶

Bases: Serializable

Model of primary credential definition value.

Parameters:

Name	Type	Description	Default
`n`	`str`	Safe RSA-2048 number	required
`s`	`str`	Randomly selected quadratic residue of n	required
`r`	`dict`	Object that defines a CL-RSA public key fragment for each attribute in the credential	required
`rctxt`	`str`	Equal to s^(xrctxt), where xrctxt is a randomly selected integer between 2 and p'q'-1	required
`z`	`str`	is equal to s^(xz), where xz is a randomly selected integer between 2 and p'q'-1. This makes up part of the CL-RSA public key, independent of the message blocks being signed	required

Source code in hiero_did_sdk_python/anoncreds/models/credential_definition.py

@dataclass
class CredDefValuePrimary(Serializable):
    """Model of primary credential definition value.

    Args:
        n: Safe RSA-2048 number
        s: Randomly selected quadratic residue of n
        r: Object that defines a CL-RSA public key fragment for each attribute in the credential
        rctxt: Equal to s^(xrctxt), where xrctxt is a randomly selected integer between 2 and p'q'-1
        z: is equal to s^(xz), where xz is a randomly selected integer between 2
            and p'q'-1. This makes up part of the CL-RSA public key, independent of
            the message blocks being signed
    """

    n: str
    s: str
    r: dict
    rctxt: str
    z: str

    @classmethod
    def from_json_payload(cls, payload: dict):
        match payload:
            case {"n": n, "s": s, "r": r, "rctxt": rctxt, "z": z}:
                return cls(n, s, r, rctxt, z)
            case _:
                raise Exception(f"{cls.__name__} JSON parsing failed: Invalid JSON structure")

    def get_json_payload(self):
        return self.__dict__

`CredDefValueRevocation` `dataclass` ¶

Bases: Serializable

Model of revocation-specific credential definition value.

Parameters:

Name	Type	Description	Default
`g`	`str`	Generator for the elliptic curve group G1	required
`g_dash`	`str`	Generator for the elliptic curve group G2	required
`h`	`str`	Elliptic curve point selected uniformly at random from G1	required
`h0`	`str`	Elliptic curve point selected uniformly at random from G1	required
`h1`	`str`	Elliptic curve point selected uniformly at random from G1	required
`h2`	`str`	Elliptic curve point selected uniformly at random from G1	required
`htilde`	`str`	Elliptic curve point selected uniformly at random from G1	required
`h_cap`	`str`	Elliptic curve point selected uniformly at random from G2	required
`u`	`str`	Elliptic curve point selected uniformly at random from G2	required
`pk`	`str`	Public key in G1 for the issuer with respect to this accumulator, computed as g^sk (in multiplicative notation), where sk is from r_key above	required
`y`	`str`	Elliptic curve point in G2. computed as h_cap^x (in multiplicative notation), where x is from r_key above	required

Source code in hiero_did_sdk_python/anoncreds/models/credential_definition.py

@dataclass
class CredDefValueRevocation(Serializable):
    """Model of revocation-specific credential definition value.

    Args:
        g: Generator for the elliptic curve group G1
        g_dash: Generator for the elliptic curve group G2
        h: Elliptic curve point selected uniformly at random from G1
        h0: Elliptic curve point selected uniformly at random from G1
        h1: Elliptic curve point selected uniformly at random from G1
        h2: Elliptic curve point selected uniformly at random from G1
        htilde: Elliptic curve point selected uniformly at random from G1
        h_cap: Elliptic curve point selected uniformly at random from G2
        u: Elliptic curve point selected uniformly at random from G2
        pk: Public key in G1 for the issuer with respect to this accumulator, computed as g^sk (in multiplicative notation), where sk is from r_key above
        y: Elliptic curve point in G2. computed as h_cap^x (in multiplicative notation), where x is from r_key above
    """

    g: str
    g_dash: str
    h: str
    h0: str
    h1: str
    h2: str
    htilde: str
    h_cap: str
    u: str
    pk: str
    y: str

    @classmethod
    def from_json_payload(cls, payload: dict):
        match payload:
            case {
                "g": g,
                "g_dash": g_dash,
                "h": h,
                "h0": h0,
                "h1": h1,
                "h2": h2,
                "htilde": htilde,
                "h_cap": h_cap,
                "u": u,
                "pk": pk,
                "y": y,
            }:
                return cls(g, g_dash, h, h0, h1, h2, htilde, h_cap, u, pk, y)
            case _:
                raise Exception(f"{cls.__name__} JSON parsing failed: Invalid JSON structure")

    def get_json_payload(self):
        return self.__dict__

`HcsRevRegEntryMessage` `dataclass` ¶

Bases: HcsMessage, AnonCredsRevRegEntry

HCS message class for submitting revocation registry entries.

Source code in hiero_did_sdk_python/anoncreds/models/revocation/revocation_registry_entry.py

class HcsRevRegEntryMessage(HcsMessage, AnonCredsRevRegEntry):
    """HCS message class for submitting revocation registry entries."""

    def is_valid(self, topic_id: str | None = None) -> bool:
        return bool(self.value) and bool(self.value.accum)

`RevRegDefValue` `dataclass` ¶

Bases: Serializable

Model representing revocation registry definition value.

Attributes:

Name	Type	Description
`public_keys`	`dict`	Revocation registry public keys
`max_cred_num`	`int`	Max number of credentials which revocation state can be tracked in the registry
`tails_location`	`str`	Registry tails file location
`tails_hash`	`str`	Registry tails file hash

Source code in hiero_did_sdk_python/anoncreds/models/revocation/revocation_registry_definition.py

@dataclass
class RevRegDefValue(Serializable):
    """Model representing revocation registry definition value.

    Attributes:
        public_keys: Revocation registry public keys
        max_cred_num: Max number of credentials which revocation state can be tracked in the registry
        tails_location: Registry tails file location
        tails_hash: Registry tails file hash
    """

    public_keys: dict
    max_cred_num: int
    tails_location: str
    tails_hash: str

    @classmethod
    def from_json_payload(cls, payload: dict):
        match payload:
            case {
                "publicKeys": public_keys,
                "maxCredNum": max_cred_num,
                "tailsLocation": tails_location,
                "tailsHash": tails_hash,
            }:
                return cls(public_keys, max_cred_num, tails_location, tails_hash)
            case _:
                raise Exception(f"{cls.__name__} JSON parsing failed: Invalid JSON structure")

    def get_json_payload(self):
        return {
            "publicKeys": self.public_keys,
            "maxCredNum": self.max_cred_num,
            "tailsLocation": self.tails_location,
            "tailsHash": self.tails_hash,
        }

`RevRegDefWithHcsMetadata` `dataclass` ¶

Bases: Serializable

HCS specific model used to publish registry definition model along with HCS metadata.

Source code in hiero_did_sdk_python/anoncreds/models/revocation/revocation_registry_definition.py

@dataclass(frozen=True)
class RevRegDefWithHcsMetadata(Serializable):
    """HCS specific model used to publish registry definition model along with HCS metadata."""

    rev_reg_def: AnonCredsRevRegDef
    hcs_metadata: RevRegDefHcsMetadata

    @classmethod
    def from_json_payload(cls, payload: dict):
        match payload:
            case {"revRegDef": rev_reg_def, "hcsMetadata": hcs_metadata}:
                return cls(rev_reg_def=AnonCredsRevRegDef.from_json_payload(rev_reg_def), hcs_metadata=hcs_metadata)
            case _:
                raise Exception(f"{cls.__name__} JSON parsing failed: Invalid JSON structure")

    def get_json_payload(self):
        return {
            "revRegDef": self.rev_reg_def.get_json_payload(),
            "hcsMetadata": self.hcs_metadata,
        }

`RevRegEntryValue` `dataclass` ¶

Bases: Serializable

Model representing revocation registry entry value.

Attributes:

Name	Type	Description
`accum`	`str`	Current CL accumulator value
`prev_accum`	`str \| None`	Previous CL accumulator value
`issued`	`list[int] \| None`	List of issued credential indexes
`revoked`	`list[int] \| None`	List of revoked credential indexes

Source code in hiero_did_sdk_python/anoncreds/models/revocation/revocation_registry_entry.py

@dataclass
class RevRegEntryValue(Serializable):
    """Model representing revocation registry entry value.

    Attributes:
        accum: Current CL accumulator value
        prev_accum: Previous CL accumulator value
        issued: List of issued credential indexes
        revoked: List of revoked credential indexes
    """

    accum: str
    prev_accum: str | None = None
    issued: list[int] | None = None
    revoked: list[int] | None = None

    @classmethod
    def from_json_payload(cls, payload: dict):
        match payload:
            case {"accum": accum, **rest}:
                return cls(
                    accum=accum,
                    prev_accum=rest.get("prevAccum"),
                    issued=rest.get("issued"),
                    revoked=rest.get("revoked"),
                )
            case _:
                raise Exception(f"{cls.__name__} JSON parsing failed: Invalid JSON structure")

    def get_json_payload(self):
        payload: dict = {"accum": self.accum}

        if self.prev_accum:
            payload["prevAccum"] = self.prev_accum

        if self.issued:
            payload["issued"] = self.issued

        if self.revoked:
            payload["revoked"] = self.revoked

        return payload