forked from openwallet-foundation/acapy
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathwire_format.py
More file actions
191 lines (142 loc) · 5.28 KB
/
Copy pathwire_format.py
File metadata and controls
191 lines (142 loc) · 5.28 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
"""Abstract wire format classes."""
import json
import logging
from abc import abstractmethod
from typing import List, Sequence, Tuple, Union
from ..core.profile import ProfileSession
from ..messaging.util import time_now
from .inbound.receipt import MessageReceipt
from .error import WireFormatParseError
LOGGER = logging.getLogger(__name__)
DIDCOMM_V0_MIME_TYPE = "application/ssi-agent-wire"
DIDCOMM_V1_MIME_TYPE = "application/didcomm-envelope-enc"
class BaseWireFormat:
"""Abstract messaging wire format."""
def __init__(self):
"""Initialize the base wire format instance."""
@abstractmethod
async def parse_message(
self,
session: ProfileSession,
message_body: Union[str, bytes],
) -> Tuple[dict, MessageReceipt]:
"""
Deserialize an incoming message and further populate the request context.
Args:
session: The profile session for providing wallet access
message_body: The body of the message
Returns:
A tuple of the parsed message and a message receipt instance
Raises:
WireFormatParseError: If the message can't be parsed
"""
@abstractmethod
async def encode_message(
self,
session: ProfileSession,
message_json: Union[str, bytes],
recipient_keys: Sequence[str],
routing_keys: Sequence[str],
sender_key: str,
) -> Union[str, bytes]:
"""
Encode an outgoing message for transport.
Args:
session: The profile session for providing wallet access
message_json: The message body to serialize
recipient_keys: A sequence of recipient verkeys
routing_keys: A sequence of routing verkeys
sender_key: The verification key of the sending agent
Returns:
The encoded message
Raises:
MessageEncodeError: If the message could not be encoded
"""
@abstractmethod
def get_recipient_keys(self, message_body: Union[str, bytes]) -> List[str]:
"""
Get all recipient keys from a wire message.
Args:
message_body: The body of the message
Returns:
List of recipient keys from the message body
Raises:
RecipientKeysError: If the recipient keys could not be extracted
"""
class JsonWireFormat(BaseWireFormat):
"""Unencrypted wire format."""
@abstractmethod
async def parse_message(
self,
session: ProfileSession,
message_body: Union[str, bytes],
) -> Tuple[dict, MessageReceipt]:
"""
Deserialize an incoming message and further populate the request context.
Args:
session: The profile session for providing wallet access
message_body: The body of the message
Returns:
A tuple of the parsed message and a message receipt instance
Raises:
WireFormatParseError: If the JSON parsing failed
"""
receipt = MessageReceipt()
receipt.in_time = time_now()
receipt.raw_message = message_body
message_dict = None
message_json = message_body
if not message_json:
raise WireFormatParseError("Message body is empty")
try:
message_dict = json.loads(message_json)
except ValueError:
raise WireFormatParseError("Message JSON parsing failed")
if not isinstance(message_dict, dict):
raise WireFormatParseError("Message JSON result is not an object")
# parse thread ID
thread_dec = message_dict.get("~thread")
receipt.thread_id = (
thread_dec and thread_dec.get("thid") or message_dict.get("@id")
)
# handle transport decorator
transport_dec = message_dict.get("~transport")
if transport_dec:
receipt.direct_response_mode = transport_dec.get("return_route")
LOGGER.debug(f"Expanded message: {message_dict}")
return message_dict, receipt
@abstractmethod
async def encode_message(
self,
session: ProfileSession,
message_json: Union[str, bytes],
recipient_keys: Sequence[str],
routing_keys: Sequence[str],
sender_key: str,
) -> Union[str, bytes]:
"""
Encode an outgoing message for transport.
Args:
session: The profile session for providing wallet access
message_json: The message body to serialize
recipient_keys: A sequence of recipient verkeys
routing_keys: A sequence of routing verkeys
sender_key: The verification key of the sending agent
Returns:
The encoded message
Raises:
MessageEncodeError: If the message could not be encoded
"""
return message_json
def get_recipient_keys(self, message_body: Union[str, bytes]) -> List[str]:
"""
Get all recipient keys from a wire message.
Args:
message_body: The body of the message
Returns:
List of recipient keys from the message body
Raises:
RecipientKeysError: If the recipient keys could not be extracted
"""
# JSON message cannot contain recipient keys
return []