use std::sync::Arc;
use affinidi_messaging_didcomm::Message;
use tokio::sync::RwLock;
use affinidi_did_resolver_cache_sdk::DIDCacheClient;
use crate::config::AppConfig;
use crate::didcomm_bridge::DIDCommBridge;
use crate::keys::seed_store::SeedStore;
use crate::messaging::shim::{DIDCommResponse, DIDCommServiceError, HandlerContext, ProblemReport};
#[cfg(feature = "didcomm")]
use crate::messaging::shim::{Extension, ServiceProblemReport};
use crate::server::AppState;
use crate::store::KeyspaceHandle;
#[cfg(feature = "didcomm")]
use super::handlers;
#[cfg(all(feature = "tee", feature = "didcomm"))]
use vta_sdk::protocols::attestation_management;
#[cfg(all(feature = "webvh", feature = "didcomm"))]
use vta_sdk::protocols::did_management;
#[cfg(all(feature = "webvh", feature = "didcomm"))]
use vta_sdk::protocols::protocol_management;
#[cfg(all(feature = "webvh", feature = "didcomm"))]
use vta_sdk::protocols::provision_integration_management;
#[cfg(feature = "didcomm")]
use vta_sdk::protocols::{
self, acl_management, audit_management, context_management, credential_exchange,
key_management, seed_management, vta_management,
};
#[cfg(feature = "didcomm")]
const TRUST_PING_TYPE: &str = "https://didcomm.org/trust-ping/2.0/ping";
#[cfg(feature = "didcomm")]
const TRUST_PONG_TYPE: &str = "https://didcomm.org/trust-ping/2.0/ping-response";
pub(crate) const MESSAGE_PICKUP_STATUS_TYPE: &str = "https://didcomm.org/messagepickup/3.0/status";
#[derive(Clone)]
pub struct VtaState {
pub keys_ks: KeyspaceHandle,
pub acl_ks: KeyspaceHandle,
pub sessions_ks: KeyspaceHandle,
pub contexts_ks: KeyspaceHandle,
pub did_templates_ks: KeyspaceHandle,
pub audit_ks: KeyspaceHandle,
pub audit_sink: vta_audit::SharedAuditSink,
pub imported_ks: KeyspaceHandle,
pub internal_ks: KeyspaceHandle,
pub service_state_ks: KeyspaceHandle,
#[cfg(feature = "webvh")]
pub webvh_ks: KeyspaceHandle,
pub issued_credentials_ks: KeyspaceHandle,
pub sealed_nonces_ks: KeyspaceHandle,
#[cfg(feature = "webvh")]
pub drains_ks: KeyspaceHandle,
#[cfg(feature = "webvh")]
pub snapshot_ks: KeyspaceHandle,
#[cfg(feature = "webvh")]
pub mediator_registry: Arc<crate::messaging::registry::MediatorListenerRegistry>,
#[cfg(feature = "webvh")]
pub drain_sweeper: Arc<crate::messaging::drain_sweeper::DrainSweeper>,
#[cfg(feature = "webvh")]
pub webvh_auth_locks: crate::operations::did_webvh::WebvhAuthLocks,
pub telemetry: vti_common::telemetry::SharedTelemetrySink,
pub seed_store: Arc<dyn SeedStore>,
pub config: Arc<RwLock<AppConfig>>,
pub did_resolver: Option<DIDCacheClient>,
pub didcomm_bridge: Arc<DIDCommBridge>,
#[cfg(feature = "didcomm")]
pub secrets_resolver: Option<Arc<affinidi_tdk::secrets_resolver::ThreadedSecretsResolver>>,
#[cfg(feature = "didcomm")]
pub signing_vm_id: Option<String>,
#[cfg(feature = "didcomm")]
pub ka_vm_id: Option<String>,
#[cfg(feature = "tee")]
pub tee_state: Option<crate::tee::TeeState>,
pub restart_tx: tokio::sync::watch::Sender<bool>,
pub store: vti_common::store::Store,
pub storage_encryption_key: Option<[u8; 32]>,
pub in_enclave: bool,
}
impl VtaState {
pub fn backup_access(&self) -> crate::restore::BackupAccess<'_> {
crate::restore::BackupAccess {
store: &self.store,
storage_key: self.storage_encryption_key,
in_enclave: self.in_enclave,
seed_store: self.seed_store.as_ref(),
config: &self.config,
}
}
}
#[cfg(feature = "webvh")]
impl From<&VtaState> for crate::operations::provision_integration::ProvisionIntegrationDeps {
fn from(state: &VtaState) -> Self {
Self {
keys_ks: state.keys_ks.clone(),
acl_ks: state.acl_ks.clone(),
audit: std::sync::Arc::clone(&state.audit_sink),
contexts_ks: state.contexts_ks.clone(),
did_templates_ks: state.did_templates_ks.clone(),
imported_ks: state.imported_ks.clone(),
webvh_ks: state.webvh_ks.clone(),
sealed_nonces_ks: state.sealed_nonces_ks.clone(),
seed_store: state.seed_store.clone(),
config: state.config.clone(),
did_resolver: state.did_resolver.clone(),
didcomm_bridge: state.didcomm_bridge.clone(),
webvh_auth_locks: state.webvh_auth_locks.clone(),
}
}
}
impl From<&AppState> for VtaState {
fn from(state: &AppState) -> Self {
Self {
keys_ks: state.keys_ks.clone(),
acl_ks: state.acl_ks.clone(),
sessions_ks: state.sessions_ks.clone(),
contexts_ks: state.contexts_ks.clone(),
did_templates_ks: state.did_templates_ks.clone(),
audit_ks: state.audit_ks.clone(),
audit_sink: std::sync::Arc::clone(&state.audit_sink),
imported_ks: state.imported_ks.clone(),
internal_ks: state.internal_ks.clone(),
service_state_ks: state.service_state_ks.clone(),
#[cfg(feature = "webvh")]
webvh_ks: state.webvh_ks.clone(),
issued_credentials_ks: state.issued_credentials_ks.clone(),
sealed_nonces_ks: state.sealed_nonces_ks.clone(),
#[cfg(feature = "webvh")]
drains_ks: state.drains_ks.clone(),
#[cfg(feature = "webvh")]
snapshot_ks: state.snapshot_ks.clone(),
#[cfg(feature = "webvh")]
mediator_registry: Arc::clone(&state.mediator_registry),
#[cfg(feature = "webvh")]
drain_sweeper: Arc::clone(&state.drain_sweeper),
#[cfg(feature = "webvh")]
webvh_auth_locks: state.webvh_auth_locks.clone(),
telemetry: Arc::clone(&state.telemetry),
seed_store: state.seed_store.clone(),
config: Arc::clone(&state.config),
did_resolver: state.did_resolver.clone(),
didcomm_bridge: Arc::clone(&state.didcomm_bridge),
#[cfg(feature = "didcomm")]
secrets_resolver: state.secrets_resolver.clone(),
#[cfg(feature = "didcomm")]
signing_vm_id: state.signing_vm_id.clone(),
#[cfg(feature = "didcomm")]
ka_vm_id: state.ka_vm_id.clone(),
#[cfg(feature = "tee")]
tee_state: state.tee.as_ref().map(|tc| tc.state.clone()),
restart_tx: state.restart_tx.clone(),
store: state.store.clone(),
storage_encryption_key: state.storage_encryption_key,
in_enclave: state.tee.is_some(),
}
}
}
#[cfg(feature = "didcomm")]
type HandlerResult = Result<Option<DIDCommResponse>, DIDCommServiceError>;
#[cfg(feature = "didcomm")]
fn finish(result: HandlerResult) -> Option<DIDCommResponse> {
match result {
Ok(opt) => opt,
Err(e) => Some(DIDCommResponse::problem_report(
ProblemReport::internal_error(e.to_string()),
)),
}
}
#[cfg(feature = "didcomm")]
fn trust_ping_reply(msg: &Message, sender_did: Option<&str>) -> Option<DIDCommResponse> {
#[derive(serde::Deserialize)]
struct PingBody {
#[serde(default = "default_true")]
response_requested: bool,
}
fn default_true() -> bool {
true
}
let body: PingBody = serde_json::from_value(msg.body.clone()).unwrap_or(PingBody {
response_requested: true,
});
if !body.response_requested {
return None;
}
sender_did?;
Some(DIDCommResponse::new(TRUST_PONG_TYPE, serde_json::Value::Null).thid(msg.id.clone()))
}
#[cfg(feature = "didcomm")]
pub async fn dispatch(
msg: Message,
ctx: HandlerContext,
vta_state: Arc<VtaState>,
app_state: AppState,
) -> Option<DIDCommResponse> {
let t = msg.typ.clone();
let t = t.as_str();
if t == MESSAGE_PICKUP_STATUS_TYPE {
return None;
}
if t == TRUST_PING_TYPE {
return trust_ping_reply(&msg, ctx.sender_did.as_deref());
}
if t == trust_tasks_didcomm::ENVELOPE_TYPE {
return finish(handlers::handle_trust_task(ctx, msg, Extension(app_state)).await);
}
if t == key_management::CREATE_KEY {
return finish(handlers::handle_create_key(ctx, msg, Extension(vta_state)).await);
}
if t == key_management::GET_KEY {
return finish(handlers::handle_get_key(ctx, msg, Extension(vta_state)).await);
}
if t == key_management::LIST_KEYS {
return finish(handlers::handle_list_keys(ctx, msg, Extension(vta_state)).await);
}
if t == key_management::RENAME_KEY {
return finish(handlers::handle_rename_key(ctx, msg, Extension(vta_state)).await);
}
if t == key_management::REVOKE_KEY {
return finish(handlers::handle_revoke_key(ctx, msg, Extension(vta_state)).await);
}
if t == key_management::GET_KEY_SECRET {
return finish(handlers::handle_get_key_secret(ctx, msg, Extension(vta_state)).await);
}
if t == key_management::SIGN_REQUEST {
return finish(handlers::handle_sign_request(ctx, msg, Extension(vta_state)).await);
}
if t == seed_management::LIST_SEEDS {
return finish(handlers::handle_list_seeds(ctx, msg, Extension(vta_state)).await);
}
if t == seed_management::ROTATE_SEED {
return finish(handlers::handle_rotate_seed(ctx, msg, Extension(vta_state)).await);
}
if t == context_management::CREATE_CONTEXT {
return finish(handlers::handle_create_context(ctx, msg, Extension(vta_state)).await);
}
if t == context_management::GET_CONTEXT {
return finish(handlers::handle_get_context(ctx, msg, Extension(vta_state)).await);
}
if t == context_management::LIST_CONTEXTS {
return finish(handlers::handle_list_contexts(ctx, msg, Extension(vta_state)).await);
}
if t == context_management::UPDATE_CONTEXT {
return finish(handlers::handle_update_context(ctx, msg, Extension(vta_state)).await);
}
if t == context_management::UPDATE_CONTEXT_DID {
return finish(handlers::handle_update_context_did(ctx, msg, Extension(vta_state)).await);
}
if t == context_management::PREVIEW_DELETE_CONTEXT {
return finish(
handlers::handle_preview_delete_context(ctx, msg, Extension(vta_state)).await,
);
}
if t == context_management::DELETE_CONTEXT {
return finish(handlers::handle_delete_context(ctx, msg, Extension(vta_state)).await);
}
if t == acl_management::CREATE_ACL {
return finish(handlers::handle_create_acl(ctx, msg, Extension(vta_state)).await);
}
if t == acl_management::GET_ACL {
return finish(handlers::handle_get_acl(ctx, msg, Extension(vta_state)).await);
}
if t == acl_management::LIST_ACL {
return finish(handlers::handle_list_acl(ctx, msg, Extension(vta_state)).await);
}
if t == acl_management::CHANGE_ROLE {
return finish(handlers::handle_change_acl_role(ctx, msg, Extension(vta_state)).await);
}
if t == acl_management::UPDATE_ACL {
return finish(handlers::handle_update_acl(ctx, msg, Extension(vta_state)).await);
}
if t == acl_management::DELETE_ACL {
return finish(handlers::handle_delete_acl(ctx, msg, Extension(vta_state)).await);
}
if t == acl_management::SWAP_ACL || t == acl_management::ACL_SWAP_KEY {
return finish(
handlers::handle_swap_acl(ctx, msg, Extension(vta_state), Extension(app_state)).await,
);
}
if t == audit_management::LIST_LOGS {
return finish(handlers::handle_list_logs(ctx, msg, Extension(vta_state)).await);
}
if t == audit_management::GET_RETENTION {
return finish(handlers::handle_get_retention(ctx, msg, Extension(vta_state)).await);
}
if t == audit_management::UPDATE_RETENTION {
return finish(handlers::handle_update_retention(ctx, msg, Extension(vta_state)).await);
}
if t == vta_management::GET_CONFIG {
return finish(handlers::handle_get_config(ctx, msg, Extension(vta_state)).await);
}
if t == vta_management::UPDATE_CONFIG {
return finish(handlers::handle_update_config(ctx, msg, Extension(vta_state)).await);
}
if t == protocols::PROBLEM_REPORT_TYPE {
return finish(handlers::handle_problem_report(ctx, msg).await);
}
if t == vta_management::RESTART {
return finish(handlers::handle_restart(ctx, msg, Extension(vta_state)).await);
}
if t == protocols::backup_management::EXPORT_BACKUP {
return finish(handlers::handle_backup_export(ctx, msg, Extension(vta_state)).await);
}
if t == protocols::backup_management::IMPORT_BACKUP {
return finish(handlers::handle_backup_import(ctx, msg, Extension(vta_state)).await);
}
if t == credential_exchange::ISSUE {
return finish(handlers::handle_credential_issue(ctx, msg, Extension(app_state)).await);
}
if t == credential_exchange::QUERY {
return finish(handlers::handle_credential_query(ctx, msg, Extension(app_state)).await);
}
if t == credential_exchange::OFFER {
return finish(handlers::handle_credential_offer(ctx, msg, Extension(app_state)).await);
}
#[cfg(feature = "webvh")]
{
if t == did_management::CREATE_DID_WEBVH {
return finish(handlers::handle_create_did_webvh(ctx, msg, Extension(vta_state)).await);
}
if t == did_management::GET_DID_WEBVH {
return finish(handlers::handle_get_did_webvh(ctx, msg, Extension(vta_state)).await);
}
if t == did_management::GET_DID_WEBVH_LOG {
return finish(
handlers::handle_get_did_webvh_log(ctx, msg, Extension(vta_state)).await,
);
}
if t == did_management::LIST_DIDS_WEBVH {
return finish(handlers::handle_list_dids_webvh(ctx, msg, Extension(vta_state)).await);
}
if t == did_management::DELETE_DID_WEBVH {
return finish(handlers::handle_delete_did_webvh(ctx, msg, Extension(vta_state)).await);
}
if t == did_management::ADD_WEBVH_SERVER {
return finish(handlers::handle_add_webvh_server(ctx, msg, Extension(vta_state)).await);
}
if t == did_management::LIST_WEBVH_SERVERS {
return finish(
handlers::handle_list_webvh_servers(ctx, msg, Extension(vta_state)).await,
);
}
if t == did_management::LIST_WEBVH_SERVER_DOMAINS {
return finish(
handlers::handle_list_webvh_server_domains(ctx, msg, Extension(vta_state)).await,
);
}
if t == did_management::UPDATE_WEBVH_SERVER {
return finish(
handlers::handle_update_webvh_server(ctx, msg, Extension(vta_state)).await,
);
}
if t == did_management::REMOVE_WEBVH_SERVER {
return finish(
handlers::handle_remove_webvh_server(ctx, msg, Extension(vta_state)).await,
);
}
if t == did_management::UPDATE_DID_WEBVH {
return finish(handlers::handle_update_did_webvh(ctx, msg, Extension(vta_state)).await);
}
if t == did_management::ROTATE_DID_WEBVH_KEYS {
return finish(
handlers::handle_rotate_did_webvh_keys(ctx, msg, Extension(vta_state)).await,
);
}
if t == did_management::REGISTER_DID_WITH_SERVER {
return finish(
handlers::handle_register_did_with_server(ctx, msg, Extension(vta_state)).await,
);
}
}
#[cfg(feature = "webvh")]
{
use super::handlers_protocol as hp;
if t == protocol_management::DISABLE_DIDCOMM {
return finish(hp::handle_disable_didcomm(ctx, msg, Extension(vta_state)).await);
}
if t == protocol_management::ENABLE_REST {
return finish(hp::handle_enable_rest(ctx, msg, Extension(vta_state)).await);
}
if t == protocol_management::UPDATE_REST {
return finish(hp::handle_update_rest(ctx, msg, Extension(vta_state)).await);
}
if t == protocol_management::DISABLE_REST {
return finish(hp::handle_disable_rest(ctx, msg, Extension(vta_state)).await);
}
if t == protocol_management::ROLLBACK_REST {
return finish(hp::handle_rollback_rest(ctx, msg, Extension(vta_state)).await);
}
if t == protocol_management::ENABLE_TSP {
return finish(hp::handle_enable_tsp(ctx, msg, Extension(vta_state)).await);
}
if t == protocol_management::UPDATE_TSP {
return finish(hp::handle_update_tsp(ctx, msg, Extension(vta_state)).await);
}
if t == protocol_management::DISABLE_TSP {
return finish(hp::handle_disable_tsp(ctx, msg, Extension(vta_state)).await);
}
if t == protocol_management::ROLLBACK_TSP {
return finish(hp::handle_rollback_tsp(ctx, msg, Extension(vta_state)).await);
}
if t == protocol_management::UPDATE_DIDCOMM {
return finish(hp::handle_update_didcomm(ctx, msg, Extension(vta_state)).await);
}
if t == protocol_management::ROLLBACK_DIDCOMM {
return finish(hp::handle_rollback_didcomm(ctx, msg, Extension(vta_state)).await);
}
if t == protocol_management::LIST_SERVICES {
return finish(hp::handle_list_services(ctx, msg, Extension(vta_state)).await);
}
if t == protocol_management::LIST_DRAIN {
return finish(hp::handle_list_drain(ctx, msg, Extension(vta_state)).await);
}
if t == protocol_management::DRAIN_CANCEL {
return finish(hp::handle_drain_cancel(ctx, msg, Extension(vta_state)).await);
}
if t == protocol_management::MEDIATOR_REPORT {
return finish(hp::handle_mediator_report(ctx, msg, Extension(vta_state)).await);
}
}
#[cfg(feature = "webvh")]
{
if t == provision_integration_management::ProvisionSpecVersion::CURRENT.request_uri() {
return finish(
handlers::handle_provision_integration(ctx, msg, Extension(vta_state)).await,
);
}
}
if t == handlers::STEP_UP_APPROVE_REQUEST_TYPE
|| t == handlers::STEP_UP_APPROVE_REQUEST_CANONICAL
|| t == handlers::STEP_UP_APPROVE_REQUEST_CANONICAL_0_2
{
return finish(handlers::handle_step_up_approve(ctx, msg, Extension(vta_state)).await);
}
#[cfg(feature = "tee")]
{
if t == attestation_management::GET_TEE_STATUS {
return finish(handlers::handle_tee_status(ctx, msg, Extension(vta_state)).await);
}
if t == attestation_management::REQUEST_ATTESTATION {
return finish(
handlers::handle_request_attestation(ctx, msg, Extension(vta_state)).await,
);
}
}
finish(handlers::handle_unknown(ctx, msg).await)
}
#[cfg(all(test, feature = "didcomm"))]
mod envelope_only_carriage {
use super::*;
use serde_json::json;
use trust_tasks_didcomm::ENVELOPE_TYPE;
const STRANGER: &str = "did:key:z6MkhaXgBZDvotDkL5257faiztiGiC2QtKLGpbnnEGta2doK";
#[cfg(feature = "webvh")]
const LEGACY_TASK_TYPED_ARMS: &[&str] = &[
vta_sdk::protocols::acl_management::ACL_SWAP_KEY,
provision_integration_management::CANONICAL_PROVISION_INTEGRATION_0_3,
];
#[cfg(not(feature = "webvh"))]
const LEGACY_TASK_TYPED_ARMS: &[&str] = &[vta_sdk::protocols::acl_management::ACL_SWAP_KEY];
fn problem_comment(resp: &DIDCommResponse) -> Option<&str> {
(resp.type_ == vta_sdk::protocols::PROBLEM_REPORT_TYPE)
.then(|| resp.body.get("comment").and_then(|c| c.as_str()))
.flatten()
}
#[tokio::test]
async fn every_dispatched_uri_is_served_in_the_envelope_and_refused_typed_as_itself() {
let (app_state, _dir) = crate::test_support::build_signing_test_app_state().await;
let vta_state = Arc::new(VtaState::from(&app_state));
let uris = crate::trust_tasks::dispatched_uris();
assert!(!uris.is_empty(), "the dispatcher serves nothing?");
let mut typed_arm: Vec<&str> = Vec::new();
for uri in uris {
let doc = json!({
"id": format!("urn:uuid:{}", uuid::Uuid::new_v4()),
"type": uri,
"issuer": STRANGER,
"payload": {},
});
let req_id = format!("urn:uuid:{}", uuid::Uuid::new_v4());
let msg = Message::build(req_id.clone(), ENVELOPE_TYPE.to_string(), doc.clone())
.from(STRANGER.to_string())
.finalize();
let ctx = HandlerContext {
sender_did: Some(STRANGER.to_string()),
};
let resp = dispatch(msg, ctx, vta_state.clone(), app_state.clone())
.await
.unwrap_or_else(|| panic!("`{uri}` in the envelope got no reply at all"));
assert!(
!problem_comment(&resp).is_some_and(|c| c.contains("unsupported message type")),
"`{uri}` is dispatched, but the router refused its envelope: {:?}",
resp.body
);
assert_eq!(
resp.type_, ENVELOPE_TYPE,
"`{uri}`: a reply to an enveloped request rides the envelope (binding §5)"
);
let req_id = format!("urn:uuid:{}", uuid::Uuid::new_v4());
let msg = Message::build(req_id.clone(), uri.to_string(), doc)
.from(STRANGER.to_string())
.finalize();
let ctx = HandlerContext {
sender_did: Some(STRANGER.to_string()),
};
let resp = dispatch(msg, ctx, vta_state.clone(), app_state.clone())
.await
.unwrap_or_else(|| panic!("`{uri}` typed as itself got no reply at all"));
let refused = problem_comment(&resp).is_some_and(|c| c.contains(ENVELOPE_TYPE));
if !refused {
typed_arm.push(uri);
continue;
}
assert_eq!(
resp.thid.as_deref(),
Some(req_id.as_str()),
"`{uri}`: unthreaded"
);
}
typed_arm.sort_unstable();
let mut expected = LEGACY_TASK_TYPED_ARMS.to_vec();
expected.sort_unstable();
assert_eq!(
typed_arm, expected,
"the served URIs the router still answers typed as the task changed. A new one \
is a regression — carry it in the envelope instead; a removed one should be \
deleted from `LEGACY_TASK_TYPED_ARMS`"
);
}
#[test]
fn a_non_trust_task_type_is_not_told_about_the_envelope() {
assert!(
handlers::trust_task_needs_envelope("https://example.com/protocols/x/1.0/y").is_none()
);
}
}