aries_cloudagent.anoncreds package

async aries_cloudagent.anoncreds.setup(context: InjectionContext)[source]

Set up default resolvers.

Subpackages

Submodules

aries_cloudagent.anoncreds.base module

Base Registry.

exception aries_cloudagent.anoncreds.base.AnonCredsObjectAlreadyExists(message: str, obj_id: str, obj: Optional[T] = None, *args, **kwargs)[source]

Bases: AnonCredsRegistrationError, Generic[T]

Raised when an AnonCreds object already exists.

property message

Message.

exception aries_cloudagent.anoncreds.base.AnonCredsObjectNotFound(message: Optional[str] = None, resolution_metadata: Optional[dict] = None)[source]

Bases: BaseAnonCredsError

Raised when object is not found in resolver.

exception aries_cloudagent.anoncreds.base.AnonCredsRegistrationError(*args, error_code: Optional[str] = None, **kwargs)[source]

Bases: BaseAnonCredsError

Raised when registering an AnonCreds object fails.

exception aries_cloudagent.anoncreds.base.AnonCredsResolutionError(*args, error_code: Optional[str] = None, **kwargs)[source]

Bases: BaseAnonCredsError

Raised when resolving an AnonCreds object fails.

exception aries_cloudagent.anoncreds.base.AnonCredsSchemaAlreadyExists(message: str, obj_id: str, obj: Optional[T] = None, *args, **kwargs)[source]

Bases: AnonCredsObjectAlreadyExists[AnonCredsSchema]

Raised when a schema already exists.

property schema

Get Schema.

property schema_id

Get Schema Id.

exception aries_cloudagent.anoncreds.base.BaseAnonCredsError(*args, error_code: Optional[str] = None, **kwargs)[source]

Bases: BaseError

Base error class for AnonCreds.

class aries_cloudagent.anoncreds.base.BaseAnonCredsHandler[source]

Bases: ABC

Base Anon Creds Handler.

abstract async setup(context: InjectionContext)[source]

Class Setup method.

abstract property supported_identifiers_regex: Pattern

Regex to match supported identifiers.

async supports(identifier: str) → bool[source]

Determine whether this registry supports the given identifier.

class aries_cloudagent.anoncreds.base.BaseAnonCredsRegistrar[source]

Bases: BaseAnonCredsHandler

Base Anon Creds Registrar.

abstract async register_credential_definition(profile: Profile, schema: GetSchemaResult, credential_definition: CredDef, options: Optional[dict] = None) → CredDefResult[source]

Register a credential definition on the registry.

abstract async register_revocation_list(profile: Profile, rev_reg_def: RevRegDef, rev_list: RevList, options: Optional[dict] = None) → RevListResult[source]

Register a revocation list on the registry.

abstract async register_revocation_registry_definition(profile: Profile, revocation_registry_definition: RevRegDef, options: Optional[dict] = None) → RevRegDefResult[source]

Register a revocation registry definition on the registry.

abstract async register_schema(profile: Profile, schema: AnonCredsSchema, options: Optional[dict] = None) → SchemaResult[source]

Register a schema on the registry.

abstract async update_revocation_list(profile: Profile, rev_reg_def: RevRegDef, prev_list: RevList, curr_list: RevList, revoked: Sequence[int], options: Optional[dict] = None) → RevListResult[source]

Update a revocation list on the registry.

class aries_cloudagent.anoncreds.base.BaseAnonCredsResolver[source]

Bases: BaseAnonCredsHandler

Base Anon Creds Resolver.

abstract async get_credential_definition(profile: Profile, credential_definition_id: str) → GetCredDefResult[source]

Get a credential definition from the registry.

abstract async get_revocation_list(profile: Profile, revocation_registry_id: str, timestamp: int) → GetRevListResult[source]

Get a revocation list from the registry.

abstract async get_revocation_registry_definition(profile: Profile, revocation_registry_id: str) → GetRevRegDefResult[source]

Get a revocation registry definition from the registry.

