1use crate::registry::rate_limit::{NoopRateLimiter, RateLimiter};
33use crate::registry::store::RegistryStore;
34use crate::registry::validator::{
35 check_revocation_supersession, key_revocation_gate_applies, PublishValidator,
36};
37use acdp_primitives::error::AcdpError;
38use acdp_types::{
39 body::{Body, FullContext},
40 capabilities::CapabilitiesDocument,
41 primitives::{AgentDid, CtxId, LineageId, Status, Visibility},
42 publish::{PublishRequest, PublishResponse},
43 revocation::KeyRevocation,
44 search::{SearchParams, SearchResponse},
45};
46
47pub struct RegistryServer<S: RegistryStore, L: RateLimiter = NoopRateLimiter> {
53 store: S,
54 caps: CapabilitiesDocument,
55 authority: String,
56 rate_limiter: L,
57 receipt_signer: Option<acdp_types::receipt::ReceiptSigner>,
62 mint_head_receipts: bool,
69 lifecycle_enabled: bool,
76}
77
78impl<S: RegistryStore> RegistryServer<S, NoopRateLimiter> {
79 #[doc(hidden)]
83 pub fn new(store: S, caps: CapabilitiesDocument, authority: impl Into<String>) -> Self {
84 Self {
85 store,
86 caps,
87 authority: authority.into(),
88 rate_limiter: NoopRateLimiter,
89 receipt_signer: None,
90 mint_head_receipts: false,
91 lifecycle_enabled: false,
92 }
93 }
94
95 pub fn try_new(
109 store: S,
110 caps: CapabilitiesDocument,
111 authority: impl Into<String>,
112 ) -> Result<Self, AcdpError> {
113 let authority = authority.into();
114 if !acdp_types::primitives::is_valid_dns_authority(&authority) {
117 return Err(AcdpError::SchemaViolation(format!(
118 "registry authority '{authority}' is not a valid DNS hostname \
119 (must be lowercase labels, e.g. 'registry.example.com'); \
120 use RegistryServer::try_new_for_test_authority for host:port test setups"
121 )));
122 }
123 acdp_validation::validate_capabilities(&caps)?;
124 let expected_did = acdp_did::authority_to_did_web(&authority);
127 if caps.registry_did != expected_did {
128 return Err(AcdpError::SchemaViolation(format!(
129 "capabilities.registry_did '{}' does not match expected '{expected_did}' \
130 for authority '{authority}'",
131 caps.registry_did
132 )));
133 }
134 Ok(Self {
135 store,
136 caps,
137 authority,
138 rate_limiter: NoopRateLimiter,
139 receipt_signer: None,
140 mint_head_receipts: false,
141 lifecycle_enabled: false,
142 })
143 }
144
145 #[doc(hidden)]
154 pub fn try_new_for_test_authority(
155 store: S,
156 caps: CapabilitiesDocument,
157 authority: impl Into<String>,
158 ) -> Result<Self, AcdpError> {
159 let authority = authority.into();
160 acdp_validation::validate_capabilities(&caps)?;
161 let expected_did = acdp_did::authority_to_did_web(&authority);
162 if caps.registry_did != expected_did {
163 return Err(AcdpError::SchemaViolation(format!(
164 "capabilities.registry_did '{}' does not match expected '{expected_did}' \
165 for authority '{authority}'",
166 caps.registry_did
167 )));
168 }
169 Ok(Self {
170 store,
171 caps,
172 authority,
173 rate_limiter: NoopRateLimiter,
174 receipt_signer: None,
175 mint_head_receipts: false,
176 lifecycle_enabled: false,
177 })
178 }
179}
180
181impl<S: RegistryStore, L: RateLimiter> RegistryServer<S, L> {
182 pub fn with_rate_limiter<L2: RateLimiter>(self, limiter: L2) -> RegistryServer<S, L2> {
184 RegistryServer {
185 store: self.store,
186 caps: self.caps,
187 authority: self.authority,
188 rate_limiter: limiter,
189 receipt_signer: self.receipt_signer,
190 mint_head_receipts: self.mint_head_receipts,
191 lifecycle_enabled: self.lifecycle_enabled,
192 }
193 }
194
195 pub fn with_receipt_signer(
212 mut self,
213 signer: acdp_types::receipt::ReceiptSigner,
214 ) -> Result<Self, AcdpError> {
215 if signer.registry_did() != self.caps.registry_did {
216 return Err(AcdpError::SchemaViolation(format!(
217 "receipt signer registry_did '{}' ≠ capabilities.registry_did '{}'",
218 signer.registry_did(),
219 self.caps.registry_did
220 )));
221 }
222 self.require_min_acdp_version((0, 2, 0), "acdp-registry-receipts")?;
225 let profile = acdp_types::profile::Profile::RegistryReceipts.as_str();
226 if !self.caps.profiles.iter().any(|p| p == profile) {
227 self.caps.profiles.push(profile.to_string());
228 }
229 self.receipt_signer = Some(signer);
230 Ok(self)
231 }
232
233 pub fn with_lineage_head_receipts(mut self) -> Result<Self, AcdpError> {
247 if self.receipt_signer.is_none() {
248 return Err(AcdpError::SchemaViolation(
249 "acdp-registry-head-receipts requires the acdp-registry-receipts profile \
250 (RFC-ACDP-0011 §9): call with_receipt_signer first"
251 .into(),
252 ));
253 }
254 self.require_min_acdp_version((0, 3, 0), "acdp-registry-head-receipts")?;
255 let profile = acdp_types::profile::Profile::RegistryHeadReceipts.as_str();
256 if !self.caps.profiles.iter().any(|p| p == profile) {
257 self.caps.profiles.push(profile.to_string());
258 }
259 self.mint_head_receipts = true;
260 Ok(self)
261 }
262
263 pub fn with_lifecycle(mut self) -> Result<Self, AcdpError> {
278 self.require_min_acdp_version((0, 3, 0), "acdp-registry-lifecycle")?;
279 let profile = acdp_types::profile::Profile::RegistryLifecycle.as_str();
280 if !self.caps.profiles.iter().any(|p| p == profile) {
281 self.caps.profiles.push(profile.to_string());
282 }
283 self.lifecycle_enabled = true;
284 Ok(self)
285 }
286
287 fn require_min_acdp_version(&self, min: (u64, u64, u64), what: &str) -> Result<(), AcdpError> {
292 let parts: Vec<u64> = self
293 .caps
294 .acdp_version
295 .split('.')
296 .map(|p| p.parse::<u64>())
297 .collect::<Result<_, _>>()
298 .map_err(|_| {
299 AcdpError::SchemaViolation(format!(
300 "capabilities.acdp_version '{}' is not a plain MAJOR.MINOR.PATCH version",
301 self.caps.acdp_version
302 ))
303 })?;
304 let [major, minor, patch] = parts.as_slice() else {
305 return Err(AcdpError::SchemaViolation(format!(
306 "capabilities.acdp_version '{}' is not a plain MAJOR.MINOR.PATCH version",
307 self.caps.acdp_version
308 )));
309 };
310 if (*major, *minor, *patch) < min {
311 return Err(AcdpError::SchemaViolation(format!(
312 "{what} requires capabilities.acdp_version >= {}.{}.{}, got '{}'",
313 min.0, min.1, min.2, self.caps.acdp_version
314 )));
315 }
316 Ok(())
317 }
318
319 pub fn store(&self) -> &S {
322 &self.store
323 }
324
325 pub fn capabilities(&self) -> &CapabilitiesDocument {
327 &self.caps
328 }
329
330 #[cfg(feature = "client")]
346 #[cfg_attr(
347 feature = "tracing",
348 tracing::instrument(
349 name = "acdp.publish_verified",
350 skip_all,
351 fields(
352 agent_id = req.agent_id.as_str(),
353 version = req.version,
354 idempotency_key = idempotency_key.is_some(),
355 ),
356 err(Display)
357 )
358 )]
359 pub async fn publish_verified(
360 &self,
361 req: &PublishRequest,
362 idempotency_key: Option<&str>,
363 resolver: &acdp_did::WebResolver,
364 ) -> Result<PublishResponse, AcdpError> {
365 self.publish_verified_in_tenant(req, idempotency_key, resolver, None)
366 .await
367 }
368
369 #[cfg(feature = "client")]
375 pub async fn publish_verified_in_tenant(
376 &self,
377 req: &PublishRequest,
378 idempotency_key: Option<&str>,
379 resolver: &acdp_did::WebResolver,
380 tenant: Option<&str>,
381 ) -> Result<PublishResponse, AcdpError> {
382 self.check_publish_rate_limit(&req.agent_id)?;
384
385 let raw_bytes = serde_json::to_vec(req)?.len();
386 let validator = PublishValidator::for_authority(&self.caps, &self.authority);
387 let _validated = validator.validate_post_schema(req, raw_bytes)?;
388
389 acdp_verify::verify_publish_request_signature(req, resolver).await?;
391
392 let revocation_check_needed = req.context_type.is_key_revocation()
411 && key_revocation_gate_applies(&self.caps.acdp_version);
412
413 let fingerprint = if self.receipt_signer.is_some() || revocation_check_needed {
419 Some(producer_key_fingerprint(req, resolver).await?)
420 } else {
421 None
422 };
423
424 if revocation_check_needed {
425 let fp = fingerprint.as_deref().ok_or_else(|| {
434 AcdpError::RegistryInternal(
435 "key-revocation fingerprint missing despite revocation_check_needed \
436 — this is an internal invariant violation, not a caller error"
437 .into(),
438 )
439 })?;
440 KeyRevocation::from_publish_request(req)?.check_not_self_signed(fp)?;
441 }
442
443 self.commit_via_store(req, idempotency_key, tenant, fingerprint)
451 }
452
453 pub fn publish_verified_did_key(
467 &self,
468 req: &PublishRequest,
469 idempotency_key: Option<&str>,
470 ) -> Result<PublishResponse, AcdpError> {
471 self.publish_verified_did_key_in_tenant(req, idempotency_key, None)
472 }
473
474 #[cfg_attr(
480 feature = "tracing",
481 tracing::instrument(
482 name = "acdp.publish_verified_did_key",
483 skip_all,
484 fields(
485 agent_id = req.agent_id.as_str(),
486 version = req.version,
487 idempotency_key = idempotency_key.is_some(),
488 ),
489 err(Display)
490 )
491 )]
492 pub fn publish_verified_did_key_in_tenant(
493 &self,
494 req: &PublishRequest,
495 idempotency_key: Option<&str>,
496 tenant: Option<&str>,
497 ) -> Result<PublishResponse, AcdpError> {
498 self.check_publish_rate_limit(&req.agent_id)?;
499
500 let raw_bytes = serde_json::to_vec(req)?.len();
501 let validator = PublishValidator::for_authority(&self.caps, &self.authority);
502 let _validated = validator.validate_post_schema(req, raw_bytes)?;
503
504 acdp_verify::verify_publish_request_signature_offline(req)?;
506
507 let fingerprint = if self.receipt_signer.is_some() {
510 let material = acdp_did::key::resolve_did_key(req.agent_id.as_str())?;
511 Some(acdp_crypto::fingerprint::fingerprint_did_key_material(
512 &material,
513 )?)
514 } else {
515 None
516 };
517
518 self.commit_via_store(req, idempotency_key, tenant, fingerprint)
519 }
520
521 #[doc(hidden)]
532 pub fn publish_unverified_for_tests(
533 &self,
534 req: &PublishRequest,
535 ) -> Result<PublishResponse, AcdpError> {
536 self.publish_unverified_in_tenant_for_tests(req, None, None)
537 }
538
539 #[doc(hidden)]
556 pub fn publish_unverified_in_tenant_for_tests(
557 &self,
558 req: &PublishRequest,
559 idempotency_key: Option<&str>,
560 tenant: Option<&str>,
561 ) -> Result<PublishResponse, AcdpError> {
562 self.check_publish_rate_limit(&req.agent_id)?;
566
567 if self.receipt_signer.is_some() {
574 return Err(AcdpError::SchemaViolation(
575 "publish_unverified_for_tests / publish_unverified_in_tenant_for_tests are \
576 unavailable on a receipts-advertising registry (RFC-ACDP-0010 §7: no \
577 degraded mode); use publish_verified or publish_verified_did_key"
578 .into(),
579 ));
580 }
581 let raw_bytes = serde_json::to_vec(req)?.len();
591 let validator = PublishValidator::for_authority(&self.caps, &self.authority);
592 let _validated = validator.validate_post_schema(req, raw_bytes)?;
593 self.commit_via_store(req, idempotency_key, tenant, None)
594 }
595
596 #[doc(hidden)]
618 pub fn publish_pinned_verified_in_tenant(
619 &self,
620 req: &PublishRequest,
621 idempotency_key: Option<&str>,
622 tenant: Option<&str>,
623 verified_public_key_b64: &str,
624 verified_algorithm: &str,
625 ) -> Result<PublishResponse, AcdpError> {
626 self.check_publish_rate_limit(&req.agent_id)?;
627
628 let raw_bytes = serde_json::to_vec(req)?.len();
629 let validator = PublishValidator::for_authority(&self.caps, &self.authority);
630 let _validated = validator.validate_post_schema(req, raw_bytes)?;
631
632 let revocation_check_needed = req.context_type.is_key_revocation()
640 && key_revocation_gate_applies(&self.caps.acdp_version);
641
642 let fingerprint = if self.receipt_signer.is_some() || revocation_check_needed {
643 Some(fingerprint_pinned_key(
644 verified_public_key_b64,
645 verified_algorithm,
646 )?)
647 } else {
648 None
649 };
650
651 if revocation_check_needed {
652 let fp = fingerprint.as_deref().ok_or_else(|| {
659 AcdpError::RegistryInternal(
660 "key-revocation fingerprint missing despite revocation_check_needed \
661 — this is an internal invariant violation, not a caller error"
662 .into(),
663 )
664 })?;
665 KeyRevocation::from_publish_request(req)?.check_not_self_signed(fp)?;
666 }
667
668 self.commit_via_store(req, idempotency_key, tenant, fingerprint)
669 }
670
671 fn check_publish_rate_limit(
675 &self,
676 agent_id: &acdp_types::primitives::AgentDid,
677 ) -> Result<(), AcdpError> {
678 match self.rate_limiter.check_publish(agent_id) {
679 Ok(()) => Ok(()),
680 Err(e) => {
681 #[cfg(feature = "tracing")]
682 tracing::warn!(
683 agent_id = agent_id.as_str(),
684 "publish rejected by rate limiter"
685 );
686 Err(e)
687 }
688 }
689 }
690
691 fn commit_via_store(
696 &self,
697 req: &PublishRequest,
698 idempotency_key: Option<&str>,
699 tenant: Option<&str>,
700 producer_key_fingerprint: Option<String>,
701 ) -> Result<PublishResponse, AcdpError> {
702 let idempotency = if self.caps.supports_idempotency_key {
703 idempotency_key.map(|key| crate::registry::store::PendingIdempotencyCommit {
704 key,
705 ttl: chrono::Duration::seconds(
706 self.caps
707 .limits
708 .idempotency_key_ttl_seconds
709 .unwrap_or(86_400) as i64,
710 ),
711 })
712 } else {
713 None
714 };
715 #[allow(clippy::type_complexity)]
718 let minter: Option<
719 Box<dyn Fn(&Body) -> Result<serde_json::Value, AcdpError> + Send + Sync>,
720 > = match (&self.receipt_signer, producer_key_fingerprint) {
721 (Some(signer), Some(fp)) => Some(Box::new(move |body: &Body| {
722 let receipt = signer.mint(
723 &body.ctx_id,
724 &body.lineage_id,
725 &body.origin_registry,
726 body.created_at,
727 &body.content_hash,
728 &fp,
729 )?;
730 serde_json::to_value(receipt).map_err(AcdpError::from)
731 })),
732 _ => None,
733 };
734 let minted_expected = minter.is_some();
735
736 let admission_closure =
751 if key_revocation_gate_applies(&self.caps.acdp_version) && req.supersedes.is_some() {
752 Some(move |prev: &Body| check_revocation_supersession(prev, req))
753 } else {
754 None
755 };
756 #[allow(clippy::type_complexity)]
757 let predecessor_admission: Option<
758 &(dyn Fn(&Body) -> Result<(), AcdpError> + Send + Sync),
759 > = admission_closure
760 .as_ref()
761 .map(|f| f as &(dyn Fn(&Body) -> Result<(), AcdpError> + Send + Sync));
762
763 let outcome = self
764 .store
765 .commit_publish(crate::registry::store::PublishCommit {
766 req,
767 authority: &self.authority,
768 idempotency,
769 tenant,
770 receipt_minter: minter.as_deref(),
771 predecessor_admission,
772 })?;
773 let (response, replayed) = match outcome {
774 crate::registry::store::PublishCommitOutcome::Inserted(r) => (r, false),
775 crate::registry::store::PublishCommitOutcome::IdempotentReplay(r) => (r, true),
776 };
777 #[cfg(feature = "tracing")]
778 tracing::debug!(
779 ctx_id = %response.ctx_id.0,
780 lineage_id = %response.lineage_id.0,
781 version = response.version,
782 replayed,
783 "publish committed"
784 );
785 if minted_expected && !replayed && response.registry_receipt.is_none() {
799 return Err(AcdpError::RegistryInternal(
800 "receipt signer is configured but the store returned no receipt — \
801 the RegistryStore implementation must invoke PublishCommit::receipt_minter \
802 inside its commit (RFC-ACDP-0010 §7: no degraded mode)"
803 .into(),
804 ));
805 }
806 Ok(response)
807 }
808
809 pub fn retrieve(
822 &self,
823 ctx_id: &CtxId,
824 requester: Option<&AgentDid>,
825 ) -> Result<Option<FullContext>, AcdpError> {
826 let Some(ctx) = self.store.get(ctx_id)? else {
827 return Ok(None);
828 };
829 if !can_retrieve(&ctx.body, requester, &self.caps) {
830 return Ok(None);
831 }
832 Ok(Some(ctx))
833 }
834
835 pub fn retrieve_body(
837 &self,
838 ctx_id: &CtxId,
839 requester: Option<&AgentDid>,
840 ) -> Result<Option<Body>, AcdpError> {
841 Ok(self.retrieve(ctx_id, requester)?.map(|c| c.body))
842 }
843
844 pub fn lineage(
851 &self,
852 lineage_id: &LineageId,
853 requester: Option<&AgentDid>,
854 ) -> Result<Vec<FullContext>, AcdpError> {
855 let all = self.store.lineage(lineage_id)?;
856 Ok(all
857 .into_iter()
858 .filter(|ctx| can_retrieve(&ctx.body, requester, &self.caps))
859 .collect())
860 }
861
862 pub fn current(
882 &self,
883 lineage_id: &LineageId,
884 requester: Option<&AgentDid>,
885 ) -> Result<Option<FullContext>, AcdpError> {
886 let all = self.store.lineage(lineage_id)?;
887 for mut ctx in all.into_iter().rev() {
893 if !matches!(
894 ctx.registry_state.status,
895 Status::Superseded | Status::Retracted
896 ) && can_retrieve(&ctx.body, requester, &self.caps)
897 {
898 if self.mint_head_receipts {
899 let signer = self.receipt_signer.as_ref().ok_or_else(|| {
904 AcdpError::RegistryInternal(
905 "head-receipt minting enabled without a receipt signer \
906 (RFC-ACDP-0011 §9 prerequisite violated)"
907 .into(),
908 )
909 })?;
910 let receipt = signer.mint_lineage_head(
911 lineage_id,
912 &ctx.body.ctx_id,
913 ctx.body.version,
914 &ctx.registry_state.status,
915 chrono::Utc::now(),
916 )?;
917 ctx.lineage_head_receipt = Some(serde_json::to_value(receipt)?);
918 }
919 return Ok(Some(ctx));
920 }
921 }
922 Ok(None)
923 }
924
925 pub fn search(
938 &self,
939 params: &SearchParams,
940 requester: Option<&AgentDid>,
941 ) -> Result<SearchResponse, AcdpError> {
942 if requester.is_none() && !self.caps.anonymous_public_reads {
947 return Err(AcdpError::NotAuthorized(
948 "anonymous search requires authentication \
949 (registry caps: anonymous_public_reads=false)"
950 .into(),
951 ));
952 }
953 self.store
958 .search(params, requester, self.caps.anonymous_public_reads)
959 }
960
961 fn lifecycle_precheck(
993 &self,
994 event: &acdp_types::lifecycle::LifecycleEvent,
995 expected_type: &acdp_types::lifecycle::LifecycleEventType,
996 requester: Option<&AgentDid>,
997 ) -> Result<FullContext, AcdpError> {
998 if !self.lifecycle_enabled {
999 return Err(AcdpError::NotImplemented(
1000 "this registry does not advertise acdp-registry-lifecycle \
1001 (RFC-ACDP-0013 §6: lifecycle endpoints are not implemented)"
1002 .into(),
1003 ));
1004 }
1005 let ctx = self
1007 .retrieve(&event.ctx_id, requester)?
1008 .ok_or_else(|| AcdpError::NotFound(format!("context '{}' not found", event.ctx_id)))?;
1009 event.validate()?;
1011 if &event.event_type != expected_type {
1012 return Err(AcdpError::SchemaViolation(format!(
1013 "event_type '{}' does not match this endpoint (expected '{}', \
1014 RFC-ACDP-0013 §6 step 2)",
1015 event.event_type, expected_type
1016 )));
1017 }
1018 let now = chrono::Utc::now();
1019 if event.occurred_at > now + chrono::Duration::seconds(120) {
1020 return Err(AcdpError::SchemaViolation(format!(
1021 "event occurred_at '{}' is in the future beyond the 120s skew allowance \
1022 (RFC-ACDP-0013 §4)",
1023 event.occurred_at.format("%Y-%m-%dT%H:%M:%S%.3fZ")
1024 )));
1025 }
1026 if event.actor != ctx.body.agent_id {
1028 return Err(AcdpError::NotAuthorized(format!(
1029 "event actor '{}' is not the context's producer — only the producer \
1030 (agent_id) may use the lifecycle endpoints (RFC-ACDP-0013 §6 step 3)",
1031 event.actor
1032 )));
1033 }
1034 event.actor_bound_signature()?;
1038 Ok(ctx)
1039 }
1040
1041 fn lifecycle_commit(
1045 &self,
1046 event: &acdp_types::lifecycle::LifecycleEvent,
1047 ) -> Result<FullContext, AcdpError> {
1048 Ok(self.store.commit_lifecycle_event(event)?.into_context())
1049 }
1050
1051 #[cfg(feature = "client")]
1053 async fn lifecycle_transition_verified(
1054 &self,
1055 event: &acdp_types::lifecycle::LifecycleEvent,
1056 expected_type: acdp_types::lifecycle::LifecycleEventType,
1057 requester: Option<&AgentDid>,
1058 resolver: &acdp_did::WebResolver,
1059 ) -> Result<FullContext, AcdpError> {
1060 let ctx = self.lifecycle_precheck(event, &expected_type, requester)?;
1061 acdp_verify::verify_lifecycle_event(
1065 &serde_json::to_value(event)?,
1066 &event.ctx_id,
1067 &ctx.body.agent_id,
1068 None, resolver,
1070 )
1071 .await?;
1072 self.lifecycle_commit(event)
1073 }
1074
1075 fn lifecycle_transition_verified_did_key(
1077 &self,
1078 event: &acdp_types::lifecycle::LifecycleEvent,
1079 expected_type: acdp_types::lifecycle::LifecycleEventType,
1080 requester: Option<&AgentDid>,
1081 ) -> Result<FullContext, AcdpError> {
1082 let ctx = self.lifecycle_precheck(event, &expected_type, requester)?;
1083 acdp_verify::verify_lifecycle_event_offline(
1084 &serde_json::to_value(event)?,
1085 &event.ctx_id,
1086 &ctx.body.agent_id,
1087 None,
1088 )?;
1089 self.lifecycle_commit(event)
1090 }
1091
1092 #[cfg(feature = "client")]
1105 pub async fn retract_verified(
1106 &self,
1107 event: &acdp_types::lifecycle::LifecycleEvent,
1108 requester: Option<&AgentDid>,
1109 resolver: &acdp_did::WebResolver,
1110 ) -> Result<FullContext, AcdpError> {
1111 self.lifecycle_transition_verified(
1112 event,
1113 acdp_types::lifecycle::LifecycleEventType::Retracted,
1114 requester,
1115 resolver,
1116 )
1117 .await
1118 }
1119
1120 #[cfg(feature = "client")]
1127 pub async fn republish_verified(
1128 &self,
1129 event: &acdp_types::lifecycle::LifecycleEvent,
1130 requester: Option<&AgentDid>,
1131 resolver: &acdp_did::WebResolver,
1132 ) -> Result<FullContext, AcdpError> {
1133 self.lifecycle_transition_verified(
1134 event,
1135 acdp_types::lifecycle::LifecycleEventType::Republished,
1136 requester,
1137 resolver,
1138 )
1139 .await
1140 }
1141
1142 pub fn retract_verified_did_key(
1147 &self,
1148 event: &acdp_types::lifecycle::LifecycleEvent,
1149 requester: Option<&AgentDid>,
1150 ) -> Result<FullContext, AcdpError> {
1151 self.lifecycle_transition_verified_did_key(
1152 event,
1153 acdp_types::lifecycle::LifecycleEventType::Retracted,
1154 requester,
1155 )
1156 }
1157
1158 pub fn republish_verified_did_key(
1160 &self,
1161 event: &acdp_types::lifecycle::LifecycleEvent,
1162 requester: Option<&AgentDid>,
1163 ) -> Result<FullContext, AcdpError> {
1164 self.lifecycle_transition_verified_did_key(
1165 event,
1166 acdp_types::lifecycle::LifecycleEventType::Republished,
1167 requester,
1168 )
1169 }
1170
1171 #[doc(hidden)]
1176 pub fn retract_unverified_for_tests(
1177 &self,
1178 event: &acdp_types::lifecycle::LifecycleEvent,
1179 requester: Option<&AgentDid>,
1180 ) -> Result<FullContext, AcdpError> {
1181 self.lifecycle_precheck(
1182 event,
1183 &acdp_types::lifecycle::LifecycleEventType::Retracted,
1184 requester,
1185 )?;
1186 self.lifecycle_commit(event)
1187 }
1188
1189 #[doc(hidden)]
1191 pub fn republish_unverified_for_tests(
1192 &self,
1193 event: &acdp_types::lifecycle::LifecycleEvent,
1194 requester: Option<&AgentDid>,
1195 ) -> Result<FullContext, AcdpError> {
1196 self.lifecycle_precheck(
1197 event,
1198 &acdp_types::lifecycle::LifecycleEventType::Republished,
1199 requester,
1200 )?;
1201 self.lifecycle_commit(event)
1202 }
1203
1204 pub fn record_registry_lifecycle_event(
1216 &self,
1217 event: &acdp_types::lifecycle::LifecycleEvent,
1218 ) -> Result<FullContext, AcdpError> {
1219 if !self.lifecycle_enabled {
1220 return Err(AcdpError::NotImplemented(
1221 "this registry does not advertise acdp-registry-lifecycle \
1222 (RFC-ACDP-0013 §6)"
1223 .into(),
1224 ));
1225 }
1226 event.validate()?;
1227 if !event.event_type.is_registered() {
1228 return Err(AcdpError::SchemaViolation(format!(
1229 "event_type '{}' is not registered for acceptance in 0.3.0 \
1230 (RFC-ACDP-0013 §7.3)",
1231 event.event_type
1232 )));
1233 }
1234 if event.actor.as_str() != self.caps.registry_did {
1235 return Err(AcdpError::NotAuthorized(format!(
1236 "registry-initiated event actor '{}' ≠ this registry's DID '{}' \
1237 (RFC-ACDP-0013 §6)",
1238 event.actor, self.caps.registry_did
1239 )));
1240 }
1241 if self.receipt_signer.is_some() && !event.is_signed() {
1242 return Err(AcdpError::SchemaViolation(
1243 "a registry advertising acdp-registry-receipts MUST sign its lifecycle \
1244 events (RFC-ACDP-0013 §5)"
1245 .into(),
1246 ));
1247 }
1248 if event.is_signed() {
1249 event.actor_bound_signature()?;
1251 }
1252 self.lifecycle_commit(event)
1253 }
1254}
1255
1256pub(crate) fn can_retrieve(
1258 body: &Body,
1259 requester: Option<&AgentDid>,
1260 caps: &CapabilitiesDocument,
1261) -> bool {
1262 match body.visibility {
1263 Visibility::Public => caps.anonymous_public_reads || requester.is_some(),
1264 Visibility::Restricted | Visibility::Private => match requester {
1265 None => false,
1266 Some(r) => {
1267 r == &body.agent_id
1268 || body
1269 .audience
1270 .as_deref()
1271 .is_some_and(|a| a.iter().any(|d| d == r))
1272 }
1273 },
1274 }
1275}
1276
1277#[cfg(feature = "client")]
1292async fn producer_key_fingerprint(
1293 req: &PublishRequest,
1294 resolver: &acdp_did::WebResolver,
1295) -> Result<String, AcdpError> {
1296 acdp_crypto::fingerprint::fingerprint_for_key_id(
1297 &req.signature.key_id,
1298 &req.signature.algorithm,
1299 resolver,
1300 )
1301 .await
1302}
1303
1304fn fingerprint_pinned_key(public_key_b64: &str, algorithm: &str) -> Result<String, AcdpError> {
1309 use base64::{engine::general_purpose::STANDARD, Engine};
1310
1311 let raw = STANDARD
1312 .decode(public_key_b64)
1313 .map_err(|e| AcdpError::KeyResolution(format!("pinned key is not valid base64: {e}")))?;
1314 match algorithm {
1315 "ed25519" => {
1316 let arr: [u8; 32] = raw.as_slice().try_into().map_err(|_| {
1317 AcdpError::KeyResolution(format!(
1318 "pinned ed25519 key must be 32 bytes, got {}",
1319 raw.len()
1320 ))
1321 })?;
1322 Ok(acdp_crypto::fingerprint::fingerprint_ed25519(&arr))
1323 }
1324 "ecdsa-p256" => acdp_crypto::fingerprint::fingerprint_p256_sec1(&raw),
1325 other => Err(AcdpError::UnsupportedAlgorithm(format!(
1326 "cannot fingerprint a pinned key for algorithm '{other}'"
1327 ))),
1328 }
1329}
1330
1331#[cfg(test)]
1332mod tests {
1333 use super::*;
1334 use crate::registry::store::InMemoryStore;
1335 use acdp_crypto::SigningKey;
1336 use acdp_producer::Producer;
1337 use acdp_types::capabilities::Limits;
1338 use acdp_types::primitives::{AgentDid, ContextType, Visibility};
1339
1340 fn caps() -> CapabilitiesDocument {
1341 CapabilitiesDocument {
1342 acdp_version: "0.1.0".into(),
1343 registry_did: "did:web:registry.example.com".into(),
1344 supported_signature_algorithms: vec!["ed25519".into()],
1345 supported_did_methods: vec!["did:web".into()],
1346 profiles: vec!["acdp-registry-core".into()],
1347 limits: Limits {
1348 max_payload_bytes: 1_048_576,
1349 max_embedded_bytes: 65_536,
1350 idempotency_key_ttl_seconds: None,
1351 max_publish_per_minute: None,
1352 },
1353 read_authentication_methods: vec![],
1354 anonymous_public_reads: true,
1355 supports_idempotency_key: false,
1356 extensions: Default::default(),
1357 }
1358 }
1359
1360 fn producer() -> Producer {
1361 Producer::new(
1362 SigningKey::from_bytes(&[1u8; 32]),
1363 AgentDid::new("did:web:agents.example.com:test"),
1364 "did:web:agents.example.com:test#key-1",
1365 )
1366 }
1367
1368 #[test]
1369 fn publish_v1_then_retrieve() {
1370 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1371 let p = producer();
1372 let req = p
1373 .publish_request()
1374 .title("v1")
1375 .context_type(ContextType::DataSnapshot)
1376 .visibility(Visibility::Public)
1377 .build()
1378 .unwrap();
1379 let resp = server.publish_unverified_for_tests(&req).unwrap();
1380 assert_eq!(resp.version, 1);
1381 let ctx = server.retrieve(&resp.ctx_id, None).unwrap().unwrap();
1382 assert_eq!(ctx.body.title, "v1");
1383 let lineage = server.lineage(&resp.lineage_id, None).unwrap();
1385 assert_eq!(lineage.len(), 1);
1386 let cur = server.current(&resp.lineage_id, None).unwrap().unwrap();
1388 assert_eq!(cur.body.ctx_id, resp.ctx_id);
1389 }
1390
1391 #[test]
1392 fn supersession_marks_predecessor_and_returns_v2() {
1393 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1394 let p = producer();
1395 let v1_req = p
1396 .publish_request()
1397 .title("v1")
1398 .context_type(ContextType::DataSnapshot)
1399 .visibility(Visibility::Public)
1400 .build()
1401 .unwrap();
1402 let v1 = server.publish_unverified_for_tests(&v1_req).unwrap();
1403
1404 let v2_req = p
1405 .supersede(v1.ctx_id.clone())
1406 .version(2)
1407 .title("v2")
1408 .context_type(ContextType::DataSnapshot)
1409 .visibility(Visibility::Public)
1410 .build()
1411 .unwrap();
1412 let v2 = server.publish_unverified_for_tests(&v2_req).unwrap();
1413 assert_eq!(v2.version, 2);
1414 let v1_ctx = server.retrieve(&v1.ctx_id, None).unwrap().unwrap();
1416 assert!(matches!(
1417 v1_ctx.registry_state.status,
1418 acdp_types::Status::Superseded
1419 ));
1420 assert_eq!(v1.lineage_id, v2.lineage_id);
1422 let cur = server.current(&v1.lineage_id, None).unwrap().unwrap();
1424 assert_eq!(cur.body.ctx_id, v2.ctx_id);
1425 }
1426
1427 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1435 async fn concurrent_supersession_exactly_one_succeeds() {
1436 use std::sync::Arc;
1437 let server = Arc::new(RegistryServer::new(
1438 InMemoryStore::new(),
1439 caps(),
1440 "registry.example.com",
1441 ));
1442 let p = producer();
1443 let v1_req = p
1444 .publish_request()
1445 .title("v1")
1446 .context_type(ContextType::DataSnapshot)
1447 .visibility(Visibility::Public)
1448 .build()
1449 .unwrap();
1450 let v1 = server.publish_unverified_for_tests(&v1_req).unwrap();
1451
1452 let v2a_req = p
1457 .supersede(v1.ctx_id.clone())
1458 .version(2)
1459 .title("v2-A")
1460 .context_type(ContextType::DataSnapshot)
1461 .visibility(Visibility::Public)
1462 .build()
1463 .unwrap();
1464 let v2b_req = p
1465 .supersede(v1.ctx_id.clone())
1466 .version(2)
1467 .title("v2-B")
1468 .context_type(ContextType::DataSnapshot)
1469 .visibility(Visibility::Public)
1470 .build()
1471 .unwrap();
1472
1473 let s1 = Arc::clone(&server);
1474 let s2 = Arc::clone(&server);
1475 let h1 = tokio::task::spawn_blocking(move || s1.publish_unverified_for_tests(&v2a_req));
1476 let h2 = tokio::task::spawn_blocking(move || s2.publish_unverified_for_tests(&v2b_req));
1477 let (r1, r2) = (h1.await.unwrap(), h2.await.unwrap());
1478
1479 let outcomes = [r1, r2];
1480 let successes = outcomes.iter().filter(|r| r.is_ok()).count();
1481 let failures = outcomes.iter().filter(|r| r.is_err()).count();
1482 assert_eq!(
1483 successes, 1,
1484 "exactly one concurrent supersession MUST succeed; got {successes} successes / {failures} failures"
1485 );
1486 assert_eq!(failures, 1);
1487 for r in &outcomes {
1490 if let Err(e) = r {
1491 match e {
1492 AcdpError::SupersededTarget { reason, .. } => assert_eq!(
1493 *reason,
1494 acdp_primitives::error::SupersessionReason::AlreadySuperseded,
1495 "concurrent loser MUST be AlreadySuperseded"
1496 ),
1497 other => panic!("concurrent loser had wrong error: {other:?}"),
1498 }
1499 }
1500 }
1501 }
1502
1503 #[test]
1504 fn hostile_supersession_by_non_owner_rejected_predecessor_unchanged() {
1505 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1510 let victim = producer_for(7, "did:web:agents.example.com:victim");
1511 let v1_req = victim
1512 .publish_request()
1513 .title("v1")
1514 .context_type(ContextType::DataSnapshot)
1515 .visibility(Visibility::Public)
1516 .build()
1517 .unwrap();
1518 let v1 = server.publish_unverified_for_tests(&v1_req).unwrap();
1519
1520 let attacker = producer_for(9, "did:web:evil.example.com:attacker");
1523 let v2_req = attacker
1524 .supersede(v1.ctx_id.clone())
1525 .version(2)
1526 .title("hijacked")
1527 .context_type(ContextType::DataSnapshot)
1528 .visibility(Visibility::Public)
1529 .build()
1530 .unwrap();
1531 let err = server.publish_unverified_for_tests(&v2_req).unwrap_err();
1532 match err {
1534 AcdpError::SupersededTarget { reason, .. } => {
1535 assert_eq!(reason, acdp_primitives::error::SupersessionReason::NotFound);
1536 }
1537 other => panic!("expected uniform SupersededTarget::NotFound, got {other:?}"),
1538 }
1539 let cur = server.current(&v1.lineage_id, None).unwrap().unwrap();
1541 assert_eq!(cur.body.ctx_id, v1.ctx_id);
1542 assert_eq!(cur.body.title, "v1");
1543 assert_eq!(
1544 cur.registry_state.status,
1545 acdp_types::primitives::Status::Active
1546 );
1547 }
1548
1549 #[test]
1550 fn owner_supersession_still_succeeds_after_ownership_check() {
1551 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1552 let p = producer();
1553 let v1_req = p
1554 .publish_request()
1555 .title("v1")
1556 .context_type(ContextType::DataSnapshot)
1557 .visibility(Visibility::Public)
1558 .build()
1559 .unwrap();
1560 let v1 = server.publish_unverified_for_tests(&v1_req).unwrap();
1561 let v2_req = p
1562 .supersede(v1.ctx_id.clone())
1563 .version(2)
1564 .title("v2")
1565 .context_type(ContextType::DataSnapshot)
1566 .visibility(Visibility::Public)
1567 .build()
1568 .unwrap();
1569 let v2 = server.publish_unverified_for_tests(&v2_req).unwrap();
1570 assert_eq!(v2.version, 2);
1571 let cur = server.current(&v1.lineage_id, None).unwrap().unwrap();
1572 assert_eq!(cur.body.ctx_id, v2.ctx_id);
1573 }
1574
1575 #[test]
1576 fn supersession_with_unknown_target_rejected_as_not_found() {
1577 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1578 let p = producer();
1579 let phantom =
1580 CtxId("acdp://registry.example.com/12345678-1234-4321-8123-deadbeefcafe".into());
1581 let req = p
1582 .supersede(phantom)
1583 .version(2)
1584 .title("v2-orphan")
1585 .context_type(ContextType::DataSnapshot)
1586 .visibility(Visibility::Public)
1587 .build()
1588 .unwrap();
1589 let err = server.publish_unverified_for_tests(&req).unwrap_err();
1590 match err {
1591 AcdpError::SupersededTarget { reason, .. } => {
1592 assert_eq!(reason, acdp_primitives::error::SupersessionReason::NotFound);
1593 }
1594 other => panic!("expected SupersededTarget::NotFound, got {other:?}"),
1595 }
1596 }
1597
1598 #[test]
1599 fn version_mismatch_rejected() {
1600 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1601 let p = producer();
1602 let v1_req = p
1603 .publish_request()
1604 .title("v1")
1605 .context_type(ContextType::DataSnapshot)
1606 .visibility(Visibility::Public)
1607 .build()
1608 .unwrap();
1609 let v1 = server.publish_unverified_for_tests(&v1_req).unwrap();
1610 let v3_req = p
1612 .supersede(v1.ctx_id.clone())
1613 .version(3)
1614 .title("v3-skipped")
1615 .context_type(ContextType::DataSnapshot)
1616 .visibility(Visibility::Public)
1617 .build()
1618 .unwrap();
1619 let err = server.publish_unverified_for_tests(&v3_req).unwrap_err();
1620 match err {
1621 AcdpError::SupersededTarget { reason, .. } => {
1622 assert_eq!(
1623 reason,
1624 acdp_primitives::error::SupersessionReason::VersionMismatch
1625 );
1626 }
1627 other => panic!("expected VersionMismatch, got {other:?}"),
1628 }
1629 }
1630
1631 #[test]
1632 fn search_finds_published_context() {
1633 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1634 let p = producer();
1635 let req = p
1636 .publish_request()
1637 .title("Q1 portfolio risk")
1638 .context_type(ContextType::DataSnapshot)
1639 .visibility(Visibility::Public)
1640 .build()
1641 .unwrap();
1642 server.publish_unverified_for_tests(&req).unwrap();
1643 let resp = server
1644 .search(
1645 &SearchParams {
1646 q: Some("portfolio".into()),
1647 ..Default::default()
1648 },
1649 None,
1650 )
1651 .unwrap();
1652 assert_eq!(resp.matches.len(), 1);
1653 assert_eq!(resp.matches[0].title, "Q1 portfolio risk");
1654 }
1655
1656 #[test]
1662 fn lineage_filters_restricted_for_stranger() {
1663 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1664 let p = producer();
1665 let audience = AgentDid::new("did:web:audience.example.com:reader");
1666 let req = p
1667 .publish_request()
1668 .title("restricted v1")
1669 .context_type(ContextType::DataSnapshot)
1670 .visibility(Visibility::Restricted)
1671 .audience(vec![audience.clone()])
1672 .build()
1673 .unwrap();
1674 let resp = server.publish_unverified_for_tests(&req).unwrap();
1675
1676 let stranger = AgentDid::new("did:web:other.example.com:reader");
1677 let stranger_view = server.lineage(&resp.lineage_id, Some(&stranger)).unwrap();
1678 assert!(
1679 stranger_view.is_empty(),
1680 "stranger MUST NOT see restricted bodies via lineage(); got {} entries",
1681 stranger_view.len()
1682 );
1683
1684 let audience_view = server.lineage(&resp.lineage_id, Some(&audience)).unwrap();
1685 assert_eq!(
1686 audience_view.len(),
1687 1,
1688 "audience member MUST see the restricted body via lineage()"
1689 );
1690 }
1691
1692 #[test]
1695 fn current_filters_private_for_stranger() {
1696 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1697 let p = producer();
1698 let req = p
1699 .publish_request()
1700 .title("private v1")
1701 .context_type(ContextType::DataSnapshot)
1702 .visibility(Visibility::Private)
1703 .build()
1704 .unwrap();
1705 let resp = server.publish_unverified_for_tests(&req).unwrap();
1706
1707 let stranger = AgentDid::new("did:web:other.example.com:reader");
1708 assert!(
1709 server
1710 .current(&resp.lineage_id, Some(&stranger))
1711 .unwrap()
1712 .is_none(),
1713 "stranger MUST NOT see private contexts via current()"
1714 );
1715
1716 let producer_did = AgentDid::new("did:web:agents.example.com:test");
1717 assert!(
1718 server
1719 .current(&resp.lineage_id, Some(&producer_did))
1720 .unwrap()
1721 .is_some(),
1722 "producer MUST see private contexts via current()"
1723 );
1724 }
1725
1726 #[test]
1737 fn current_returns_none_when_all_superseded() {
1738 use crate::registry::store::RegistryStore;
1739 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1740 let p = producer();
1741 let req = p
1742 .publish_request()
1743 .title("v1")
1744 .context_type(ContextType::DataSnapshot)
1745 .visibility(Visibility::Public)
1746 .build()
1747 .unwrap();
1748 let resp = server.publish_unverified_for_tests(&req).unwrap();
1749 server.store().mark_superseded(&resp.ctx_id).unwrap();
1751
1752 let cur = server.current(&resp.lineage_id, None).unwrap();
1753 assert!(
1754 cur.is_none(),
1755 "all-superseded lineage MUST resolve to None per RFC-ACDP-0004 §5; got {cur:?}"
1756 );
1757 }
1758
1759 #[test]
1767 fn search_suppresses_public_when_anonymous_public_reads_false() {
1768 let mut c = caps();
1769 c.anonymous_public_reads = false;
1770 let server = RegistryServer::new(InMemoryStore::new(), c, "registry.example.com");
1771 let p = producer();
1772 let req = p
1773 .publish_request()
1774 .title("public-but-flag-off")
1775 .context_type(ContextType::DataSnapshot)
1776 .visibility(Visibility::Public)
1777 .build()
1778 .unwrap();
1779 server.publish_unverified_for_tests(&req).unwrap();
1780
1781 let err = server
1783 .search(
1784 &SearchParams {
1785 q: Some("public-but-flag-off".into()),
1786 ..Default::default()
1787 },
1788 None,
1789 )
1790 .unwrap_err();
1791 assert!(
1792 matches!(err, AcdpError::NotAuthorized(_)),
1793 "vis-009: anonymous search MUST be NotAuthorized when \
1794 anonymous_public_reads=false; got {err:?}"
1795 );
1796
1797 let stranger = AgentDid::new("did:web:other.example.com:reader");
1800 let authed = server
1801 .search(
1802 &SearchParams {
1803 q: Some("public-but-flag-off".into()),
1804 ..Default::default()
1805 },
1806 Some(&stranger),
1807 )
1808 .unwrap();
1809 assert_eq!(
1810 authed.matches.len(),
1811 1,
1812 "authenticated search MUST see public contexts regardless of anonymous_public_reads"
1813 );
1814 }
1815
1816 #[test]
1819 fn try_new_rejects_did_authority_mismatch() {
1820 let mut c = caps();
1821 c.registry_did = "did:web:other.example.com".into(); let res = RegistryServer::try_new(InMemoryStore::new(), c, "registry.example.com");
1823 match res {
1824 Err(AcdpError::SchemaViolation(msg)) => {
1825 assert!(msg.contains("does not match expected"))
1826 }
1827 Err(other) => panic!("expected SchemaViolation, got {other:?}"),
1828 Ok(_) => panic!("expected Err"),
1829 }
1830 }
1831
1832 #[test]
1833 fn try_new_rejects_caps_missing_ed25519() {
1834 let mut c = caps();
1835 c.supported_signature_algorithms = vec!["ecdsa-p256".into()]; let res = RegistryServer::try_new(InMemoryStore::new(), c, "registry.example.com");
1837 assert!(matches!(res, Err(AcdpError::SchemaViolation(_))));
1838 }
1839
1840 #[test]
1841 fn try_new_accepts_valid_caps() {
1842 RegistryServer::try_new(InMemoryStore::new(), caps(), "registry.example.com").unwrap();
1843 }
1844
1845 #[test]
1848 fn try_new_accepts_valid_dns_authority() {
1849 RegistryServer::try_new(InMemoryStore::new(), caps(), "registry.example.com").unwrap();
1850 }
1851
1852 #[test]
1853 fn try_new_rejects_host_port_authority() {
1854 let res = RegistryServer::try_new(InMemoryStore::new(), caps(), "localhost:8443");
1857 assert!(matches!(res, Err(AcdpError::SchemaViolation(_))));
1858 }
1859
1860 #[test]
1861 fn try_new_rejects_uppercase_authority() {
1862 let res = RegistryServer::try_new(InMemoryStore::new(), caps(), "Registry.Example.Com");
1863 assert!(matches!(res, Err(AcdpError::SchemaViolation(_))));
1864 }
1865
1866 #[test]
1867 fn try_new_rejects_url_form_authority() {
1868 let res =
1869 RegistryServer::try_new(InMemoryStore::new(), caps(), "https://registry.example.com");
1870 assert!(matches!(res, Err(AcdpError::SchemaViolation(_))));
1871 }
1872
1873 #[test]
1874 fn try_new_for_test_accepts_host_port() {
1875 let mut c = caps();
1878 c.registry_did = acdp_did::authority_to_did_web("localhost:8443");
1879 RegistryServer::try_new_for_test_authority(InMemoryStore::new(), c, "localhost:8443")
1880 .unwrap();
1881 }
1882
1883 fn producer_for(seed: u8, did: &str) -> Producer {
1886 Producer::new(
1887 SigningKey::from_bytes(&[seed; 32]),
1888 AgentDid::new(did),
1889 format!("{did}#key-1"),
1890 )
1891 }
1892
1893 #[test]
1894 fn retrieve_restricted_blocks_stranger_returns_none() {
1895 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1896 let owner = AgentDid::new("did:web:agents.example.com:owner");
1897 let audience_member = AgentDid::new("did:web:agents.example.com:friend");
1898 let p = producer_for(2, owner.as_str());
1899 let req = p
1900 .publish_request()
1901 .title("restricted")
1902 .context_type(ContextType::DataSnapshot)
1903 .visibility(Visibility::Restricted)
1904 .audience(vec![audience_member.clone()])
1905 .build()
1906 .unwrap();
1907 let resp = server.publish_unverified_for_tests(&req).unwrap();
1908 let stranger = AgentDid::new("did:web:agents.example.com:stranger");
1909
1910 assert!(server.retrieve(&resp.ctx_id, None).unwrap().is_none());
1911 assert!(server
1912 .retrieve(&resp.ctx_id, Some(&stranger))
1913 .unwrap()
1914 .is_none());
1915 assert!(server
1916 .retrieve(&resp.ctx_id, Some(&owner))
1917 .unwrap()
1918 .is_some());
1919 assert!(server
1920 .retrieve(&resp.ctx_id, Some(&audience_member))
1921 .unwrap()
1922 .is_some());
1923 }
1924
1925 #[test]
1926 fn search_restricted_filters_strangers() {
1927 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1928 let owner = AgentDid::new("did:web:agents.example.com:owner");
1929 let p = producer_for(3, owner.as_str());
1930 let req = p
1931 .publish_request()
1932 .title("hush hush")
1933 .context_type(ContextType::DataSnapshot)
1934 .visibility(Visibility::Restricted)
1935 .audience(vec![AgentDid::new("did:web:agents.example.com:friend")])
1936 .build()
1937 .unwrap();
1938 server.publish_unverified_for_tests(&req).unwrap();
1939
1940 let stranger = AgentDid::new("did:web:agents.example.com:stranger");
1941 let r_anon = server.search(&SearchParams::default(), None).unwrap();
1942 assert!(
1943 r_anon.matches.is_empty(),
1944 "anonymous must not see restricted"
1945 );
1946 let r_stranger = server
1947 .search(&SearchParams::default(), Some(&stranger))
1948 .unwrap();
1949 assert!(r_stranger.matches.is_empty());
1950 let r_owner = server
1951 .search(&SearchParams::default(), Some(&owner))
1952 .unwrap();
1953 assert_eq!(r_owner.matches.len(), 1);
1954 }
1955
1956 #[test]
1960 fn search_private_visible_only_to_producer() {
1961 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1962 let owner = AgentDid::new("did:web:agents.example.com:owner");
1963 let audience_member = AgentDid::new("did:web:agents.example.com:friend");
1964 let p = producer_for(4, owner.as_str());
1965 let req = p
1966 .publish_request()
1967 .title("private note")
1968 .context_type(ContextType::DataSnapshot)
1969 .visibility(Visibility::Private)
1970 .audience(vec![audience_member.clone()])
1971 .build()
1972 .unwrap();
1973 let resp = server.publish_unverified_for_tests(&req).unwrap();
1974
1975 let r_audience = server
1976 .search(&SearchParams::default(), Some(&audience_member))
1977 .unwrap();
1978 assert!(
1979 r_audience.matches.is_empty(),
1980 "audience must NOT see private in search"
1981 );
1982 let r_owner = server
1983 .search(&SearchParams::default(), Some(&owner))
1984 .unwrap();
1985 assert_eq!(
1986 r_owner.matches.len(),
1987 1,
1988 "owner sees their own private context"
1989 );
1990
1991 assert!(server
1993 .retrieve(&resp.ctx_id, Some(&audience_member))
1994 .unwrap()
1995 .is_some());
1996 }
1997
1998 #[cfg(feature = "client")]
2010 #[tokio::test]
2011 async fn publish_verified_rejects_non_did_web_key_id() {
2012 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
2013 let p = producer();
2014 let mut req = p
2015 .publish_request()
2016 .title("v1")
2017 .context_type(ContextType::DataSnapshot)
2018 .visibility(Visibility::Public)
2019 .build()
2020 .unwrap();
2021 let did_key = acdp_did::key::did_key_from_ed25519(
2028 &SigningKey::from_bytes(&[9u8; 32]).verifying_key_bytes(),
2029 );
2030 req.signature.key_id = acdp_did::key::did_key_url(&did_key).unwrap();
2031 let resolver = acdp_did::WebResolver::new();
2032 let err = server
2033 .publish_verified(&req, None, &resolver)
2034 .await
2035 .unwrap_err();
2036 match err {
2037 AcdpError::KeyNotAuthorized(msg) => assert!(msg.contains("did:web")),
2038 other => panic!("expected KeyNotAuthorized for non-did:web, got {other:?}"),
2039 }
2040 }
2041
2042 #[cfg(feature = "client")]
2043 #[tokio::test]
2044 async fn publish_verified_rejects_agent_id_keyid_mismatch() {
2045 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
2046 let p = producer();
2047 let mut req = p
2048 .publish_request()
2049 .title("v1")
2050 .context_type(ContextType::DataSnapshot)
2051 .visibility(Visibility::Public)
2052 .build()
2053 .unwrap();
2054 req.signature.key_id = "did:web:other.example.com:agent#key-1".into();
2055 let resolver = acdp_did::WebResolver::new();
2056 let err = server
2057 .publish_verified(&req, None, &resolver)
2058 .await
2059 .unwrap_err();
2060 match err {
2061 AcdpError::KeyNotAuthorized(msg) => assert!(msg.contains("agent_id")),
2062 other => panic!("expected KeyNotAuthorized for agent_id mismatch, got {other:?}"),
2063 }
2064 }
2065
2066 #[cfg(feature = "client")]
2067 #[tokio::test]
2068 async fn publish_verified_rejects_keyid_without_fragment() {
2069 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
2070 let p = producer();
2071 let mut req = p
2072 .publish_request()
2073 .title("v1")
2074 .context_type(ContextType::DataSnapshot)
2075 .visibility(Visibility::Public)
2076 .build()
2077 .unwrap();
2078 req.signature.key_id = "did:web:agents.example.com:test".into(); let resolver = acdp_did::WebResolver::new();
2080 let err = server
2081 .publish_verified(&req, None, &resolver)
2082 .await
2083 .unwrap_err();
2084 assert!(
2087 matches!(
2088 err,
2089 AcdpError::SchemaViolation(_) | AcdpError::KeyResolution(_)
2090 ),
2091 "expected fragment-rejection error, got {err:?}"
2092 );
2093 }
2094
2095 fn caps_with_idempotency() -> CapabilitiesDocument {
2098 let mut c = caps();
2099 c.supports_idempotency_key = true;
2100 c.limits.idempotency_key_ttl_seconds = Some(86_400);
2101 c
2102 }
2103
2104 #[test]
2105 fn idempotency_same_hash_returns_original_response() {
2106 let server = RegistryServer::new(
2107 InMemoryStore::new(),
2108 caps_with_idempotency(),
2109 "registry.example.com",
2110 );
2111 let p = producer();
2112 let req = p
2113 .publish_request()
2114 .title("once")
2115 .context_type(ContextType::DataSnapshot)
2116 .visibility(Visibility::Public)
2117 .build()
2118 .unwrap();
2119 let first = server
2124 .publish_unverified_in_tenant_for_tests(&req, Some("k-001"), None)
2125 .unwrap();
2126 let replayed = server
2127 .publish_unverified_in_tenant_for_tests(&req, Some("k-001"), None)
2128 .unwrap();
2129 assert_eq!(replayed.ctx_id, first.ctx_id);
2130 assert_eq!(replayed.lineage_id, first.lineage_id);
2131 assert_eq!(replayed.created_at, first.created_at);
2132 let resp = server.search(&SearchParams::default(), None).unwrap();
2134 assert_eq!(
2135 resp.matches.len(),
2136 1,
2137 "idempotent replay must not persist a second context"
2138 );
2139 }
2140
2141 #[test]
2142 fn idempotency_evicts_after_ttl() {
2143 let store = InMemoryStore::new();
2144 let agent = AgentDid::new("did:web:agents.example.com:test");
2145 let resp = PublishResponse {
2146 registry_receipt: None,
2147 ctx_id: acdp_types::CtxId("acdp://r/12345678-1234-4321-8123-000000000099".into()),
2148 lineage_id: acdp_types::LineageId(
2149 "lin:sha256:9999999999999999999999999999999999999999999999999999999999999999"
2150 .into(),
2151 ),
2152 version: 1,
2153 created_at: chrono::Utc::now(),
2154 status: Status::Active,
2155 };
2156 let past = chrono::Utc::now() - chrono::Duration::seconds(1);
2158 store
2159 .idempotency_record(
2160 &agent,
2161 "expired",
2162 &acdp_types::ContentHash("sha256:0".into()),
2163 &resp,
2164 past,
2165 )
2166 .unwrap();
2167 let prior = store.idempotency_lookup(&agent, "expired").unwrap();
2169 assert!(
2170 prior.is_none(),
2171 "lazy TTL eviction should drop expired record"
2172 );
2173 }
2174
2175 #[test]
2182 fn publish_unverified_in_tenant_for_tests_refuses_receipts_registry() {
2183 let mut c = caps();
2184 c.acdp_version = "0.2.0".into();
2185 let server = RegistryServer::new(InMemoryStore::new(), c, "registry.example.com")
2186 .with_receipt_signer(
2187 acdp_types::receipt::ReceiptSigner::new(
2188 SigningKey::from_bytes(&[0x33u8; 32]),
2189 "did:web:registry.example.com",
2190 "did:web:registry.example.com#receipt-key-1",
2191 )
2192 .unwrap(),
2193 )
2194 .unwrap();
2195 let p = producer();
2196 let req = p
2197 .publish_request()
2198 .title("should be refused")
2199 .context_type(ContextType::DataSnapshot)
2200 .visibility(Visibility::Public)
2201 .build()
2202 .unwrap();
2203 let err = server
2204 .publish_unverified_in_tenant_for_tests(&req, None, None)
2205 .unwrap_err();
2206 assert!(
2207 matches!(err, AcdpError::SchemaViolation(_)),
2208 "expected SchemaViolation refusal on a receipts-advertising registry, got {err:?}"
2209 );
2210 }
2211
2212 struct RecordingStore {
2225 inner: InMemoryStore,
2226 tenants: std::sync::Mutex<Vec<Option<String>>>,
2227 }
2228
2229 impl RecordingStore {
2230 fn new() -> Self {
2231 Self {
2232 inner: InMemoryStore::new(),
2233 tenants: std::sync::Mutex::new(Vec::new()),
2234 }
2235 }
2236 }
2237
2238 impl RegistryStore for RecordingStore {
2239 fn put(&self, body: Body) -> Result<(), AcdpError> {
2240 self.inner.put(body)
2241 }
2242
2243 fn get(&self, ctx_id: &CtxId) -> Result<Option<FullContext>, AcdpError> {
2244 self.inner.get(ctx_id)
2245 }
2246
2247 fn lineage(&self, lineage_id: &LineageId) -> Result<Vec<FullContext>, AcdpError> {
2248 self.inner.lineage(lineage_id)
2249 }
2250
2251 fn current(&self, lineage_id: &LineageId) -> Result<Option<FullContext>, AcdpError> {
2252 self.inner.current(lineage_id)
2253 }
2254
2255 fn mark_superseded(&self, ctx_id: &CtxId) -> Result<(), AcdpError> {
2256 self.inner.mark_superseded(ctx_id)
2257 }
2258
2259 fn first_version_ctx_id(&self, lineage_id: &LineageId) -> Result<Option<CtxId>, AcdpError> {
2260 self.inner.first_version_ctx_id(lineage_id)
2261 }
2262
2263 fn search(
2264 &self,
2265 params: &SearchParams,
2266 requester: Option<&AgentDid>,
2267 anonymous_public_reads: bool,
2268 ) -> Result<SearchResponse, AcdpError> {
2269 self.inner.search(params, requester, anonymous_public_reads)
2270 }
2271
2272 fn idempotency_lookup(
2273 &self,
2274 agent_id: &AgentDid,
2275 key: &str,
2276 ) -> Result<Option<crate::registry::store::IdempotencyRecord>, AcdpError> {
2277 self.inner.idempotency_lookup(agent_id, key)
2278 }
2279
2280 fn idempotency_record(
2281 &self,
2282 agent_id: &AgentDid,
2283 key: &str,
2284 hash: &acdp_types::primitives::ContentHash,
2285 response: &PublishResponse,
2286 expires_at: chrono::DateTime<chrono::Utc>,
2287 ) -> Result<(), AcdpError> {
2288 self.inner
2289 .idempotency_record(agent_id, key, hash, response, expires_at)
2290 }
2291
2292 fn idempotency_evict_expired(
2293 &self,
2294 now: chrono::DateTime<chrono::Utc>,
2295 ) -> Result<(), AcdpError> {
2296 self.inner.idempotency_evict_expired(now)
2297 }
2298
2299 fn commit_publish(
2300 &self,
2301 commit: crate::registry::store::PublishCommit<'_>,
2302 ) -> Result<crate::registry::store::PublishCommitOutcome, AcdpError> {
2303 self.tenants
2304 .lock()
2305 .unwrap()
2306 .push(commit.tenant.map(String::from));
2307 self.inner.commit_publish(commit)
2308 }
2309
2310 fn commit_lifecycle_event(
2311 &self,
2312 event: &acdp_types::lifecycle::LifecycleEvent,
2313 ) -> Result<crate::registry::store::LifecycleCommitOutcome, AcdpError> {
2314 self.inner.commit_lifecycle_event(event)
2315 }
2316 }
2317
2318 #[test]
2327 fn tenant_reaches_publish_commit_verbatim() {
2328 let server = RegistryServer::new(RecordingStore::new(), caps(), "registry.example.com");
2329 let p = producer();
2330
2331 let req_a = p
2332 .publish_request()
2333 .title("tenant a")
2334 .context_type(ContextType::DataSnapshot)
2335 .visibility(Visibility::Public)
2336 .build()
2337 .unwrap();
2338 server
2339 .publish_unverified_in_tenant_for_tests(&req_a, None, Some("tenant-a"))
2340 .unwrap();
2341
2342 let req_b = p
2343 .publish_request()
2344 .title("tenant none")
2345 .context_type(ContextType::DataSnapshot)
2346 .visibility(Visibility::Public)
2347 .build()
2348 .unwrap();
2349 server
2350 .publish_unverified_in_tenant_for_tests(&req_b, None, None)
2351 .unwrap();
2352
2353 let recorded = server.store().tenants.lock().unwrap().clone();
2354 assert_eq!(
2355 recorded,
2356 vec![Some("tenant-a".to_string()), None],
2357 "tenant must reach PublishCommit.tenant verbatim, in call order"
2358 );
2359 }
2360
2361 struct AlwaysDeny;
2364 impl crate::registry::RateLimiter for AlwaysDeny {
2365 fn check_publish(&self, agent_id: &AgentDid) -> Result<(), AcdpError> {
2366 Err(AcdpError::RateLimited(format!("blocked: {agent_id}")))
2367 }
2368 }
2369
2370 #[test]
2371 fn rate_limiter_blocks_publish_before_persist() {
2372 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com")
2373 .with_rate_limiter(AlwaysDeny);
2374 let p = producer();
2375 let req = p
2376 .publish_request()
2377 .title("blocked")
2378 .context_type(ContextType::DataSnapshot)
2379 .visibility(Visibility::Public)
2380 .build()
2381 .unwrap();
2382 let err = server.publish_unverified_for_tests(&req).unwrap_err();
2383 assert!(matches!(err, AcdpError::RateLimited(_)));
2384 let resp = server.search(&SearchParams::default(), None).unwrap();
2386 assert!(
2387 resp.matches.is_empty(),
2388 "rate-limited publish must not persist"
2389 );
2390 }
2391
2392 #[test]
2393 fn created_at_is_ms_truncated() {
2394 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
2395 let p = producer();
2396 let req = p
2397 .publish_request()
2398 .title("ms")
2399 .context_type(ContextType::DataSnapshot)
2400 .visibility(Visibility::Public)
2401 .build()
2402 .unwrap();
2403 let resp = server.publish_unverified_for_tests(&req).unwrap();
2404 assert_eq!(
2406 resp.created_at.timestamp_subsec_nanos() % 1_000_000,
2407 0,
2408 "created_at must be millisecond-truncated per RFC-ACDP-0001 §5.3"
2409 );
2410 }
2411
2412 fn did_key_request() -> acdp_types::publish::PublishRequest {
2415 let p = Producer::new_did_key(SigningKey::from_bytes(&[7u8; 32]));
2416 p.publish_request()
2417 .title("did:key publish")
2418 .context_type(ContextType::DataSnapshot)
2419 .visibility(Visibility::Public)
2420 .build()
2421 .unwrap()
2422 }
2423
2424 #[test]
2428 fn did_key_publish_rejected_when_not_advertised() {
2429 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
2430 let err = server
2431 .publish_verified_did_key(&did_key_request(), None)
2432 .unwrap_err();
2433 assert!(
2434 matches!(err, AcdpError::KeyResolution(ref m) if m.contains("supported_did_methods")),
2435 "got {err:?}"
2436 );
2437 }
2438
2439 #[test]
2443 fn did_key_publish_verified_end_to_end() {
2444 let mut c = caps();
2445 c.supported_did_methods.push("did:key".into());
2446 let server = RegistryServer::new(InMemoryStore::new(), c, "registry.example.com");
2447 let req = did_key_request();
2448 let resp = server.publish_verified_did_key(&req, None).unwrap();
2449 assert_eq!(resp.ctx_id.authority(), "registry.example.com");
2450
2451 let mut tampered = did_key_request();
2453 tampered.title = "tampered".into();
2454 let err = server
2455 .publish_verified_did_key(&tampered, None)
2456 .unwrap_err();
2457 assert!(matches!(err, AcdpError::HashMismatch { .. }), "got {err:?}");
2458 }
2459
2460 #[test]
2466 fn receiptless_idempotent_replay_survives_enabling_receipts() {
2467 let mut c = caps();
2468 c.acdp_version = "0.2.0".into();
2469 c.supported_did_methods.push("did:key".into());
2470 c.supports_idempotency_key = true;
2471 c.limits.idempotency_key_ttl_seconds = Some(86_400);
2472 let server = RegistryServer::new(InMemoryStore::new(), c, "registry.example.com")
2473 .with_receipt_signer(
2474 acdp_types::receipt::ReceiptSigner::new(
2475 SigningKey::from_bytes(&[0x11u8; 32]),
2476 "did:web:registry.example.com",
2477 "did:web:registry.example.com#receipt-key-1",
2478 )
2479 .unwrap(),
2480 )
2481 .unwrap();
2482
2483 let req = did_key_request();
2486 let pre_receipts_response = acdp_types::publish::PublishResponse {
2487 ctx_id: CtxId(format!(
2488 "acdp://registry.example.com/{}",
2489 uuid::Uuid::new_v4()
2490 )),
2491 lineage_id: acdp_crypto::derive_lineage_id(&CtxId(
2492 "acdp://registry.example.com/v1".into(),
2493 )),
2494 version: 1,
2495 created_at: acdp_primitives::time::trunc_ms(chrono::Utc::now()),
2496 status: Status::Active,
2497 registry_receipt: None,
2498 };
2499 server
2500 .store()
2501 .idempotency_record(
2502 &req.agent_id,
2503 "pre-receipts-key",
2504 &req.content_hash,
2505 &pre_receipts_response,
2506 chrono::Utc::now() + chrono::Duration::hours(1),
2507 )
2508 .unwrap();
2509
2510 let resp = server
2513 .publish_verified_did_key(&req, Some("pre-receipts-key"))
2514 .expect("replay of a pre-receipts record must succeed");
2515 assert_eq!(resp.ctx_id, pre_receipts_response.ctx_id);
2516 assert!(
2517 resp.registry_receipt.is_none(),
2518 "replay returns the original response verbatim"
2519 );
2520
2521 let p2 = Producer::new_did_key(SigningKey::from_bytes(&[8u8; 32]));
2523 let fresh = p2
2524 .publish_request()
2525 .title("fresh after enabling receipts")
2526 .context_type(ContextType::DataSnapshot)
2527 .visibility(Visibility::Public)
2528 .build()
2529 .unwrap();
2530 let fresh_resp = server.publish_verified_did_key(&fresh, None).unwrap();
2531 assert!(
2532 fresh_resp.registry_receipt.is_some(),
2533 "new inserts on a receipts registry must mint"
2534 );
2535 }
2536
2537 #[test]
2544 fn pinned_verified_publish_mints_receipt_with_correct_fingerprint() {
2545 use base64::{engine::general_purpose::STANDARD, Engine};
2546
2547 let key = SigningKey::from_bytes(&[3u8; 32]);
2548 let verifying_key_bytes = key.verifying_key_bytes();
2549 let pub_b64 = STANDARD.encode(verifying_key_bytes);
2550 let did = "did:web:agents.example.com:pinned-agent";
2551 let p = Producer::new(key, AgentDid::new(did), format!("{did}#key-1"));
2552 let req = p
2553 .publish_request()
2554 .title("pinned publish")
2555 .context_type(ContextType::DataSnapshot)
2556 .visibility(Visibility::Public)
2557 .build()
2558 .unwrap();
2559
2560 let mut c = caps();
2561 c.acdp_version = "0.2.0".into();
2562 let server = RegistryServer::new(InMemoryStore::new(), c, "registry.example.com")
2563 .with_receipt_signer(
2564 acdp_types::receipt::ReceiptSigner::new(
2565 SigningKey::from_bytes(&[0x22u8; 32]),
2566 "did:web:registry.example.com",
2567 "did:web:registry.example.com#receipt-key-1",
2568 )
2569 .unwrap(),
2570 )
2571 .unwrap();
2572
2573 let resp = server
2574 .publish_pinned_verified_in_tenant(&req, None, None, &pub_b64, "ed25519")
2575 .expect("pinned-verified publish must succeed on a receipts registry");
2576 let receipt = resp
2577 .registry_receipt
2578 .expect("a receipts-advertising registry must mint a receipt");
2579 assert_eq!(
2580 receipt["key_fingerprint"].as_str().unwrap(),
2581 acdp_crypto::fingerprint::fingerprint_ed25519(&verifying_key_bytes)
2582 );
2583 }
2584
2585 #[test]
2589 fn pinned_verified_publish_without_receipt_signer_succeeds_with_no_receipt() {
2590 use base64::{engine::general_purpose::STANDARD, Engine};
2591
2592 let key = SigningKey::from_bytes(&[4u8; 32]);
2593 let pub_b64 = STANDARD.encode(key.verifying_key_bytes());
2594 let did = "did:web:agents.example.com:pinned-agent-2";
2595 let p = Producer::new(key, AgentDid::new(did), format!("{did}#key-1"));
2596 let req = p
2597 .publish_request()
2598 .title("pinned publish, no receipts")
2599 .context_type(ContextType::DataSnapshot)
2600 .visibility(Visibility::Public)
2601 .build()
2602 .unwrap();
2603
2604 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
2605 let resp = server
2606 .publish_pinned_verified_in_tenant(&req, None, None, &pub_b64, "ed25519")
2607 .unwrap();
2608 assert!(resp.registry_receipt.is_none());
2609 }
2610
2611 #[test]
2614 fn did_key_publish_path_refuses_did_web() {
2615 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
2616 let p = producer();
2617 let req = p
2618 .publish_request()
2619 .title("did:web on the offline path")
2620 .context_type(ContextType::DataSnapshot)
2621 .visibility(Visibility::Public)
2622 .build()
2623 .unwrap();
2624 let err = server.publish_verified_did_key(&req, None).unwrap_err();
2625 assert!(
2626 matches!(err, AcdpError::KeyResolution(_)),
2627 "did:web on the offline path must be refused, got {err:?}"
2628 );
2629 }
2630}