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
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
use affinidi_did_resolver_cache_sdk::{
DIDCacheClient, DIDMethod, ResolveResponse,
networking::{WSRequest, WSResponse, WSResponseError, WSResponseType},
};
use axum::{
extract::{
State, WebSocketUpgrade,
ws::{Message, WebSocket},
},
response::IntoResponse,
};
use tokio::select;
use tracing::{Instrument, debug, info, span, warn};
use crate::SharedData;
/// For did:webvh DIDs, fetch the raw DID log (did.jsonl) from the source HTTP endpoint.
/// This log is sent alongside the resolved document so clients can independently
/// verify the cryptographic chain, preventing a compromised cache server from
/// serving tampered DID documents.
async fn fetch_webvh_log(did: &str) -> (Option<String>, Option<String>) {
let parsed_url = match didwebvh_rs::url::WebVHURL::parse_did_url(did) {
Ok(url) => url,
Err(e) => {
warn!("Failed to parse WebVH DID URL for log fetch: {e}");
return (None, None);
}
};
let log_url = match parsed_url.get_http_url(Some("did.jsonl")) {
Ok(url) => url,
Err(e) => {
warn!("Failed to construct log URL for WebVH DID: {e}");
return (None, None);
}
};
let client = match reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(10))
.build()
{
Ok(c) => c,
Err(e) => {
warn!("Failed to create HTTP client for WebVH log fetch: {e}");
return (None, None);
}
};
let did_log = match client.get(log_url).send().await {
Ok(resp) if resp.status().is_success() => match resp.text().await {
Ok(text) => Some(text),
Err(e) => {
warn!("Failed to read WebVH log response body: {e}");
None
}
},
Ok(resp) => {
warn!("WebVH log fetch returned HTTP {}: {}", resp.status(), did);
None
}
Err(e) => {
warn!("Failed to fetch WebVH log for {}: {e}", did);
None
}
};
// Fetch witness proofs if log was successfully retrieved
let did_witness_log = if did_log.is_some() {
let witness_url = match parsed_url.get_http_url(Some("did-witness.json")) {
Ok(url) => url,
Err(_) => return (did_log, None),
};
match client.get(witness_url).send().await {
Ok(resp) if resp.status().is_success() => resp.text().await.ok(),
_ => None,
}
} else {
None
};
(did_log, did_witness_log)
}
/// Build a WSResponse, fetching the raw DID log for WebVH DIDs.
async fn build_response(response: ResolveResponse) -> WSResponseType {
let (did_log, did_witness_log) = if response.method == DIDMethod::WEBVH {
fetch_webvh_log(&response.did).await
} else {
(None, None)
};
WSResponseType::Response(Box::new(WSResponse {
did: response.did.clone(),
hash: response.did_hash,
document: response.doc,
did_log,
did_witness_log,
}))
}
// Handles the switching of the protocol to a websocket connection
pub async fn websocket_handler(
//session: Session,
ws: WebSocketUpgrade,
State(state): State<SharedData>,
) -> impl IntoResponse {
let _span = span!(
tracing::Level::DEBUG,
"websocket_handler",
// session = session.session_id
);
/*async move { ws.on_upgrade(move |socket| handle_socket(socket, state, session)) }
.instrument(_span)
.await*/
async move { ws.on_upgrade(move |socket| handle_socket(socket, state)) }
.instrument(_span)
.await
}
/// WebSocket state machine. This is spawned per connection.
//async fn handle_socket(mut socket: WebSocket, state: SharedData, session: Session) {
async fn handle_socket(mut socket: WebSocket, state: SharedData) {
let _span = span!(
tracing::Level::DEBUG,
"handle_socket",
//session = session.session_id
);
async move {
state.stats().await.increment_ws_opened();
info!("Websocket connection established");
loop {
select! {
value = socket.recv() => {
if let Some(msg) = value {
match msg {
Ok(msg) => {
match msg {
Message::Text(msg) => {
debug!("ws: Received text message: {:?}", msg);
let request: WSRequest = match serde_json::from_str(&msg) {
Ok(request) => request,
Err(e) => {
warn!("ws: Error parsing message: {:?}", e);
break;
}
};
match state.resolver.resolve(&request.did).await {
Ok(response) => {
let mut stats = state.stats().await;
stats.increment_resolver_success();
if response.cache_hit { stats.increment_cache_hit();}
stats.increment_did_method_success(response.method.clone());
drop(stats);
debug!("resolved DID: ({}) cache_hit?({})", response.did, response.cache_hit);
let message = build_response(response).await;
if let Err(e) = socket.send(Message::Text(serde_json::to_string(&message).unwrap().into())).await {
warn!("ws: Error sending response: {:?}", e);
break;
} else {
debug!("Sent response: {:?}", message);
}
}
Err(e) => {
// Couldn't resolve the DID, send an error back
let hash = DIDCacheClient::hash_did(&request.did);
warn!("Couldn't resolve DID: ({}) Reason: {}", &request.did, e);
state.stats().await.increment_resolver_error();
if let Err(e) = socket.send(Message::Text(serde_json::to_string(&WSResponseType::Error(WSResponseError {did: request.did, hash, error: e.to_string()})).unwrap().into())).await {
warn!("ws: Error sending error response: {:?}", e);
break;
}
}
}
}
Message::Binary(msg) => {
debug!("ws: Received binary message: {:?}", msg);
let request: WSRequest = match serde_json::from_slice(msg.as_ref()) {
Ok(request) => request,
Err(e) => {
warn!("ws: Error parsing message: {:?}", e);
break;
}
};
match state.resolver.resolve(&request.did).await {
Ok(response) => {
let mut stats = state.stats().await;
stats.increment_resolver_success();
if response.cache_hit { stats.increment_cache_hit();}
stats.increment_did_method_success(response.method.clone());
drop(stats);
debug!("resolved DID: ({}) cache_hit?({})", response.did, response.cache_hit);
let message = build_response(response).await;
if let Err(e) = socket.send(Message::Text(serde_json::to_string(&message).unwrap().into())).await {
warn!("ws: Error sending response: {:?}", e);
break;
} else {
debug!("Sent response: {:?}", message);
}
}
Err(e) => {
// Couldn't resolve the DID, send an error back
let hash = DIDCacheClient::hash_did(&request.did);
warn!("Couldn't resolve DID: ({}) Reason: {}", &request.did, e);
state.stats().await.increment_resolver_error();
if let Err(e) = socket.send(Message::Text(serde_json::to_string(&WSResponseType::Error(WSResponseError {did: request.did, hash, error: e.to_string()})).unwrap().into())).await {
warn!("ws: Error sending error response: {:?}", e);
break;
}
}
}
}
Message::Ping(_) => {
// Don't need to do anything, the library will automatically respond with a pong
}
Message::Pong(_) => {
// Don't need to do anything
}
Message::Close(_) => {
debug!("Received close message, closing connection");
break;
}
}
}
Err(err) => {
warn!("Error receiving message: {:?}", err);
continue;
}
}
} else {
debug!("Received None, closing connection");
break;
}
}
}
}
// We're done, close the connection
state.stats().await.increment_ws_closed();
info!("Websocket connection closed");
}
.instrument(_span)
.await
}