abstract async get_schema(profile: Profile, schema_id: str) → GetSchemaResult[source]

Get a schema from the registry.

aries_cloudagent.anoncreds.error_messages module

Error messages for anoncreds.

aries_cloudagent.anoncreds.events module

Events fired by AnonCreds interface.

class aries_cloudagent.anoncreds.events.CredDefFinishedEvent(payload: CredDefFinishedPayload)[source]

Bases: Event

Event for cred def finished.

property payload: CredDefFinishedPayload

Return payload.

classmethod with_payload(schema_id: str, cred_def_id: str, issuer_id: str, support_revocation: bool, max_cred_num: int, options: Optional[dict] = None)[source]

With payload.

class aries_cloudagent.anoncreds.events.CredDefFinishedPayload(schema_id: str, cred_def_id: str, issuer_id: str, support_revocation: bool, max_cred_num: int, options: dict)[source]

Bases: NamedTuple

Payload of cred def finished event.

cred_def_id: str

Alias for field number 1

issuer_id: str

Alias for field number 2

max_cred_num: int

Alias for field number 4

options: dict

Alias for field number 5

schema_id: str

Alias for field number 0

support_revocation: bool

Alias for field number 3

class aries_cloudagent.anoncreds.events.RevListFinishedEvent(payload: RevListFinishedPayload)[source]

Bases: Event

Event for rev list finished.

property payload: RevListFinishedPayload

Return payload.

classmethod with_payload(rev_reg_id: str, revoked: list, options: Optional[dict] = None)[source]

With payload.

class aries_cloudagent.anoncreds.events.RevListFinishedPayload(rev_reg_id: str, revoked: list, options: dict)[source]

Bases: NamedTuple

Payload of rev list finished event.

options: dict

Alias for field number 2

rev_reg_id: str

Alias for field number 0

revoked: list

Alias for field number 1

class aries_cloudagent.anoncreds.events.RevRegDefFinishedEvent(payload: RevRegDefFinishedPayload)[source]

Bases: Event

Event for rev reg def finished.

property payload: RevRegDefFinishedPayload

Return payload.

classmethod with_payload(rev_reg_def_id: str, rev_reg_def: RevRegDef, options: Optional[dict] = None)[source]

With payload.

class aries_cloudagent.anoncreds.events.RevRegDefFinishedPayload(rev_reg_def_id: str, rev_reg_def: RevRegDef, options: dict)[source]

Bases: NamedTuple

Payload of rev reg def finished event.

options: dict

Alias for field number 2

rev_reg_def: RevRegDef

Alias for field number 1

rev_reg_def_id: str

Alias for field number 0

aries_cloudagent.anoncreds.holder module

Indy holder implementation.

class aries_cloudagent.anoncreds.holder.AnonCredsHolder(profile: Profile)[source]

Bases: object

AnonCreds holder class.

MASTER_SECRET_ID = 'default'
RECORD_TYPE_MIME_TYPES = 'attribute-mime-types'
async create_credential_request(credential_offer: dict, credential_definition: CredDef, holder_did: str) → Tuple[str, str][source]

Create a credential request for the given credential offer.

Parameters

  • credential_offer – The credential offer to create request for

  • credential_definition – The credential definition to create an offer for

  • holder_did – the DID of the agent making the request (may not be a real DID)

Returns

A tuple of the credential request and credential request metadata

async create_presentation(presentation_request: dict, requested_credentials: dict, schemas: Dict[str, AnonCredsSchema], credential_definitions: Dict[str, CredDef], rev_states: Optional[dict] = None) → str[source]

Get credentials stored in the wallet.

Parameters

  • presentation_request – Valid indy format presentation request

  • requested_credentials – Indy format requested credentials

  • schemas – Indy formatted schemas JSON

  • credential_definitions – Indy formatted credential definitions JSON

  • rev_states – Indy format revocation states JSON

async create_revocation_state(cred_rev_id: str, rev_reg_def: dict, rev_list: dict, tails_file_path: str) → str[source]

Create current revocation state for a received credential.

