Microservice to bring 2FA to self hosted PDSes
1use crate::AppState;
2use crate::helpers::TokenCheckError::InvalidToken;
3use anyhow::anyhow;
4use axum::{
5 body::{Body, to_bytes},
6 extract::Request,
7 http::header::CONTENT_TYPE,
8 http::{HeaderMap, StatusCode, Uri},
9 response::{IntoResponse, Response},
10};
11use axum_template::TemplateEngine;
12use chrono::Utc;
13use jacquard_common::{
14 service_auth, service_auth::PublicKey, types::did::Did, types::did_doc::VerificationMethod,
15 types::nsid::Nsid,
16};
17use jacquard_identity::{PublicResolver, resolver::IdentityResolver};
18use josekit::jwe::alg::direct::DirectJweAlgorithm;
19use lettre::{
20 Message,
21 message::{MultiPart, SinglePart, header},
22};
23use rand::Rng;
24use serde::de::DeserializeOwned;
25use serde_json::{Map, Value};
26use sha2::{Digest, Sha256};
27use sqlx::SqlitePool;
28use std::sync::Arc;
29use tracing::{error, log};
30
31///Used to generate the email 2fa code
32const UPPERCASE_BASE32_CHARS: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ234567";
33
34/// The result of a proxied call that attempts to parse JSON.
35pub enum ProxiedResult<T> {
36 /// Successfully parsed JSON body along with original response headers.
37 Parsed { value: T, _headers: HeaderMap },
38 /// Could not or should not parse: return the original (or rebuilt) response as-is.
39 Passthrough(Response<Body>),
40}
41
42/// Proxy the incoming request to the PDS base URL plus the provided path and attempt to parse
43/// the successful response body as JSON into `T`.
44///
45pub async fn proxy_get_json<T>(
46 state: &AppState,
47 mut req: Request,
48 path: &str,
49) -> Result<ProxiedResult<T>, StatusCode>
50where
51 T: DeserializeOwned,
52{
53 let uri = format!("{}{}", state.app_config.pds_base_url, path);
54 *req.uri_mut() = Uri::try_from(uri).map_err(|_| StatusCode::BAD_REQUEST)?;
55
56 let result = state
57 .reverse_proxy_client
58 .request(req)
59 .await
60 .map_err(|_| StatusCode::BAD_REQUEST)?
61 .into_response();
62
63 if result.status() != StatusCode::OK {
64 return Ok(ProxiedResult::Passthrough(result));
65 }
66
67 let response_headers = result.headers().clone();
68 let body = result.into_body();
69 let body_bytes = to_bytes(body, usize::MAX)
70 .await
71 .map_err(|_| StatusCode::BAD_REQUEST)?;
72
73 match serde_json::from_slice::<T>(&body_bytes) {
74 Ok(value) => Ok(ProxiedResult::Parsed {
75 value,
76 _headers: response_headers,
77 }),
78 Err(err) => {
79 error!(%err, "failed to parse proxied JSON response; returning original body");
80 let mut builder = Response::builder().status(StatusCode::OK);
81 if let Some(headers) = builder.headers_mut() {
82 *headers = response_headers;
83 }
84 let resp = builder
85 .body(Body::from(body_bytes))
86 .map_err(|_| StatusCode::BAD_REQUEST)?;
87 Ok(ProxiedResult::Passthrough(resp))
88 }
89 }
90}
91
92/// Forward an incoming request to the PDS as a POST, filtering out hop-by-hop headers.
93pub async fn proxy_post(
94 state: &AppState,
95 headers: &HeaderMap,
96 path: &str,
97 body_bytes: Vec<u8>,
98) -> Result<Response<Body>, StatusCode> {
99 let uri = format!("{}{}", state.app_config.pds_base_url, path);
100
101 let mut req = axum::http::Request::post(uri);
102 if let Some(req_headers) = req.headers_mut() {
103 for (name, value) in headers.iter() {
104 let skip = matches!(
105 name.as_str(),
106 "host" | "content-length" | "transfer-encoding" | "connection"
107 );
108 if !skip {
109 req_headers.append(name.clone(), value.clone());
110 }
111 }
112 }
113
114 let req = req
115 .body(Body::from(body_bytes))
116 .map_err(|_| StatusCode::BAD_REQUEST)?;
117
118 let proxied = state
119 .reverse_proxy_client
120 .request(req)
121 .await
122 .map_err(|_| StatusCode::BAD_REQUEST)?
123 .into_response();
124
125 Ok(proxied)
126}
127
128/// Build a JSON error response with the required Content-Type header
129/// Content-Type: application/json;charset=utf-8
130/// Body shape: { "error": string, "message": string }
131pub fn json_error_response(
132 status: StatusCode,
133 error: impl Into<String>,
134 message: impl Into<String>,
135) -> Result<Response<Body>, StatusCode> {
136 let body_str = match serde_json::to_string(&serde_json::json!({
137 "error": error.into(),
138 "message": message.into(),
139 })) {
140 Ok(s) => s,
141 Err(_) => return Err(StatusCode::BAD_REQUEST),
142 };
143
144 Response::builder()
145 .status(status)
146 .header(CONTENT_TYPE, "application/json;charset=utf-8")
147 .body(Body::from(body_str))
148 .map_err(|_| StatusCode::BAD_REQUEST)
149}
150
151/// Build a JSON error response with the required Content-Type header
152/// Content-Type: application/json (oauth endpoint does not like utf ending)
153/// Body shape: { "error": string, "error_description": string }
154pub fn oauth_json_error_response(
155 status: StatusCode,
156 error: impl Into<String>,
157 message: impl Into<String>,
158) -> Result<Response<Body>, StatusCode> {
159 let body_str = match serde_json::to_string(&serde_json::json!({
160 "error": error.into(),
161 "error_description": message.into(),
162 })) {
163 Ok(s) => s,
164 Err(_) => return Err(StatusCode::BAD_REQUEST),
165 };
166
167 Response::builder()
168 .status(status)
169 .header(CONTENT_TYPE, "application/json")
170 .body(Body::from(body_str))
171 .map_err(|_| StatusCode::BAD_REQUEST)
172}
173
174/// Creates a random token of 10 characters for email 2FA
175pub fn get_random_token() -> String {
176 let mut rng = rand::rng();
177
178 let mut full_code = String::with_capacity(10);
179 for _ in 0..10 {
180 let idx = rng.random_range(0..UPPERCASE_BASE32_CHARS.len());
181 full_code.push(UPPERCASE_BASE32_CHARS[idx] as char);
182 }
183
184 let slice_one = &full_code[0..5];
185 let slice_two = &full_code[5..10];
186 format!("{slice_one}-{slice_two}")
187}
188
189pub enum TokenCheckError {
190 InvalidToken,
191 ExpiredToken,
192}
193
194pub enum AuthResult {
195 WrongIdentityOrPassword,
196 /// The string here is the email address to create a hint for oauth
197 TwoFactorRequired(String),
198 /// User does not have 2FA enabled, or using an app password, or passes it
199 ProxyThrough,
200 TokenCheckFailed(TokenCheckError),
201}
202
203pub enum IdentifierType {
204 Email,
205 Did,
206 Handle,
207}
208
209impl IdentifierType {
210 fn what_is_it(identifier: String) -> Self {
211 if identifier.contains("@") {
212 IdentifierType::Email
213 } else if identifier.contains("did:") {
214 IdentifierType::Did
215 } else {
216 IdentifierType::Handle
217 }
218 }
219}
220
221/// Creates a hex string from the password and salt to find app passwords
222fn scrypt_hex(password: &str, salt: &str) -> anyhow::Result<String> {
223 let params = scrypt::Params::new(14, 8, 1, 64)?;
224 let mut derived = [0u8; 64];
225 scrypt::scrypt(password.as_bytes(), salt.as_bytes(), ¶ms, &mut derived)?;
226 Ok(hex::encode(derived))
227}
228
229/// Hashes the app password. did is used as the salt.
230pub fn hash_app_password(did: &str, password: &str) -> anyhow::Result<String> {
231 let mut hasher = Sha256::new();
232 hasher.update(did.as_bytes());
233 let sha = hasher.finalize();
234 let salt = hex::encode(&sha[..16]);
235 let hash_hex = scrypt_hex(password, &salt)?;
236 Ok(format!("{salt}:{hash_hex}"))
237}
238
239async fn verify_password(password: &str, password_scrypt: &str) -> anyhow::Result<bool> {
240 // Expected format: "salt:hash" where hash is hex of scrypt(password, salt, 64 bytes)
241 let mut parts = password_scrypt.splitn(2, ':');
242 let salt = match parts.next() {
243 Some(s) if !s.is_empty() => s,
244 _ => return Ok(false),
245 };
246 let stored_hash_hex = match parts.next() {
247 Some(h) if !h.is_empty() => h,
248 _ => return Ok(false),
249 };
250
251 // Derive using the shared helper and compare
252 let derived_hex = match scrypt_hex(password, salt) {
253 Ok(h) => h,
254 Err(_) => return Ok(false),
255 };
256
257 Ok(derived_hex.as_str() == stored_hash_hex)
258}
259
260/// Handles the auth checks along with sending a 2fa email
261pub async fn preauth_check(
262 state: &AppState,
263 identifier: &str,
264 password: &str,
265 two_factor_code: Option<String>,
266 oauth: bool,
267) -> anyhow::Result<AuthResult> {
268 // Determine identifier type
269 let id_type = IdentifierType::what_is_it(identifier.to_string());
270
271 // Query account DB for did and passwordScrypt based on identifier type
272 let account_row: Option<(String, String, String, String)> = match id_type {
273 IdentifierType::Email => {
274 sqlx::query_as::<_, (String, String, String, String)>(
275 "SELECT account.did, account.passwordScrypt, account.email, actor.handle
276 FROM actor
277 LEFT JOIN account ON actor.did = account.did
278 where account.email = ? LIMIT 1",
279 )
280 .bind(identifier)
281 .fetch_optional(&state.account_pool)
282 .await?
283 }
284 IdentifierType::Handle => {
285 sqlx::query_as::<_, (String, String, String, String)>(
286 "SELECT account.did, account.passwordScrypt, account.email, actor.handle
287 FROM actor
288 LEFT JOIN account ON actor.did = account.did
289 where actor.handle = ? LIMIT 1",
290 )
291 .bind(identifier)
292 .fetch_optional(&state.account_pool)
293 .await?
294 }
295 IdentifierType::Did => {
296 sqlx::query_as::<_, (String, String, String, String)>(
297 "SELECT account.did, account.passwordScrypt, account.email, actor.handle
298 FROM actor
299 LEFT JOIN account ON actor.did = account.did
300 where account.did = ? LIMIT 1",
301 )
302 .bind(identifier)
303 .fetch_optional(&state.account_pool)
304 .await?
305 }
306 };
307
308 if let Some((did, password_scrypt, email, handle)) = account_row {
309 // Verify password before proceeding to 2FA email step
310 let verified = verify_password(password, &password_scrypt).await?;
311 if !verified {
312 if oauth {
313 //OAuth does not allow app password logins so just go ahead and send it along it's way
314 return Ok(AuthResult::WrongIdentityOrPassword);
315 }
316 //Theres a chance it could be an app password so check that as well
317 return match verify_app_password(&state.account_pool, &did, password).await {
318 Ok(valid) => {
319 if valid {
320 //Was a valid app password up to the PDS now
321 Ok(AuthResult::ProxyThrough)
322 } else {
323 Ok(AuthResult::WrongIdentityOrPassword)
324 }
325 }
326 Err(err) => {
327 log::error!("Error checking the app password: {err}");
328 Err(err)
329 }
330 };
331 }
332
333 // Check two-factor requirement for this DID in the gatekeeper DB
334 let required_opt = sqlx::query_as::<_, (u8,)>(
335 "SELECT required FROM two_factor_accounts WHERE did = ? LIMIT 1",
336 )
337 .bind(did.clone())
338 .fetch_optional(&state.pds_gatekeeper_pool)
339 .await?;
340
341 let two_factor_required = match required_opt {
342 Some(row) => row.0 != 0,
343 None => false,
344 };
345
346 if two_factor_required {
347 //Two factor is required and a taken was provided
348 if let Some(two_factor_code) = two_factor_code {
349 //if the two_factor_code is set need to see if we have a valid token
350 if !two_factor_code.is_empty() {
351 return match assert_valid_token(
352 &state.account_pool,
353 did.clone(),
354 two_factor_code,
355 )
356 .await
357 {
358 Ok(_) => {
359 let result_of_cleanup =
360 delete_all_email_tokens(&state.account_pool, did.clone()).await;
361 if result_of_cleanup.is_err() {
362 log::error!(
363 "There was an error deleting the email tokens after login: {:?}",
364 result_of_cleanup.err()
365 )
366 }
367 Ok(AuthResult::ProxyThrough)
368 }
369 Err(err) => Ok(AuthResult::TokenCheckFailed(err)),
370 };
371 }
372 }
373
374 return match create_two_factor_token(&state.account_pool, did).await {
375 Ok(code) => {
376 let mut email_data = Map::new();
377 email_data.insert("token".to_string(), Value::from(code.clone()));
378 email_data.insert("handle".to_string(), Value::from(handle.clone()));
379 let email_body = state
380 .template_engine
381 .render("two_factor_code.hbs", email_data)?;
382
383 let email_message = Message::builder()
384 //TODO prob get the proper type in the state
385 .from(state.app_config.mailer_from.parse()?)
386 .to(email.parse()?)
387 .subject(&state.app_config.email_subject)
388 .multipart(
389 MultiPart::alternative() // This is composed of two parts.
390 .singlepart(
391 SinglePart::builder()
392 .header(header::ContentType::TEXT_PLAIN)
393 .body(format!("We received a sign-in request for the account @{handle}. Use the code: {code} to sign in. If this wasn't you, we recommend taking steps to protect your account by changing your password at https://bsky.app/settings.")), // Every message should have a plain text fallback.
394 )
395 .singlepart(
396 SinglePart::builder()
397 .header(header::ContentType::TEXT_HTML)
398 .body(email_body),
399 ),
400 )?;
401 match state.mailer.send(email_message).await {
402 Ok(_) => Ok(AuthResult::TwoFactorRequired(mask_email(email))),
403 Err(err) => {
404 log::error!("Error sending the 2FA email: {err}");
405 Err(anyhow!(err))
406 }
407 }
408 }
409 Err(err) => {
410 log::error!("error on creating a 2fa token: {err}");
411 Err(anyhow!(err))
412 }
413 };
414 }
415 }
416
417 // No local 2FA requirement (or account not found)
418 Ok(AuthResult::ProxyThrough)
419}
420
421pub async fn create_two_factor_token(
422 account_db: &SqlitePool,
423 did: String,
424) -> anyhow::Result<String> {
425 let purpose = "2fa_code";
426
427 let token = get_random_token();
428 let right_now = Utc::now();
429
430 let res = sqlx::query(
431 "INSERT INTO email_token (purpose, did, token, requestedAt)
432 VALUES (?, ?, ?, ?)
433 ON CONFLICT(purpose, did) DO UPDATE SET
434 token=excluded.token,
435 requestedAt=excluded.requestedAt
436 WHERE did=excluded.did",
437 )
438 .bind(purpose)
439 .bind(&did)
440 .bind(&token)
441 .bind(right_now)
442 .execute(account_db)
443 .await;
444
445 match res {
446 Ok(_) => Ok(token),
447 Err(err) => {
448 log::error!("Error creating a two factor token: {err}");
449 Err(anyhow::anyhow!(err))
450 }
451 }
452}
453
454pub async fn delete_all_email_tokens(account_db: &SqlitePool, did: String) -> anyhow::Result<()> {
455 sqlx::query("DELETE FROM email_token WHERE did = ?")
456 .bind(did)
457 .execute(account_db)
458 .await?;
459 Ok(())
460}
461
462pub async fn assert_valid_token(
463 account_db: &SqlitePool,
464 did: String,
465 token: String,
466) -> Result<(), TokenCheckError> {
467 let token_upper = token.to_ascii_uppercase();
468 let purpose = "2fa_code";
469
470 let row: Option<(String,)> = sqlx::query_as(
471 "SELECT requestedAt FROM email_token WHERE purpose = ? AND did = ? AND token = ? LIMIT 1",
472 )
473 .bind(purpose)
474 .bind(did)
475 .bind(token_upper)
476 .fetch_optional(account_db)
477 .await
478 .map_err(|err| {
479 log::error!("Error getting the 2fa token: {err}");
480 InvalidToken
481 })?;
482
483 match row {
484 None => Err(InvalidToken),
485 Some(row) => {
486 // Token lives for 15 minutes
487 let expiration_ms = 15 * 60_000;
488
489 let requested_at_utc = match chrono::DateTime::parse_from_rfc3339(&row.0) {
490 Ok(dt) => dt.with_timezone(&Utc),
491 Err(_) => {
492 return Err(TokenCheckError::InvalidToken);
493 }
494 };
495
496 let now = Utc::now();
497 let age_ms = (now - requested_at_utc).num_milliseconds();
498 let expired = age_ms > expiration_ms;
499 if expired {
500 return Err(TokenCheckError::ExpiredToken);
501 }
502
503 Ok(())
504 }
505 }
506}
507
508/// We just need to confirm if it's there or not. Will let the PDS do the actual figuring of permissions
509pub async fn verify_app_password(
510 account_db: &SqlitePool,
511 did: &str,
512 password: &str,
513) -> anyhow::Result<bool> {
514 let password_scrypt = hash_app_password(did, password)?;
515
516 let row: Option<(i64,)> = sqlx::query_as(
517 "SELECT Count(*) FROM app_password WHERE did = ? AND passwordScrypt = ? LIMIT 1",
518 )
519 .bind(did)
520 .bind(password_scrypt)
521 .fetch_optional(account_db)
522 .await?;
523
524 Ok(match row {
525 None => false,
526 Some((count,)) => count > 0,
527 })
528}
529
530/// Mask an email address into a hint like "2***0@p***m".
531pub fn mask_email(email: String) -> String {
532 // Basic split on first '@'
533 let mut parts = email.splitn(2, '@');
534 let local = match parts.next() {
535 Some(l) => l,
536 None => return email.to_string(),
537 };
538 let domain_rest = match parts.next() {
539 Some(d) if !d.is_empty() => d,
540 _ => return email.to_string(),
541 };
542
543 // Helper to mask a single label (keep first and last, middle becomes ***).
544 fn mask_label(s: &str) -> String {
545 let chars: Vec<char> = s.chars().collect();
546 match chars.len() {
547 0 => String::new(),
548 1 => format!("{}***", chars[0]),
549 2 => format!("{}***{}", chars[0], chars[1]),
550 _ => format!("{}***{}", chars[0], chars[chars.len() - 1]),
551 }
552 }
553
554 // Mask local
555 let masked_local = mask_label(local);
556
557 // Mask first domain label only, keep the rest of the domain intact
558 let mut dom_parts = domain_rest.splitn(2, '.');
559 let first_label = dom_parts.next().unwrap_or("");
560 let rest = dom_parts.next();
561 let masked_first = mask_label(first_label);
562 let masked_domain = if let Some(rest) = rest {
563 format!("{}.{rest}", masked_first)
564 } else {
565 masked_first
566 };
567
568 format!("{masked_local}@{masked_domain}")
569}
570
571pub enum VerifyServiceAuthError {
572 AuthFailed,
573 Error(anyhow::Error),
574}
575
576/// Verifies the service auth token that is appended to an XRPC proxy request
577pub async fn verify_service_auth(
578 jwt: &str,
579 lxm: &Nsid<'static>,
580 public_resolver: Arc<PublicResolver>,
581 service_did: &Did<'static>,
582 //The did of the user wanting to create an account
583 requested_did: &Did<'static>,
584) -> Result<(), VerifyServiceAuthError> {
585 let parsed =
586 service_auth::parse_jwt(jwt).map_err(|e| VerifyServiceAuthError::Error(e.into()))?;
587
588 let claims = parsed.claims();
589
590 let did_doc = public_resolver
591 .resolve_did_doc(&requested_did)
592 .await
593 .map_err(|err| {
594 log::error!("Error resolving the service auth for: {}", claims.iss);
595 return VerifyServiceAuthError::Error(err.into());
596 })?;
597
598 // Parse the DID document response to get verification methods
599 let doc = did_doc.parse().map_err(|err| {
600 log::error!("Error parsing the service auth did doc: {}", claims.iss);
601 VerifyServiceAuthError::Error(anyhow::anyhow!(err))
602 })?;
603
604 let verification_methods = doc.verification_method.as_deref().ok_or_else(|| {
605 VerifyServiceAuthError::Error(anyhow::anyhow!(
606 "No verification methods in did doc: {}",
607 &claims.iss
608 ))
609 })?;
610
611 let signing_key = extract_signing_key(verification_methods).ok_or_else(|| {
612 VerifyServiceAuthError::Error(anyhow::anyhow!(
613 "No signing key found in did doc: {}",
614 &claims.iss
615 ))
616 })?;
617
618 service_auth::verify_signature(&parsed, &signing_key).map_err(|err| {
619 log::error!("Error verifying service auth signature: {}", err);
620 VerifyServiceAuthError::AuthFailed
621 })?;
622
623 // Now validate claims (audience, expiration, etc.)
624 claims.validate(service_did).map_err(|e| {
625 log::error!("Error validating service auth claims: {}", e);
626 VerifyServiceAuthError::AuthFailed
627 })?;
628
629 if claims.aud != *service_did {
630 log::error!("Invalid audience (did:web): {}", claims.aud);
631 return Err(VerifyServiceAuthError::AuthFailed);
632 }
633
634 let lxm_from_claims = claims.lxm.as_ref().ok_or_else(|| {
635 VerifyServiceAuthError::Error(anyhow::anyhow!("No lxm claim in service auth JWT"))
636 })?;
637
638 if lxm_from_claims != lxm {
639 return Err(VerifyServiceAuthError::Error(anyhow::anyhow!(
640 "Invalid XRPC endpoint requested"
641 )));
642 }
643 Ok(())
644}
645
646/// Ripped from Jacquard
647///
648/// Extract the signing key from a DID document's verification methods.
649///
650/// This looks for a key with type "atproto" or the first available key
651/// if no atproto-specific key is found.
652fn extract_signing_key(methods: &[VerificationMethod]) -> Option<PublicKey> {
653 // First try to find an atproto-specific key
654 let atproto_method = methods
655 .iter()
656 .find(|m| m.r#type.as_ref() == "Multikey" || m.r#type.as_ref() == "atproto");
657
658 let method = atproto_method.or_else(|| methods.first())?;
659
660 // Parse the multikey
661 let public_key_multibase = method.public_key_multibase.as_ref()?;
662
663 // Decode multibase
664 let (_, key_bytes) = multibase::decode(public_key_multibase.as_ref()).ok()?;
665
666 // First two bytes are the multicodec prefix
667 if key_bytes.len() < 2 {
668 return None;
669 }
670
671 let codec = &key_bytes[..2];
672 let key_material = &key_bytes[2..];
673
674 match codec {
675 // p256-pub (0x1200)
676 [0x80, 0x24] => PublicKey::from_p256_bytes(key_material).ok(),
677 // secp256k1-pub (0xe7)
678 [0xe7, 0x01] => PublicKey::from_k256_bytes(key_material).ok(),
679 _ => None,
680 }
681}
682
683/// Payload for gate JWE tokens
684#[derive(serde::Serialize, serde::Deserialize, Debug)]
685pub struct GateTokenPayload {
686 pub handle: String,
687 pub created_at: String,
688}
689
690/// Generate a secure JWE token for gate verification
691pub fn generate_gate_token(handle: &str, encryption_key: &[u8]) -> Result<String, anyhow::Error> {
692 use josekit::jwe::{JweHeader, alg::direct::DirectJweAlgorithm};
693
694 let payload = GateTokenPayload {
695 handle: handle.to_string(),
696 created_at: Utc::now().to_rfc3339(),
697 };
698
699 let payload_json = serde_json::to_string(&payload)?;
700
701 let mut header = JweHeader::new();
702 header.set_token_type("JWT");
703 header.set_content_encryption("A128CBC-HS256");
704
705 let encrypter = DirectJweAlgorithm::Dir.encrypter_from_bytes(encryption_key)?;
706
707 // Encrypt
708 let jwe = josekit::jwe::serialize_compact(payload_json.as_bytes(), &header, &encrypter)?;
709
710 Ok(jwe)
711}
712
713/// Verify and decrypt a gate JWE token, returning the payload if valid
714pub fn verify_gate_token(
715 token: &str,
716 encryption_key: &[u8],
717) -> Result<GateTokenPayload, anyhow::Error> {
718 let decrypter = DirectJweAlgorithm::Dir.decrypter_from_bytes(encryption_key)?;
719 let (payload_bytes, _header) = josekit::jwe::deserialize_compact(token, &decrypter)?;
720 let payload: GateTokenPayload = serde_json::from_slice(&payload_bytes)?;
721
722 Ok(payload)
723}