"""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."""
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."""
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,
},
)