Parameters

  • cred_rev_id – credential revocation id in revocation registry

  • rev_reg_def – revocation registry definition

  • rev_reg_delta – revocation delta

  • timestamp – delta timestamp

Returns

the revocation state

async credential_revoked(ledger: BaseLedger, credential_id: str, fro: Optional[int] = None, to: Optional[int] = None) → bool[source]

Check ledger for revocation status of credential by cred id.

Parameters

credential_id – Credential id to check

async delete_credential(credential_id: str)[source]

Remove a credential stored in the wallet.

Parameters

credential_id – Credential id to remove

async get_credential(credential_id: str) → str[source]

Get a credential stored in the wallet.

Parameters

credential_id – Credential id to retrieve

async get_credentials(start: int, count: int, wql: dict)[source]

Get credentials stored in the wallet.

Parameters

  • start – Starting index

  • count – Number of records to return

  • wql – wql query dict

async get_credentials_for_presentation_request_by_referent(presentation_request: dict, referents: Sequence[str], start: int, count: int, extra_query: Optional[dict] = None)[source]

Get credentials stored in the wallet.

Parameters

  • presentation_request – Valid presentation request from issuer

  • referents – Presentation request referents to use to search for creds

  • start – Starting index

  • count – Maximum number of records to return

  • extra_query – wql query dict

async get_master_secret() → str[source]

Get or create the default master secret.

async get_mime_type(credential_id: str, attr: Optional[str] = None) → Union[dict, str][source]

Get MIME type per attribute (or for all attributes).

Parameters

  • credential_id – credential id

  • attr – attribute of interest or omit for all

Returns: Attribute MIME type or dict mapping attribute names to MIME types

attr_meta_json = all_meta.tags.get(attr)

property profile: AskarAnoncredsProfile

Accessor for the profile instance.

async store_credential(credential_definition: dict, credential_data: dict, credential_request_metadata: dict, credential_attr_mime_types: Optional[dict] = None, credential_id: Optional[str] = None, rev_reg_def: Optional[dict] = None) → str[source]

Store a credential in the wallet.

Parameters

  • credential_definition – Credential definition for this credential

  • credential_data – Credential data generated by the issuer

  • credential_request_metadata – credential request metadata generated by the issuer

  • credential_attr_mime_types – dict mapping attribute names to (optional) MIME types to store as non-secret record, if specified

  • credential_id – optionally override the stored credential id

  • rev_reg_def – revocation registry definition in json

Returns

the ID of the stored credential

exception aries_cloudagent.anoncreds.holder.AnonCredsHolderError(*args, error_code: Optional[str] = None, **kwargs)[source]

Bases: BaseError

Base class for holder exceptions.

aries_cloudagent.anoncreds.issuer module

anoncreds-rs issuer implementation.

class aries_cloudagent.anoncreds.issuer.AnonCredsIssuer(profile: Profile)[source]

Bases: object

AnonCreds issuer class.

This class provides methods for creating and registering AnonCreds objects needed to issue credentials. It also provides methods for storing and retrieving local representations of these objects from the wallet.

A general pattern is followed when creating and registering objects:

  1. Create the object locally

  2. Register the object with the anoncreds registry

  3. Store the object in the wallet

The wallet storage is used to keep track of the state of the object.

If the object is fully registered immediately after sending to the registry (state of finished), the object is saved to the wallet with an id matching the id returned from the registry.

If the object is not fully registered but pending (state of wait), the object is saved to the wallet with an id matching the job id returned from the registry.

If the object fails to register (state of failed), the object is saved to the wallet with an id matching the job id returned from the registry.

When an object finishes registration after being in a pending state (moving from state wait to state finished), the wallet entry matching the job id is removed and an entry matching the registered id is added.

async create_and_register_credential_definition(issuer_id: str, schema_id: str, tag: Optional[str] = None, signature_type: Optional[str] = None, options: Optional[dict] = None) → CredDefResult[source]

Create a new credential definition and store it in the wallet.

