Source code for aries_cloudagent.protocols.present_proof.v2_0.models.pres_exchange

"""Presentation exchange record."""

import logging
from typing import Any, Mapping, Union

from marshmallow import Schema, fields, validate

from .....core.profile import ProfileSession
from .....messaging.models.base_record import BaseExchangeRecord, BaseExchangeSchema
from .....messaging.valid import UUID4_EXAMPLE
from .....storage.base import StorageError
from ..messages.pres import V20Pres, V20PresSchema
from ..messages.pres_format import V20PresFormat
from ..messages.pres_proposal import V20PresProposal, V20PresProposalSchema
from ..messages.pres_request import V20PresRequest, V20PresRequestSchema
from ..messages.pres_webhook import V20PresExRecordWebhook
from . import UNENCRYPTED_TAGS

LOGGER = logging.getLogger(__name__)


[docs]class V20PresExRecord(BaseExchangeRecord): """Represents a v2.0 presentation exchange."""
[docs] class Meta: """V20PresExRecord metadata.""" schema_class = "V20PresExRecordSchema"
RECORD_TYPE = "pres_ex_v20" RECORD_ID_NAME = "pres_ex_id" RECORD_TOPIC = "present_proof_v2_0" TAG_NAMES = {"~thread_id"} if UNENCRYPTED_TAGS else {"thread_id"} INITIATOR_SELF = "self" INITIATOR_EXTERNAL = "external" ROLE_PROVER = "prover" ROLE_VERIFIER = "verifier" STATE_PROPOSAL_SENT = "proposal-sent" STATE_PROPOSAL_RECEIVED = "proposal-received" STATE_REQUEST_SENT = "request-sent" STATE_REQUEST_RECEIVED = "request-received" STATE_PRESENTATION_SENT = "presentation-sent" STATE_PRESENTATION_RECEIVED = "presentation-received" STATE_DONE = "done" STATE_ABANDONED = "abandoned" def __init__( self, *, pres_ex_id: str = None, connection_id: str = None, thread_id: str = None, initiator: str = None, role: str = None, state: str = None, pres_proposal: Union[V20PresProposal, Mapping] = None, # aries message pres_request: Union[V20PresRequest, Mapping] = None, # aries message pres: Union[V20Pres, Mapping] = None, # aries message verified: str = None, verified_msgs: list = None, auto_present: bool = False, auto_verify: bool = False, error_msg: str = None, trace: bool = False, # backward compat: BaseRecord.FromStorage() by_format: Mapping = None, # backward compat: BaseRecord.FromStorage() auto_remove: bool = False, **kwargs, ): """Initialize a new PresExRecord.""" super().__init__(pres_ex_id, state, trace=trace, **kwargs) self.connection_id = connection_id self.thread_id = thread_id self.initiator = initiator self.role = role self.state = state self._pres_proposal = V20PresProposal.serde(pres_proposal) self._pres_request = V20PresRequest.serde(pres_request) self._pres = V20Pres.serde(pres) self.verified = verified self.verified_msgs = verified_msgs self.auto_present = auto_present self.auto_verify = auto_verify self.error_msg = error_msg self.auto_remove = auto_remove @property def pres_ex_id(self) -> str: """Accessor for the ID associated with this exchange record.""" return self._id @property def by_format(self) -> Mapping: """Record proposal, request, and presentation attachments by format.""" result = {} for item, cls in { "pres_proposal": V20PresProposal, # note: proof request attached for indy "pres_request": V20PresRequest, "pres": V20Pres, }.items(): msg = getattr(self, item) if msg: result.update( { item: { V20PresFormat.Format.get(f.format).api: msg.attachment( V20PresFormat.Format.get(f.format) ) for f in msg.formats } } ) return result @property def pres_proposal(self) -> V20PresProposal: """Accessor; get deserialized view.""" return None if self._pres_proposal is None else self._pres_proposal.de @pres_proposal.setter def pres_proposal(self, value): """Setter; store de/serialized views.""" self._pres_proposal = V20PresProposal.serde(value) @property def pres_request(self) -> V20PresRequest: """Accessor; get deserialized view.""" return None if self._pres_request is None else self._pres_request.de @pres_request.setter def pres_request(self, value): """Setter; store de/serialized views.""" self._pres_request = V20PresRequest.serde(value) @property def pres(self) -> V20Pres: """Accessor; get deserialized view.""" return None if self._pres is None else self._pres.de @pres.setter def pres(self, value): """Setter; store de/serialized views.""" self._pres = V20Pres.serde(value)
[docs] async def save_error_state( self, session: ProfileSession, *, state: str = None, reason: str = None, log_params: Mapping[str, Any] = None, log_override: bool = False, ): """Save record error state if need be; log and swallow any storage error. Args: session: The profile session to use reason: A reason to add to the log log_params: Additional parameters to log override: Override configured logging regimen, print to stderr instead """ if self._last_state == state: # already done return self.state = state or V20PresExRecord.STATE_ABANDONED if reason: self.error_msg = reason try: await self.save( session, reason=reason, log_params=log_params, log_override=log_override, ) except StorageError as err: LOGGER.exception(err)
# Override
[docs] async def emit_event(self, session: ProfileSession, payload: Any = None): """Emit an event. Args: session: The profile session to use payload: The event payload """ if not self.RECORD_TOPIC: return if self.state: topic = f"{self.EVENT_NAMESPACE}::{self.RECORD_TOPIC}::{self.state}" else: topic = f"{self.EVENT_NAMESPACE}::{self.RECORD_TOPIC}" if session.profile.settings.get("debug.webhooks"): if not payload: payload = self.serialize() else: payload = V20PresExRecordWebhook(**self.__dict__) payload = payload.__dict__ await session.profile.notify(topic, payload)
@property def record_value(self) -> Mapping: """Accessor for the JSON record value generated for this credential exchange.""" return { **{ prop: getattr(self, prop) for prop in ( "connection_id", "initiator", "role", "state", "verified", "verified_msgs", "auto_present", "auto_verify", "error_msg", "trace", "auto_remove", ) }, **{ prop: getattr(self, f"_{prop}").ser for prop in ( "pres_proposal", "pres_request", "pres", ) if getattr(self, prop) is not None }, } def __eq__(self, other: Any) -> bool: """Comparison between records.""" return super().__eq__(other)
[docs]class V20PresExRecordSchema(BaseExchangeSchema): """Schema for de/serialization of v2.0 presentation exchange records."""
[docs] class Meta: """V20PresExRecordSchema metadata.""" model_class = V20PresExRecord
pres_ex_id = fields.Str( required=False, metadata={ "description": "Presentation exchange identifier", "example": UUID4_EXAMPLE, }, ) connection_id = fields.Str( required=False, metadata={"description": "Connection identifier", "example": UUID4_EXAMPLE}, ) thread_id = fields.Str( required=False, metadata={"description": "Thread identifier", "example": UUID4_EXAMPLE}, ) initiator = fields.Str( required=False, validate=validate.OneOf( V20PresExRecord.get_attributes_by_prefix("INITIATOR_", walk_mro=False) ), metadata={ "description": "Present-proof exchange initiator: self or external", "example": V20PresExRecord.INITIATOR_SELF, }, ) role = fields.Str( required=False, validate=validate.OneOf( V20PresExRecord.get_attributes_by_prefix("ROLE_", walk_mro=False) ), metadata={ "description": "Present-proof exchange role: prover or verifier", "example": V20PresExRecord.ROLE_PROVER, }, ) state = fields.Str( required=False, validate=validate.OneOf( V20PresExRecord.get_attributes_by_prefix("STATE_", walk_mro=True) ), metadata={"description": "Present-proof exchange state"}, ) pres_proposal = fields.Nested( V20PresProposalSchema(), required=False, metadata={"description": "Presentation proposal message"}, ) pres_request = fields.Nested( V20PresRequestSchema(), required=False, metadata={"description": "Presentation request message"}, ) pres = fields.Nested( V20PresSchema(), required=False, metadata={"description": "Presentation message"}, ) by_format = fields.Nested( Schema.from_dict( { "pres_proposal": fields.Dict(required=False), "pres_request": fields.Dict(required=False), "pres": fields.Dict(required=False), }, name="V20PresExRecordByFormatSchema", ), required=False, dump_only=True, metadata={ "description": ( "Attachment content by format for proposal, request, and presentation" ) }, ) verified = fields.Str( required=False, validate=validate.OneOf(["true", "false"]), metadata={ "description": "Whether presentation is verified: 'true' or 'false'", "example": "true", }, ) verified_msgs = fields.List( fields.Str( required=False, metadata={"description": "Proof verification warning or error information"}, ), required=False, ) auto_present = fields.Bool( required=False, metadata={ "description": "Prover choice to auto-present proof as verifier requests", "example": False, }, ) auto_verify = fields.Bool( required=False, metadata={"description": "Verifier choice to auto-verify proof presentation"}, ) error_msg = fields.Str( required=False, metadata={"description": "Error message", "example": "Invalid structure"}, ) auto_remove = fields.Bool( required=False, dump_default=True, metadata={ "description": ( "Verifier choice to remove this presentation exchange record when" " complete" ), "example": False, }, )