Parameters

  • issuer_id – the ID of the issuer creating the credential definition

  • schema_id – the schema ID for the credential definition

  • tag – the tag to use for the credential definition

  • signature_type – the signature type to use for the credential definition

  • options – any additional options to use when creating the credential definition

Returns

the result of the credential definition creation

Return type

CredDefResult

async create_and_register_schema(issuer_id: str, name: str, version: str, attr_names: Sequence[str], options: Optional[dict] = None) → SchemaResult[source]

Create a new credential schema and store it in the wallet.

Parameters

  • issuer_id – the DID issuing the credential definition

  • name – the schema name

  • version – the schema version

  • attr_names – a sequence of schema attribute names

Returns

A SchemaResult instance

async create_credential(credential_offer: dict, credential_request: dict, credential_values: dict) → str[source]

Create Credential.

async create_credential_offer(credential_definition_id: str) → str[source]

Create a credential offer for the given credential definition id.

Parameters

credential_definition_id – The credential definition to create an offer for

Returns

The new credential offer

async cred_def_supports_revocation(cred_def_id: str) → bool[source]

Return whether a credential definition supports revocation.

async credential_definition_in_wallet(credential_definition_id: str) → bool[source]

Check whether a given credential definition ID is present in the wallet.

Parameters

credential_definition_id – The credential definition ID to check

async finish_cred_def(job_id: str, cred_def_id: str, options: Optional[dict] = None)[source]

Finish a cred def.

async finish_schema(job_id: str, schema_id: str)[source]

Mark a schema as finished.

async get_created_credential_definitions(issuer_id: Optional[str] = None, schema_issuer_id: Optional[str] = None, schema_id: Optional[str] = None, schema_name: Optional[str] = None, schema_version: Optional[str] = None, epoch: Optional[str] = None) → Sequence[str][source]

Retrieve IDs of credential definitions previously created.

async get_created_schemas(name: Optional[str] = None, version: Optional[str] = None, issuer_id: Optional[str] = None) → Sequence[str][source]

Retrieve IDs of schemas previously created.

async match_created_credential_definitions(cred_def_id: Optional[str] = None, issuer_id: Optional[str] = None, schema_issuer_id: Optional[str] = None, schema_id: Optional[str] = None, schema_name: Optional[str] = None, schema_version: Optional[str] = None, epoch: Optional[str] = None) → Optional[str][source]

Return cred def id of most recent matching cred def.

async notify(event: Event)[source]

Accessor for the event bus instance.

property profile: AskarAnoncredsProfile

Accessor for the profile instance.

async store_credential_definition(schema_result: SchemaResult, cred_def_result: CredDefResult, cred_def_private: anoncreds.CredentialDefinitionPrivate, key_proof: anoncreds.KeyCorrectnessProof, support_revocation: bool, max_cred_num: int, options: Optional[dict] = None)[source]

Store the cred def and it’s components in the wallet.

async store_schema(result: SchemaResult)[source]

Store schema after reaching finished state.

exception aries_cloudagent.anoncreds.issuer.AnonCredsIssuerError(*args, error_code: Optional[str] = None, **kwargs)[source]

Bases: BaseError

Generic issuer error.

aries_cloudagent.anoncreds.registry module

AnonCreds Registry.

class aries_cloudagent.anoncreds.registry.AnonCredsRegistry(registries: Optional[List[BaseAnonCredsHandler]] = None)[source]

Bases: object

AnonCredsRegistry.

async get_credential_definition(profile: Profile, credential_definition_id: str) → GetCredDefResult[source]

Get a credential definition from the registry.

async get_revocation_list(profile: Profile, rev_reg_def_id: str, timestamp: int) → GetRevListResult[source]

Get a revocation list from the registry.

async get_revocation_registry_definition(profile: Profile, revocation_registry_id: str) → GetRevRegDefResult[source]

Get a revocation registry definition from the registry.

async get_schema(profile: Profile, schema_id: str) → GetSchemaResult[source]

Get a schema from the registry.

register(registry: BaseAnonCredsHandler)[source]

Register a new registry.

async register_credential_definition(profile: Profile, schema: GetSchemaResult, credential_definition: CredDef, options: Optional[dict] = None) → CredDefResult[source]

Register a credential definition on the registry.

async register_revocation_list(profile: Profile, rev_reg_def: RevRegDef, rev_list: RevList, options: Optional[dict] = None) → RevListResult[source]

Register a revocation list on the registry.

async register_revocation_registry_definition(profile: Profile, revocation_registry_definition: RevRegDef, options: Optional[dict] = None) → RevRegDefResult[source]

Register a revocation registry definition on the registry.

async register_schema(profile: Profile, schema: AnonCredsSchema, options: Optional[dict] = None) → SchemaResult[source]

Register a schema on the registry.

async update_revocation_list(profile: Profile, rev_reg_def: RevRegDef, prev_list: RevList, curr_list: RevList, revoked: Sequence[int], options: Optional[dict] = None) → RevListResult[source]

Update a revocation list on the registry.

aries_cloudagent.anoncreds.revocation module

Revocation through ledger agnostic AnonCreds interface.

class aries_cloudagent.anoncreds.revocation.AnonCredsRevocation(profile: Profile)[source]

Bases: object

Revocation registry operations manager.

async clear_pending_revocations(txn: ProfileSession, rev_reg_def_id: str, crid_mask: Optional[Sequence[int]] = None)[source]

Clear pending revocations.

async create_and_register_revocation_list(rev_reg_def_id: str, options: Optional[dict] = None)[source]

Create and register a revocation list.

async create_and_register_revocation_registry_definition(issuer_id: str, cred_def_id: str, registry_type: str, tag: str, max_cred_num: int, options: Optional[dict] = None) → RevRegDefResult[source]

Create a new revocation registry and register on network.

Parameters

  • issuer_id (str) – issuer identifier

  • cred_def_id (str) – credential definition identifier

  • registry_type (str) – revocation registry type

  • tag (str) – revocation registry tag

  • max_cred_num (int) – maximum number of credentials supported

  • options (dict) – revocation registry options

Returns

revocation registry definition result

Return type

RevRegDefResult

async create_credential(credential_offer: dict, credential_request: dict, credential_values: dict, *, retries: int = 5) → Tuple[str, str, str][source]

Create a credential.

Parameters

  • credential_offer – Credential Offer to create credential for

  • credential_request – Credential request to create credential for

  • credential_values – Values to go in credential

  • revoc_reg_id – ID of the revocation registry

  • retries – number of times to retry credential creation

Returns

A tuple of created credential and revocation id

async decommission_registry(cred_def_id: str)[source]

Decommission post-init registries and start the next registry generation.

async finish_revocation_list(job_id: str, rev_reg_def_id: str, revoked: list)[source]

Mark a revocation list as finished.

async finish_revocation_registry_definition(job_id: str, rev_reg_def_id: str, options: Optional[dict] = None)[source]

Mark a rev reg def as finished.

generate_public_tails_uri(rev_reg_def: RevRegDef)[source]

Construct tails uri from rev_reg_def.

async get_created_revocation_list(rev_reg_def_id: str) → Optional[RevList][source]

Return rev list from record in wallet.

async get_created_revocation_registry_definition(rev_reg_def_id: str) → Optional[RevRegDef][source]

Retrieve rev reg def by ID from rev reg defs previously created.

async get_created_revocation_registry_definition_state(rev_reg_def_id: str) → Optional[str][source]

Retrieve rev reg def by ID from rev reg defs previously created.

async get_created_revocation_registry_definitions(cred_def_id: Optional[str] = None, state: Optional[str] = None) → Sequence[str][source]

Retrieve IDs of rev reg defs previously created.

get_local_tails_path(rev_reg_def: RevRegDef) → str[source]

Get the local path to the tails file.

async get_or_create_active_registry(cred_def_id: str) → RevRegDefResult[source]

Get or create a revocation registry for the given cred def id.

async get_or_fetch_local_tails_path(rev_reg_def: RevRegDef) → str[source]

Return path to local tails file.

If not present, retrieve from tails server.

async get_pending_revocations(rev_reg_def_id: str) → List[int][source]

Retrieve the list of credential revocation ids pending revocation.

async get_revocation_lists_with_pending_revocations() → Sequence[str][source]

Return a list of rev reg def ids with pending revocations.

async handle_full_registry(rev_reg_def_id: str)[source]

Update the registry status and start the next registry generation.

async mark_pending_revocations(rev_reg_def_id: str, *crids: int)[source]

Cred rev ids stored to publish later.

async notify(event: Event)[source]

Emit an event on the event bus.

property profile: AskarAnoncredsProfile

Accessor for the profile instance.

async retrieve_tails(rev_reg_def: RevRegDef) → str[source]

Retrieve tails file from server.

async revoke_pending_credentials(revoc_reg_id: str, *, additional_crids: Optional[Sequence[int]] = None, limit_crids: Optional[Sequence[int]] = None) → RevokeResult[source]

Revoke a set of credentials in a revocation registry.

Parameters

  • revoc_reg_id – ID of the revocation registry

  • additional_crids – sequences of additional credential indexes to revoke

  • limit_crids – a sequence of credential indexes to limit revocation to If None, all pending revocations will be published. If given, the intersection of pending and limit crids will be published.

Returns

Tuple with the update revocation list, list of cred rev ids not revoked

async set_active_registry(rev_reg_def_id: str)[source]

Mark a registry as active.

async set_rev_reg_state(rev_reg_id, state)[source]

Update Revocation Registry state.

async set_tails_file_public_uri(rev_reg_id, tails_public_uri)[source]

Update Revocation Registry tails file public uri.

async store_revocation_registry_definition(result: RevRegDefResult, rev_reg_def_private: anoncreds.RevocationRegistryDefinitionPrivate, options: Optional[dict] = None)[source]

Store a revocation registry definition.

async store_revocation_registry_list(result: RevListResult)[source]

Store a revocation registry list.

async update_revocation_list(rev_reg_def_id: str, prev: RevList, curr: RevList, revoked: Sequence[int], options: Optional[dict] = None)[source]

Publish and update to a revocation list.

async upload_tails_file(rev_reg_def: RevRegDef)[source]

Upload the local tails file to the tails server.

exception aries_cloudagent.anoncreds.revocation.AnonCredsRevocationError(*args, error_code: Optional[str] = None, **kwargs)[source]

Bases: BaseError

Generic revocation error.

exception aries_cloudagent.anoncreds.revocation.AnonCredsRevocationRegistryFullError(*args, error_code: Optional[str] = None, **kwargs)[source]

Bases: AnonCredsRevocationError

Revocation registry is full when issuing a new credential.

class aries_cloudagent.anoncreds.revocation.RevokeResult(prev: RevList, curr: Optional[RevList] = None, revoked: Optional[Sequence[int]] = None, failed: Optional[Sequence[str]] = None)[source]

Bases: NamedTuple

RevokeResult.

curr: Optional[RevList]

Alias for field number 1

failed: Optional[Sequence[str]]

Alias for field number 3

prev: RevList

Alias for field number 0

revoked: Optional[Sequence[int]]

Alias for field number 2

aries_cloudagent.anoncreds.revocation_setup module

aries_cloudagent.anoncreds.routes module

aries_cloudagent.anoncreds.util module

Utilities for dealing with Indy conventions.

async aries_cloudagent.anoncreds.util.generate_pr_nonce() → str[source]

Generate a nonce for a proof request.

aries_cloudagent.anoncreds.util.handle_value_error(e: ValueError)[source]

Handle ValueError message as web response type.

aries_cloudagent.anoncreds.util.indy_client_dir(subpath: Optional[str] = None, create: bool = False) → str[source]

Return ‘/’-terminated subdirectory of indy-client directory.

Parameters

  • subpath – subpath within indy-client structure

  • create – whether to create subdirectory if absent

aries_cloudagent.anoncreds.verifier module