1use crate::registry::rate_limit::{NoopRateLimiter, RateLimiter};
31use crate::registry::store::RegistryStore;
32use crate::registry::validator::{key_revocation_gate_applies, PublishValidator};
33use acdp_primitives::error::AcdpError;
34use acdp_types::{
35 body::{Body, FullContext},
36 capabilities::CapabilitiesDocument,
37 primitives::{AgentDid, CtxId, LineageId, Status, Visibility},
38 publish::{PublishRequest, PublishResponse},
39 revocation::KeyRevocation,
40 search::{SearchParams, SearchResponse},
41};
42
43pub struct RegistryServer<S: RegistryStore, L: RateLimiter = NoopRateLimiter> {
49 store: S,
50 caps: CapabilitiesDocument,
51 authority: String,
52 rate_limiter: L,
53 receipt_signer: Option<acdp_types::receipt::ReceiptSigner>,
58 mint_head_receipts: bool,
65 lifecycle_enabled: bool,
72}
73
74impl<S: RegistryStore> RegistryServer<S, NoopRateLimiter> {
75 #[doc(hidden)]
79 pub fn new(store: S, caps: CapabilitiesDocument, authority: impl Into<String>) -> Self {
80 Self {
81 store,
82 caps,
83 authority: authority.into(),
84 rate_limiter: NoopRateLimiter,
85 receipt_signer: None,
86 mint_head_receipts: false,
87 lifecycle_enabled: false,
88 }
89 }
90
91 pub fn try_new(
105 store: S,
106 caps: CapabilitiesDocument,
107 authority: impl Into<String>,
108 ) -> Result<Self, AcdpError> {
109 let authority = authority.into();
110 if !acdp_types::primitives::is_valid_dns_authority(&authority) {
113 return Err(AcdpError::SchemaViolation(format!(
114 "registry authority '{authority}' is not a valid DNS hostname \
115 (must be lowercase labels, e.g. 'registry.example.com'); \
116 use RegistryServer::try_new_for_test_authority for host:port test setups"
117 )));
118 }
119 acdp_validation::validate_capabilities(&caps)?;
120 let expected_did = acdp_did::authority_to_did_web(&authority);
123 if caps.registry_did != expected_did {
124 return Err(AcdpError::SchemaViolation(format!(
125 "capabilities.registry_did '{}' does not match expected '{expected_did}' \
126 for authority '{authority}'",
127 caps.registry_did
128 )));
129 }
130 Ok(Self {
131 store,
132 caps,
133 authority,
134 rate_limiter: NoopRateLimiter,
135 receipt_signer: None,
136 mint_head_receipts: false,
137 lifecycle_enabled: false,
138 })
139 }
140
141 #[doc(hidden)]
150 pub fn try_new_for_test_authority(
151 store: S,
152 caps: CapabilitiesDocument,
153 authority: impl Into<String>,
154 ) -> Result<Self, AcdpError> {
155 let authority = authority.into();
156 acdp_validation::validate_capabilities(&caps)?;
157 let expected_did = acdp_did::authority_to_did_web(&authority);
158 if caps.registry_did != expected_did {
159 return Err(AcdpError::SchemaViolation(format!(
160 "capabilities.registry_did '{}' does not match expected '{expected_did}' \
161 for authority '{authority}'",
162 caps.registry_did
163 )));
164 }
165 Ok(Self {
166 store,
167 caps,
168 authority,
169 rate_limiter: NoopRateLimiter,
170 receipt_signer: None,
171 mint_head_receipts: false,
172 lifecycle_enabled: false,
173 })
174 }
175}
176
177impl<S: RegistryStore, L: RateLimiter> RegistryServer<S, L> {
178 pub fn with_rate_limiter<L2: RateLimiter>(self, limiter: L2) -> RegistryServer<S, L2> {
180 RegistryServer {
181 store: self.store,
182 caps: self.caps,
183 authority: self.authority,
184 rate_limiter: limiter,
185 receipt_signer: self.receipt_signer,
186 mint_head_receipts: self.mint_head_receipts,
187 lifecycle_enabled: self.lifecycle_enabled,
188 }
189 }
190
191 pub fn with_receipt_signer(
207 mut self,
208 signer: acdp_types::receipt::ReceiptSigner,
209 ) -> Result<Self, AcdpError> {
210 if signer.registry_did() != self.caps.registry_did {
211 return Err(AcdpError::SchemaViolation(format!(
212 "receipt signer registry_did '{}' ≠ capabilities.registry_did '{}'",
213 signer.registry_did(),
214 self.caps.registry_did
215 )));
216 }
217 self.require_min_acdp_version((0, 2, 0), "acdp-registry-receipts")?;
220 let profile = acdp_types::profile::Profile::RegistryReceipts.as_str();
221 if !self.caps.profiles.iter().any(|p| p == profile) {
222 self.caps.profiles.push(profile.to_string());
223 }
224 self.receipt_signer = Some(signer);
225 Ok(self)
226 }
227
228 pub fn with_lineage_head_receipts(mut self) -> Result<Self, AcdpError> {
242 if self.receipt_signer.is_none() {
243 return Err(AcdpError::SchemaViolation(
244 "acdp-registry-head-receipts requires the acdp-registry-receipts profile \
245 (RFC-ACDP-0011 §9): call with_receipt_signer first"
246 .into(),
247 ));
248 }
249 self.require_min_acdp_version((0, 3, 0), "acdp-registry-head-receipts")?;
250 let profile = acdp_types::profile::Profile::RegistryHeadReceipts.as_str();
251 if !self.caps.profiles.iter().any(|p| p == profile) {
252 self.caps.profiles.push(profile.to_string());
253 }
254 self.mint_head_receipts = true;
255 Ok(self)
256 }
257
258 pub fn with_lifecycle(mut self) -> Result<Self, AcdpError> {
273 self.require_min_acdp_version((0, 3, 0), "acdp-registry-lifecycle")?;
274 let profile = acdp_types::profile::Profile::RegistryLifecycle.as_str();
275 if !self.caps.profiles.iter().any(|p| p == profile) {
276 self.caps.profiles.push(profile.to_string());
277 }
278 self.lifecycle_enabled = true;
279 Ok(self)
280 }
281
282 fn require_min_acdp_version(&self, min: (u64, u64, u64), what: &str) -> Result<(), AcdpError> {
287 let parts: Vec<u64> = self
288 .caps
289 .acdp_version
290 .split('.')
291 .map(|p| p.parse::<u64>())
292 .collect::<Result<_, _>>()
293 .map_err(|_| {
294 AcdpError::SchemaViolation(format!(
295 "capabilities.acdp_version '{}' is not a plain MAJOR.MINOR.PATCH version",
296 self.caps.acdp_version
297 ))
298 })?;
299 let [major, minor, patch] = parts.as_slice() else {
300 return Err(AcdpError::SchemaViolation(format!(
301 "capabilities.acdp_version '{}' is not a plain MAJOR.MINOR.PATCH version",
302 self.caps.acdp_version
303 )));
304 };
305 if (*major, *minor, *patch) < min {
306 return Err(AcdpError::SchemaViolation(format!(
307 "{what} requires capabilities.acdp_version >= {}.{}.{}, got '{}'",
308 min.0, min.1, min.2, self.caps.acdp_version
309 )));
310 }
311 Ok(())
312 }
313
314 pub fn store(&self) -> &S {
317 &self.store
318 }
319
320 pub fn capabilities(&self) -> &CapabilitiesDocument {
322 &self.caps
323 }
324
325 #[cfg(feature = "client")]
341 #[cfg_attr(
342 feature = "tracing",
343 tracing::instrument(
344 name = "acdp.publish_verified",
345 skip_all,
346 fields(
347 agent_id = req.agent_id.as_str(),
348 version = req.version,
349 idempotency_key = idempotency_key.is_some(),
350 ),
351 err(Display)
352 )
353 )]
354 pub async fn publish_verified(
355 &self,
356 req: &PublishRequest,
357 idempotency_key: Option<&str>,
358 resolver: &acdp_did::WebResolver,
359 ) -> Result<PublishResponse, AcdpError> {
360 self.publish_verified_in_tenant(req, idempotency_key, resolver, None)
361 .await
362 }
363
364 #[cfg(feature = "client")]
370 pub async fn publish_verified_in_tenant(
371 &self,
372 req: &PublishRequest,
373 idempotency_key: Option<&str>,
374 resolver: &acdp_did::WebResolver,
375 tenant: Option<&str>,
376 ) -> Result<PublishResponse, AcdpError> {
377 self.check_publish_rate_limit(&req.agent_id)?;
379
380 let raw_bytes = serde_json::to_vec(req)?.len();
381 let validator = PublishValidator::for_authority(&self.caps, &self.authority);
382 let _validated = validator.validate_post_schema(req, raw_bytes)?;
383
384 acdp_verify::verify_publish_request_signature(req, resolver).await?;
386
387 let revocation_check_needed = req.context_type.is_key_revocation()
406 && key_revocation_gate_applies(&self.caps.acdp_version);
407
408 let fingerprint = if self.receipt_signer.is_some() || revocation_check_needed {
414 Some(producer_key_fingerprint(req, resolver).await?)
415 } else {
416 None
417 };
418
419 if revocation_check_needed {
420 let fp = fingerprint.as_deref().ok_or_else(|| {
429 AcdpError::RegistryInternal(
430 "key-revocation fingerprint missing despite revocation_check_needed \
431 — this is an internal invariant violation, not a caller error"
432 .into(),
433 )
434 })?;
435 KeyRevocation::from_publish_request(req)?.check_not_self_signed(fp)?;
436 }
437
438 self.commit_via_store(req, idempotency_key, tenant, fingerprint)
446 }
447
448 pub fn publish_verified_did_key(
462 &self,
463 req: &PublishRequest,
464 idempotency_key: Option<&str>,
465 ) -> Result<PublishResponse, AcdpError> {
466 self.publish_verified_did_key_in_tenant(req, idempotency_key, None)
467 }
468
469 #[cfg_attr(
475 feature = "tracing",
476 tracing::instrument(
477 name = "acdp.publish_verified_did_key",
478 skip_all,
479 fields(
480 agent_id = req.agent_id.as_str(),
481 version = req.version,
482 idempotency_key = idempotency_key.is_some(),
483 ),
484 err(Display)
485 )
486 )]
487 pub fn publish_verified_did_key_in_tenant(
488 &self,
489 req: &PublishRequest,
490 idempotency_key: Option<&str>,
491 tenant: Option<&str>,
492 ) -> Result<PublishResponse, AcdpError> {
493 self.check_publish_rate_limit(&req.agent_id)?;
494
495 let raw_bytes = serde_json::to_vec(req)?.len();
496 let validator = PublishValidator::for_authority(&self.caps, &self.authority);
497 let _validated = validator.validate_post_schema(req, raw_bytes)?;
498
499 acdp_verify::verify_publish_request_signature_offline(req)?;
501
502 let fingerprint = if self.receipt_signer.is_some() {
505 let material = acdp_did::key::resolve_did_key(req.agent_id.as_str())?;
506 Some(acdp_crypto::fingerprint::fingerprint_did_key_material(
507 &material,
508 )?)
509 } else {
510 None
511 };
512
513 self.commit_via_store(req, idempotency_key, tenant, fingerprint)
514 }
515
516 #[doc(hidden)]
523 pub fn publish_unverified_for_tests(
524 &self,
525 req: &PublishRequest,
526 ) -> Result<PublishResponse, AcdpError> {
527 self.check_publish_rate_limit(&req.agent_id)?;
531
532 if self.receipt_signer.is_some() {
538 return Err(AcdpError::SchemaViolation(
539 "publish_unverified_for_tests is unavailable on a receipts-advertising \
540 registry (RFC-ACDP-0010 §7: no degraded mode); use publish_verified or \
541 publish_verified_did_key"
542 .into(),
543 ));
544 }
545 let raw_bytes = serde_json::to_vec(req)?.len();
555 let validator = PublishValidator::for_authority(&self.caps, &self.authority);
556 let _validated = validator.validate_post_schema(req, raw_bytes)?;
557 self.commit_via_store(req, None, None, None)
558 }
559
560 #[doc(hidden)]
582 pub fn publish_pinned_verified_in_tenant(
583 &self,
584 req: &PublishRequest,
585 idempotency_key: Option<&str>,
586 tenant: Option<&str>,
587 verified_public_key_b64: &str,
588 verified_algorithm: &str,
589 ) -> Result<PublishResponse, AcdpError> {
590 self.check_publish_rate_limit(&req.agent_id)?;
591
592 let raw_bytes = serde_json::to_vec(req)?.len();
593 let validator = PublishValidator::for_authority(&self.caps, &self.authority);
594 let _validated = validator.validate_post_schema(req, raw_bytes)?;
595
596 let revocation_check_needed = req.context_type.is_key_revocation()
604 && key_revocation_gate_applies(&self.caps.acdp_version);
605
606 let fingerprint = if self.receipt_signer.is_some() || revocation_check_needed {
607 Some(fingerprint_pinned_key(
608 verified_public_key_b64,
609 verified_algorithm,
610 )?)
611 } else {
612 None
613 };
614
615 if revocation_check_needed {
616 let fp = fingerprint.as_deref().ok_or_else(|| {
623 AcdpError::RegistryInternal(
624 "key-revocation fingerprint missing despite revocation_check_needed \
625 — this is an internal invariant violation, not a caller error"
626 .into(),
627 )
628 })?;
629 KeyRevocation::from_publish_request(req)?.check_not_self_signed(fp)?;
630 }
631
632 self.commit_via_store(req, idempotency_key, tenant, fingerprint)
633 }
634
635 fn check_publish_rate_limit(
639 &self,
640 agent_id: &acdp_types::primitives::AgentDid,
641 ) -> Result<(), AcdpError> {
642 match self.rate_limiter.check_publish(agent_id) {
643 Ok(()) => Ok(()),
644 Err(e) => {
645 #[cfg(feature = "tracing")]
646 tracing::warn!(
647 agent_id = agent_id.as_str(),
648 "publish rejected by rate limiter"
649 );
650 Err(e)
651 }
652 }
653 }
654
655 fn commit_via_store(
660 &self,
661 req: &PublishRequest,
662 idempotency_key: Option<&str>,
663 tenant: Option<&str>,
664 producer_key_fingerprint: Option<String>,
665 ) -> Result<PublishResponse, AcdpError> {
666 let idempotency = if self.caps.supports_idempotency_key {
667 idempotency_key.map(|key| crate::registry::store::PendingIdempotencyCommit {
668 key,
669 ttl: chrono::Duration::seconds(
670 self.caps
671 .limits
672 .idempotency_key_ttl_seconds
673 .unwrap_or(86_400) as i64,
674 ),
675 })
676 } else {
677 None
678 };
679 #[allow(clippy::type_complexity)]
682 let minter: Option<
683 Box<dyn Fn(&Body) -> Result<serde_json::Value, AcdpError> + Send + Sync>,
684 > = match (&self.receipt_signer, producer_key_fingerprint) {
685 (Some(signer), Some(fp)) => Some(Box::new(move |body: &Body| {
686 let receipt = signer.mint(
687 &body.ctx_id,
688 &body.lineage_id,
689 &body.origin_registry,
690 body.created_at,
691 &body.content_hash,
692 &fp,
693 )?;
694 serde_json::to_value(receipt).map_err(AcdpError::from)
695 })),
696 _ => None,
697 };
698 let minted_expected = minter.is_some();
699 let outcome = self
700 .store
701 .commit_publish(crate::registry::store::PublishCommit {
702 req,
703 authority: &self.authority,
704 idempotency,
705 tenant,
706 receipt_minter: minter.as_deref(),
707 })?;
708 let (response, replayed) = match outcome {
709 crate::registry::store::PublishCommitOutcome::Inserted(r) => (r, false),
710 crate::registry::store::PublishCommitOutcome::IdempotentReplay(r) => (r, true),
711 };
712 #[cfg(feature = "tracing")]
713 tracing::debug!(
714 ctx_id = %response.ctx_id.0,
715 lineage_id = %response.lineage_id.0,
716 version = response.version,
717 replayed,
718 "publish committed"
719 );
720 if minted_expected && !replayed && response.registry_receipt.is_none() {
734 return Err(AcdpError::RegistryInternal(
735 "receipt signer is configured but the store returned no receipt — \
736 the RegistryStore implementation must invoke PublishCommit::receipt_minter \
737 inside its commit (RFC-ACDP-0010 §7: no degraded mode)"
738 .into(),
739 ));
740 }
741 Ok(response)
742 }
743
744 pub fn retrieve(
757 &self,
758 ctx_id: &CtxId,
759 requester: Option<&AgentDid>,
760 ) -> Result<Option<FullContext>, AcdpError> {
761 let Some(ctx) = self.store.get(ctx_id)? else {
762 return Ok(None);
763 };
764 if !can_retrieve(&ctx.body, requester, &self.caps) {
765 return Ok(None);
766 }
767 Ok(Some(ctx))
768 }
769
770 pub fn retrieve_body(
772 &self,
773 ctx_id: &CtxId,
774 requester: Option<&AgentDid>,
775 ) -> Result<Option<Body>, AcdpError> {
776 Ok(self.retrieve(ctx_id, requester)?.map(|c| c.body))
777 }
778
779 pub fn lineage(
786 &self,
787 lineage_id: &LineageId,
788 requester: Option<&AgentDid>,
789 ) -> Result<Vec<FullContext>, AcdpError> {
790 let all = self.store.lineage(lineage_id)?;
791 Ok(all
792 .into_iter()
793 .filter(|ctx| can_retrieve(&ctx.body, requester, &self.caps))
794 .collect())
795 }
796
797 pub fn current(
817 &self,
818 lineage_id: &LineageId,
819 requester: Option<&AgentDid>,
820 ) -> Result<Option<FullContext>, AcdpError> {
821 let all = self.store.lineage(lineage_id)?;
822 for mut ctx in all.into_iter().rev() {
828 if !matches!(
829 ctx.registry_state.status,
830 Status::Superseded | Status::Retracted
831 ) && can_retrieve(&ctx.body, requester, &self.caps)
832 {
833 if self.mint_head_receipts {
834 let signer = self.receipt_signer.as_ref().ok_or_else(|| {
839 AcdpError::RegistryInternal(
840 "head-receipt minting enabled without a receipt signer \
841 (RFC-ACDP-0011 §9 prerequisite violated)"
842 .into(),
843 )
844 })?;
845 let receipt = signer.mint_lineage_head(
846 lineage_id,
847 &ctx.body.ctx_id,
848 ctx.body.version,
849 &ctx.registry_state.status,
850 chrono::Utc::now(),
851 )?;
852 ctx.lineage_head_receipt = Some(serde_json::to_value(receipt)?);
853 }
854 return Ok(Some(ctx));
855 }
856 }
857 Ok(None)
858 }
859
860 pub fn search(
873 &self,
874 params: &SearchParams,
875 requester: Option<&AgentDid>,
876 ) -> Result<SearchResponse, AcdpError> {
877 if requester.is_none() && !self.caps.anonymous_public_reads {
882 return Err(AcdpError::NotAuthorized(
883 "anonymous search requires authentication \
884 (registry caps: anonymous_public_reads=false)"
885 .into(),
886 ));
887 }
888 self.store
893 .search(params, requester, self.caps.anonymous_public_reads)
894 }
895
896 fn lifecycle_precheck(
928 &self,
929 event: &acdp_types::lifecycle::LifecycleEvent,
930 expected_type: &acdp_types::lifecycle::LifecycleEventType,
931 requester: Option<&AgentDid>,
932 ) -> Result<FullContext, AcdpError> {
933 if !self.lifecycle_enabled {
934 return Err(AcdpError::NotImplemented(
935 "this registry does not advertise acdp-registry-lifecycle \
936 (RFC-ACDP-0013 §6: lifecycle endpoints are not implemented)"
937 .into(),
938 ));
939 }
940 let ctx = self
942 .retrieve(&event.ctx_id, requester)?
943 .ok_or_else(|| AcdpError::NotFound(format!("context '{}' not found", event.ctx_id)))?;
944 event.validate()?;
946 if &event.event_type != expected_type {
947 return Err(AcdpError::SchemaViolation(format!(
948 "event_type '{}' does not match this endpoint (expected '{}', \
949 RFC-ACDP-0013 §6 step 2)",
950 event.event_type, expected_type
951 )));
952 }
953 let now = chrono::Utc::now();
954 if event.occurred_at > now + chrono::Duration::seconds(120) {
955 return Err(AcdpError::SchemaViolation(format!(
956 "event occurred_at '{}' is in the future beyond the 120s skew allowance \
957 (RFC-ACDP-0013 §4)",
958 event.occurred_at.format("%Y-%m-%dT%H:%M:%S%.3fZ")
959 )));
960 }
961 if event.actor != ctx.body.agent_id {
963 return Err(AcdpError::NotAuthorized(format!(
964 "event actor '{}' is not the context's producer — only the producer \
965 (agent_id) may use the lifecycle endpoints (RFC-ACDP-0013 §6 step 3)",
966 event.actor
967 )));
968 }
969 event.actor_bound_signature()?;
973 Ok(ctx)
974 }
975
976 fn lifecycle_commit(
980 &self,
981 event: &acdp_types::lifecycle::LifecycleEvent,
982 ) -> Result<FullContext, AcdpError> {
983 Ok(self.store.commit_lifecycle_event(event)?.into_context())
984 }
985
986 #[cfg(feature = "client")]
988 async fn lifecycle_transition_verified(
989 &self,
990 event: &acdp_types::lifecycle::LifecycleEvent,
991 expected_type: acdp_types::lifecycle::LifecycleEventType,
992 requester: Option<&AgentDid>,
993 resolver: &acdp_did::WebResolver,
994 ) -> Result<FullContext, AcdpError> {
995 let ctx = self.lifecycle_precheck(event, &expected_type, requester)?;
996 acdp_verify::verify_lifecycle_event(
1000 &serde_json::to_value(event)?,
1001 &event.ctx_id,
1002 &ctx.body.agent_id,
1003 None, resolver,
1005 )
1006 .await?;
1007 self.lifecycle_commit(event)
1008 }
1009
1010 fn lifecycle_transition_verified_did_key(
1012 &self,
1013 event: &acdp_types::lifecycle::LifecycleEvent,
1014 expected_type: acdp_types::lifecycle::LifecycleEventType,
1015 requester: Option<&AgentDid>,
1016 ) -> Result<FullContext, AcdpError> {
1017 let ctx = self.lifecycle_precheck(event, &expected_type, requester)?;
1018 acdp_verify::verify_lifecycle_event_offline(
1019 &serde_json::to_value(event)?,
1020 &event.ctx_id,
1021 &ctx.body.agent_id,
1022 None,
1023 )?;
1024 self.lifecycle_commit(event)
1025 }
1026
1027 #[cfg(feature = "client")]
1040 pub async fn retract_verified(
1041 &self,
1042 event: &acdp_types::lifecycle::LifecycleEvent,
1043 requester: Option<&AgentDid>,
1044 resolver: &acdp_did::WebResolver,
1045 ) -> Result<FullContext, AcdpError> {
1046 self.lifecycle_transition_verified(
1047 event,
1048 acdp_types::lifecycle::LifecycleEventType::Retracted,
1049 requester,
1050 resolver,
1051 )
1052 .await
1053 }
1054
1055 #[cfg(feature = "client")]
1062 pub async fn republish_verified(
1063 &self,
1064 event: &acdp_types::lifecycle::LifecycleEvent,
1065 requester: Option<&AgentDid>,
1066 resolver: &acdp_did::WebResolver,
1067 ) -> Result<FullContext, AcdpError> {
1068 self.lifecycle_transition_verified(
1069 event,
1070 acdp_types::lifecycle::LifecycleEventType::Republished,
1071 requester,
1072 resolver,
1073 )
1074 .await
1075 }
1076
1077 pub fn retract_verified_did_key(
1082 &self,
1083 event: &acdp_types::lifecycle::LifecycleEvent,
1084 requester: Option<&AgentDid>,
1085 ) -> Result<FullContext, AcdpError> {
1086 self.lifecycle_transition_verified_did_key(
1087 event,
1088 acdp_types::lifecycle::LifecycleEventType::Retracted,
1089 requester,
1090 )
1091 }
1092
1093 pub fn republish_verified_did_key(
1095 &self,
1096 event: &acdp_types::lifecycle::LifecycleEvent,
1097 requester: Option<&AgentDid>,
1098 ) -> Result<FullContext, AcdpError> {
1099 self.lifecycle_transition_verified_did_key(
1100 event,
1101 acdp_types::lifecycle::LifecycleEventType::Republished,
1102 requester,
1103 )
1104 }
1105
1106 #[doc(hidden)]
1111 pub fn retract_unverified_for_tests(
1112 &self,
1113 event: &acdp_types::lifecycle::LifecycleEvent,
1114 requester: Option<&AgentDid>,
1115 ) -> Result<FullContext, AcdpError> {
1116 self.lifecycle_precheck(
1117 event,
1118 &acdp_types::lifecycle::LifecycleEventType::Retracted,
1119 requester,
1120 )?;
1121 self.lifecycle_commit(event)
1122 }
1123
1124 #[doc(hidden)]
1126 pub fn republish_unverified_for_tests(
1127 &self,
1128 event: &acdp_types::lifecycle::LifecycleEvent,
1129 requester: Option<&AgentDid>,
1130 ) -> Result<FullContext, AcdpError> {
1131 self.lifecycle_precheck(
1132 event,
1133 &acdp_types::lifecycle::LifecycleEventType::Republished,
1134 requester,
1135 )?;
1136 self.lifecycle_commit(event)
1137 }
1138
1139 pub fn record_registry_lifecycle_event(
1151 &self,
1152 event: &acdp_types::lifecycle::LifecycleEvent,
1153 ) -> Result<FullContext, AcdpError> {
1154 if !self.lifecycle_enabled {
1155 return Err(AcdpError::NotImplemented(
1156 "this registry does not advertise acdp-registry-lifecycle \
1157 (RFC-ACDP-0013 §6)"
1158 .into(),
1159 ));
1160 }
1161 event.validate()?;
1162 if !event.event_type.is_registered() {
1163 return Err(AcdpError::SchemaViolation(format!(
1164 "event_type '{}' is not registered for acceptance in 0.3.0 \
1165 (RFC-ACDP-0013 §7.3)",
1166 event.event_type
1167 )));
1168 }
1169 if event.actor.as_str() != self.caps.registry_did {
1170 return Err(AcdpError::NotAuthorized(format!(
1171 "registry-initiated event actor '{}' ≠ this registry's DID '{}' \
1172 (RFC-ACDP-0013 §6)",
1173 event.actor, self.caps.registry_did
1174 )));
1175 }
1176 if self.receipt_signer.is_some() && !event.is_signed() {
1177 return Err(AcdpError::SchemaViolation(
1178 "a registry advertising acdp-registry-receipts MUST sign its lifecycle \
1179 events (RFC-ACDP-0013 §5)"
1180 .into(),
1181 ));
1182 }
1183 if event.is_signed() {
1184 event.actor_bound_signature()?;
1186 }
1187 self.lifecycle_commit(event)
1188 }
1189}
1190
1191pub(crate) fn can_retrieve(
1193 body: &Body,
1194 requester: Option<&AgentDid>,
1195 caps: &CapabilitiesDocument,
1196) -> bool {
1197 match body.visibility {
1198 Visibility::Public => caps.anonymous_public_reads || requester.is_some(),
1199 Visibility::Restricted | Visibility::Private => match requester {
1200 None => false,
1201 Some(r) => {
1202 r == &body.agent_id
1203 || body
1204 .audience
1205 .as_deref()
1206 .is_some_and(|a| a.iter().any(|d| d == r))
1207 }
1208 },
1209 }
1210}
1211
1212#[cfg(feature = "client")]
1227async fn producer_key_fingerprint(
1228 req: &PublishRequest,
1229 resolver: &acdp_did::WebResolver,
1230) -> Result<String, AcdpError> {
1231 acdp_crypto::fingerprint::fingerprint_for_key_id(
1232 &req.signature.key_id,
1233 &req.signature.algorithm,
1234 resolver,
1235 )
1236 .await
1237}
1238
1239fn fingerprint_pinned_key(public_key_b64: &str, algorithm: &str) -> Result<String, AcdpError> {
1244 use base64::{engine::general_purpose::STANDARD, Engine};
1245
1246 let raw = STANDARD
1247 .decode(public_key_b64)
1248 .map_err(|e| AcdpError::KeyResolution(format!("pinned key is not valid base64: {e}")))?;
1249 match algorithm {
1250 "ed25519" => {
1251 let arr: [u8; 32] = raw.as_slice().try_into().map_err(|_| {
1252 AcdpError::KeyResolution(format!(
1253 "pinned ed25519 key must be 32 bytes, got {}",
1254 raw.len()
1255 ))
1256 })?;
1257 Ok(acdp_crypto::fingerprint::fingerprint_ed25519(&arr))
1258 }
1259 "ecdsa-p256" => acdp_crypto::fingerprint::fingerprint_p256_sec1(&raw),
1260 other => Err(AcdpError::UnsupportedAlgorithm(format!(
1261 "cannot fingerprint a pinned key for algorithm '{other}'"
1262 ))),
1263 }
1264}
1265
1266#[cfg(test)]
1267mod tests {
1268 use super::*;
1269 use crate::registry::store::InMemoryStore;
1270 use acdp_crypto::SigningKey;
1271 use acdp_producer::Producer;
1272 use acdp_types::capabilities::Limits;
1273 use acdp_types::primitives::{AgentDid, ContextType, Visibility};
1274
1275 fn caps() -> CapabilitiesDocument {
1276 CapabilitiesDocument {
1277 acdp_version: "0.1.0".into(),
1278 registry_did: "did:web:registry.example.com".into(),
1279 supported_signature_algorithms: vec!["ed25519".into()],
1280 supported_did_methods: vec!["did:web".into()],
1281 profiles: vec!["acdp-registry-core".into()],
1282 limits: Limits {
1283 max_payload_bytes: 1_048_576,
1284 max_embedded_bytes: 65_536,
1285 idempotency_key_ttl_seconds: None,
1286 max_publish_per_minute: None,
1287 },
1288 read_authentication_methods: vec![],
1289 anonymous_public_reads: true,
1290 supports_idempotency_key: false,
1291 extensions: Default::default(),
1292 }
1293 }
1294
1295 fn producer() -> Producer {
1296 Producer::new(
1297 SigningKey::from_bytes(&[1u8; 32]),
1298 AgentDid::new("did:web:agents.example.com:test"),
1299 "did:web:agents.example.com:test#key-1",
1300 )
1301 }
1302
1303 #[test]
1304 fn publish_v1_then_retrieve() {
1305 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1306 let p = producer();
1307 let req = p
1308 .publish_request()
1309 .title("v1")
1310 .context_type(ContextType::DataSnapshot)
1311 .visibility(Visibility::Public)
1312 .build()
1313 .unwrap();
1314 let resp = server.publish_unverified_for_tests(&req).unwrap();
1315 assert_eq!(resp.version, 1);
1316 let ctx = server.retrieve(&resp.ctx_id, None).unwrap().unwrap();
1317 assert_eq!(ctx.body.title, "v1");
1318 let lineage = server.lineage(&resp.lineage_id, None).unwrap();
1320 assert_eq!(lineage.len(), 1);
1321 let cur = server.current(&resp.lineage_id, None).unwrap().unwrap();
1323 assert_eq!(cur.body.ctx_id, resp.ctx_id);
1324 }
1325
1326 #[test]
1327 fn supersession_marks_predecessor_and_returns_v2() {
1328 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1329 let p = producer();
1330 let v1_req = p
1331 .publish_request()
1332 .title("v1")
1333 .context_type(ContextType::DataSnapshot)
1334 .visibility(Visibility::Public)
1335 .build()
1336 .unwrap();
1337 let v1 = server.publish_unverified_for_tests(&v1_req).unwrap();
1338
1339 let v2_req = p
1340 .supersede(v1.ctx_id.clone())
1341 .version(2)
1342 .title("v2")
1343 .context_type(ContextType::DataSnapshot)
1344 .visibility(Visibility::Public)
1345 .build()
1346 .unwrap();
1347 let v2 = server.publish_unverified_for_tests(&v2_req).unwrap();
1348 assert_eq!(v2.version, 2);
1349 let v1_ctx = server.retrieve(&v1.ctx_id, None).unwrap().unwrap();
1351 assert!(matches!(
1352 v1_ctx.registry_state.status,
1353 acdp_types::Status::Superseded
1354 ));
1355 assert_eq!(v1.lineage_id, v2.lineage_id);
1357 let cur = server.current(&v1.lineage_id, None).unwrap().unwrap();
1359 assert_eq!(cur.body.ctx_id, v2.ctx_id);
1360 }
1361
1362 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1370 async fn concurrent_supersession_exactly_one_succeeds() {
1371 use std::sync::Arc;
1372 let server = Arc::new(RegistryServer::new(
1373 InMemoryStore::new(),
1374 caps(),
1375 "registry.example.com",
1376 ));
1377 let p = producer();
1378 let v1_req = p
1379 .publish_request()
1380 .title("v1")
1381 .context_type(ContextType::DataSnapshot)
1382 .visibility(Visibility::Public)
1383 .build()
1384 .unwrap();
1385 let v1 = server.publish_unverified_for_tests(&v1_req).unwrap();
1386
1387 let v2a_req = p
1392 .supersede(v1.ctx_id.clone())
1393 .version(2)
1394 .title("v2-A")
1395 .context_type(ContextType::DataSnapshot)
1396 .visibility(Visibility::Public)
1397 .build()
1398 .unwrap();
1399 let v2b_req = p
1400 .supersede(v1.ctx_id.clone())
1401 .version(2)
1402 .title("v2-B")
1403 .context_type(ContextType::DataSnapshot)
1404 .visibility(Visibility::Public)
1405 .build()
1406 .unwrap();
1407
1408 let s1 = Arc::clone(&server);
1409 let s2 = Arc::clone(&server);
1410 let h1 = tokio::task::spawn_blocking(move || s1.publish_unverified_for_tests(&v2a_req));
1411 let h2 = tokio::task::spawn_blocking(move || s2.publish_unverified_for_tests(&v2b_req));
1412 let (r1, r2) = (h1.await.unwrap(), h2.await.unwrap());
1413
1414 let outcomes = [r1, r2];
1415 let successes = outcomes.iter().filter(|r| r.is_ok()).count();
1416 let failures = outcomes.iter().filter(|r| r.is_err()).count();
1417 assert_eq!(
1418 successes, 1,
1419 "exactly one concurrent supersession MUST succeed; got {successes} successes / {failures} failures"
1420 );
1421 assert_eq!(failures, 1);
1422 for r in &outcomes {
1425 if let Err(e) = r {
1426 match e {
1427 AcdpError::SupersededTarget { reason, .. } => assert_eq!(
1428 *reason,
1429 acdp_primitives::error::SupersessionReason::AlreadySuperseded,
1430 "concurrent loser MUST be AlreadySuperseded"
1431 ),
1432 other => panic!("concurrent loser had wrong error: {other:?}"),
1433 }
1434 }
1435 }
1436 }
1437
1438 #[test]
1439 fn hostile_supersession_by_non_owner_rejected_predecessor_unchanged() {
1440 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1445 let victim = producer_for(7, "did:web:agents.example.com:victim");
1446 let v1_req = victim
1447 .publish_request()
1448 .title("v1")
1449 .context_type(ContextType::DataSnapshot)
1450 .visibility(Visibility::Public)
1451 .build()
1452 .unwrap();
1453 let v1 = server.publish_unverified_for_tests(&v1_req).unwrap();
1454
1455 let attacker = producer_for(9, "did:web:evil.example.com:attacker");
1458 let v2_req = attacker
1459 .supersede(v1.ctx_id.clone())
1460 .version(2)
1461 .title("hijacked")
1462 .context_type(ContextType::DataSnapshot)
1463 .visibility(Visibility::Public)
1464 .build()
1465 .unwrap();
1466 let err = server.publish_unverified_for_tests(&v2_req).unwrap_err();
1467 match err {
1469 AcdpError::SupersededTarget { reason, .. } => {
1470 assert_eq!(reason, acdp_primitives::error::SupersessionReason::NotFound);
1471 }
1472 other => panic!("expected uniform SupersededTarget::NotFound, got {other:?}"),
1473 }
1474 let cur = server.current(&v1.lineage_id, None).unwrap().unwrap();
1476 assert_eq!(cur.body.ctx_id, v1.ctx_id);
1477 assert_eq!(cur.body.title, "v1");
1478 assert_eq!(
1479 cur.registry_state.status,
1480 acdp_types::primitives::Status::Active
1481 );
1482 }
1483
1484 #[test]
1485 fn owner_supersession_still_succeeds_after_ownership_check() {
1486 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1487 let p = producer();
1488 let v1_req = p
1489 .publish_request()
1490 .title("v1")
1491 .context_type(ContextType::DataSnapshot)
1492 .visibility(Visibility::Public)
1493 .build()
1494 .unwrap();
1495 let v1 = server.publish_unverified_for_tests(&v1_req).unwrap();
1496 let v2_req = p
1497 .supersede(v1.ctx_id.clone())
1498 .version(2)
1499 .title("v2")
1500 .context_type(ContextType::DataSnapshot)
1501 .visibility(Visibility::Public)
1502 .build()
1503 .unwrap();
1504 let v2 = server.publish_unverified_for_tests(&v2_req).unwrap();
1505 assert_eq!(v2.version, 2);
1506 let cur = server.current(&v1.lineage_id, None).unwrap().unwrap();
1507 assert_eq!(cur.body.ctx_id, v2.ctx_id);
1508 }
1509
1510 #[test]
1511 fn supersession_with_unknown_target_rejected_as_not_found() {
1512 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1513 let p = producer();
1514 let phantom =
1515 CtxId("acdp://registry.example.com/12345678-1234-4321-8123-deadbeefcafe".into());
1516 let req = p
1517 .supersede(phantom)
1518 .version(2)
1519 .title("v2-orphan")
1520 .context_type(ContextType::DataSnapshot)
1521 .visibility(Visibility::Public)
1522 .build()
1523 .unwrap();
1524 let err = server.publish_unverified_for_tests(&req).unwrap_err();
1525 match err {
1526 AcdpError::SupersededTarget { reason, .. } => {
1527 assert_eq!(reason, acdp_primitives::error::SupersessionReason::NotFound);
1528 }
1529 other => panic!("expected SupersededTarget::NotFound, got {other:?}"),
1530 }
1531 }
1532
1533 #[test]
1534 fn version_mismatch_rejected() {
1535 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1536 let p = producer();
1537 let v1_req = p
1538 .publish_request()
1539 .title("v1")
1540 .context_type(ContextType::DataSnapshot)
1541 .visibility(Visibility::Public)
1542 .build()
1543 .unwrap();
1544 let v1 = server.publish_unverified_for_tests(&v1_req).unwrap();
1545 let v3_req = p
1547 .supersede(v1.ctx_id.clone())
1548 .version(3)
1549 .title("v3-skipped")
1550 .context_type(ContextType::DataSnapshot)
1551 .visibility(Visibility::Public)
1552 .build()
1553 .unwrap();
1554 let err = server.publish_unverified_for_tests(&v3_req).unwrap_err();
1555 match err {
1556 AcdpError::SupersededTarget { reason, .. } => {
1557 assert_eq!(
1558 reason,
1559 acdp_primitives::error::SupersessionReason::VersionMismatch
1560 );
1561 }
1562 other => panic!("expected VersionMismatch, got {other:?}"),
1563 }
1564 }
1565
1566 #[test]
1567 fn search_finds_published_context() {
1568 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1569 let p = producer();
1570 let req = p
1571 .publish_request()
1572 .title("Q1 portfolio risk")
1573 .context_type(ContextType::DataSnapshot)
1574 .visibility(Visibility::Public)
1575 .build()
1576 .unwrap();
1577 server.publish_unverified_for_tests(&req).unwrap();
1578 let resp = server
1579 .search(
1580 &SearchParams {
1581 q: Some("portfolio".into()),
1582 ..Default::default()
1583 },
1584 None,
1585 )
1586 .unwrap();
1587 assert_eq!(resp.matches.len(), 1);
1588 assert_eq!(resp.matches[0].title, "Q1 portfolio risk");
1589 }
1590
1591 #[test]
1597 fn lineage_filters_restricted_for_stranger() {
1598 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1599 let p = producer();
1600 let audience = AgentDid::new("did:web:audience.example.com:reader");
1601 let req = p
1602 .publish_request()
1603 .title("restricted v1")
1604 .context_type(ContextType::DataSnapshot)
1605 .visibility(Visibility::Restricted)
1606 .audience(vec![audience.clone()])
1607 .build()
1608 .unwrap();
1609 let resp = server.publish_unverified_for_tests(&req).unwrap();
1610
1611 let stranger = AgentDid::new("did:web:other.example.com:reader");
1612 let stranger_view = server.lineage(&resp.lineage_id, Some(&stranger)).unwrap();
1613 assert!(
1614 stranger_view.is_empty(),
1615 "stranger MUST NOT see restricted bodies via lineage(); got {} entries",
1616 stranger_view.len()
1617 );
1618
1619 let audience_view = server.lineage(&resp.lineage_id, Some(&audience)).unwrap();
1620 assert_eq!(
1621 audience_view.len(),
1622 1,
1623 "audience member MUST see the restricted body via lineage()"
1624 );
1625 }
1626
1627 #[test]
1630 fn current_filters_private_for_stranger() {
1631 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1632 let p = producer();
1633 let req = p
1634 .publish_request()
1635 .title("private v1")
1636 .context_type(ContextType::DataSnapshot)
1637 .visibility(Visibility::Private)
1638 .build()
1639 .unwrap();
1640 let resp = server.publish_unverified_for_tests(&req).unwrap();
1641
1642 let stranger = AgentDid::new("did:web:other.example.com:reader");
1643 assert!(
1644 server
1645 .current(&resp.lineage_id, Some(&stranger))
1646 .unwrap()
1647 .is_none(),
1648 "stranger MUST NOT see private contexts via current()"
1649 );
1650
1651 let producer_did = AgentDid::new("did:web:agents.example.com:test");
1652 assert!(
1653 server
1654 .current(&resp.lineage_id, Some(&producer_did))
1655 .unwrap()
1656 .is_some(),
1657 "producer MUST see private contexts via current()"
1658 );
1659 }
1660
1661 #[test]
1672 fn current_returns_none_when_all_superseded() {
1673 use crate::registry::store::RegistryStore;
1674 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1675 let p = producer();
1676 let req = p
1677 .publish_request()
1678 .title("v1")
1679 .context_type(ContextType::DataSnapshot)
1680 .visibility(Visibility::Public)
1681 .build()
1682 .unwrap();
1683 let resp = server.publish_unverified_for_tests(&req).unwrap();
1684 server.store().mark_superseded(&resp.ctx_id).unwrap();
1686
1687 let cur = server.current(&resp.lineage_id, None).unwrap();
1688 assert!(
1689 cur.is_none(),
1690 "all-superseded lineage MUST resolve to None per RFC-ACDP-0004 §5; got {cur:?}"
1691 );
1692 }
1693
1694 #[test]
1702 fn search_suppresses_public_when_anonymous_public_reads_false() {
1703 let mut c = caps();
1704 c.anonymous_public_reads = false;
1705 let server = RegistryServer::new(InMemoryStore::new(), c, "registry.example.com");
1706 let p = producer();
1707 let req = p
1708 .publish_request()
1709 .title("public-but-flag-off")
1710 .context_type(ContextType::DataSnapshot)
1711 .visibility(Visibility::Public)
1712 .build()
1713 .unwrap();
1714 server.publish_unverified_for_tests(&req).unwrap();
1715
1716 let err = server
1718 .search(
1719 &SearchParams {
1720 q: Some("public-but-flag-off".into()),
1721 ..Default::default()
1722 },
1723 None,
1724 )
1725 .unwrap_err();
1726 assert!(
1727 matches!(err, AcdpError::NotAuthorized(_)),
1728 "vis-009: anonymous search MUST be NotAuthorized when \
1729 anonymous_public_reads=false; got {err:?}"
1730 );
1731
1732 let stranger = AgentDid::new("did:web:other.example.com:reader");
1735 let authed = server
1736 .search(
1737 &SearchParams {
1738 q: Some("public-but-flag-off".into()),
1739 ..Default::default()
1740 },
1741 Some(&stranger),
1742 )
1743 .unwrap();
1744 assert_eq!(
1745 authed.matches.len(),
1746 1,
1747 "authenticated search MUST see public contexts regardless of anonymous_public_reads"
1748 );
1749 }
1750
1751 #[test]
1754 fn try_new_rejects_did_authority_mismatch() {
1755 let mut c = caps();
1756 c.registry_did = "did:web:other.example.com".into(); let res = RegistryServer::try_new(InMemoryStore::new(), c, "registry.example.com");
1758 match res {
1759 Err(AcdpError::SchemaViolation(msg)) => {
1760 assert!(msg.contains("does not match expected"))
1761 }
1762 Err(other) => panic!("expected SchemaViolation, got {other:?}"),
1763 Ok(_) => panic!("expected Err"),
1764 }
1765 }
1766
1767 #[test]
1768 fn try_new_rejects_caps_missing_ed25519() {
1769 let mut c = caps();
1770 c.supported_signature_algorithms = vec!["ecdsa-p256".into()]; let res = RegistryServer::try_new(InMemoryStore::new(), c, "registry.example.com");
1772 assert!(matches!(res, Err(AcdpError::SchemaViolation(_))));
1773 }
1774
1775 #[test]
1776 fn try_new_accepts_valid_caps() {
1777 RegistryServer::try_new(InMemoryStore::new(), caps(), "registry.example.com").unwrap();
1778 }
1779
1780 #[test]
1783 fn try_new_accepts_valid_dns_authority() {
1784 RegistryServer::try_new(InMemoryStore::new(), caps(), "registry.example.com").unwrap();
1785 }
1786
1787 #[test]
1788 fn try_new_rejects_host_port_authority() {
1789 let res = RegistryServer::try_new(InMemoryStore::new(), caps(), "localhost:8443");
1792 assert!(matches!(res, Err(AcdpError::SchemaViolation(_))));
1793 }
1794
1795 #[test]
1796 fn try_new_rejects_uppercase_authority() {
1797 let res = RegistryServer::try_new(InMemoryStore::new(), caps(), "Registry.Example.Com");
1798 assert!(matches!(res, Err(AcdpError::SchemaViolation(_))));
1799 }
1800
1801 #[test]
1802 fn try_new_rejects_url_form_authority() {
1803 let res =
1804 RegistryServer::try_new(InMemoryStore::new(), caps(), "https://registry.example.com");
1805 assert!(matches!(res, Err(AcdpError::SchemaViolation(_))));
1806 }
1807
1808 #[test]
1809 fn try_new_for_test_accepts_host_port() {
1810 let mut c = caps();
1813 c.registry_did = acdp_did::authority_to_did_web("localhost:8443");
1814 RegistryServer::try_new_for_test_authority(InMemoryStore::new(), c, "localhost:8443")
1815 .unwrap();
1816 }
1817
1818 fn producer_for(seed: u8, did: &str) -> Producer {
1821 Producer::new(
1822 SigningKey::from_bytes(&[seed; 32]),
1823 AgentDid::new(did),
1824 format!("{did}#key-1"),
1825 )
1826 }
1827
1828 #[test]
1829 fn retrieve_restricted_blocks_stranger_returns_none() {
1830 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1831 let owner = AgentDid::new("did:web:agents.example.com:owner");
1832 let audience_member = AgentDid::new("did:web:agents.example.com:friend");
1833 let p = producer_for(2, owner.as_str());
1834 let req = p
1835 .publish_request()
1836 .title("restricted")
1837 .context_type(ContextType::DataSnapshot)
1838 .visibility(Visibility::Restricted)
1839 .audience(vec![audience_member.clone()])
1840 .build()
1841 .unwrap();
1842 let resp = server.publish_unverified_for_tests(&req).unwrap();
1843 let stranger = AgentDid::new("did:web:agents.example.com:stranger");
1844
1845 assert!(server.retrieve(&resp.ctx_id, None).unwrap().is_none());
1846 assert!(server
1847 .retrieve(&resp.ctx_id, Some(&stranger))
1848 .unwrap()
1849 .is_none());
1850 assert!(server
1851 .retrieve(&resp.ctx_id, Some(&owner))
1852 .unwrap()
1853 .is_some());
1854 assert!(server
1855 .retrieve(&resp.ctx_id, Some(&audience_member))
1856 .unwrap()
1857 .is_some());
1858 }
1859
1860 #[test]
1861 fn search_restricted_filters_strangers() {
1862 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1863 let owner = AgentDid::new("did:web:agents.example.com:owner");
1864 let p = producer_for(3, owner.as_str());
1865 let req = p
1866 .publish_request()
1867 .title("hush hush")
1868 .context_type(ContextType::DataSnapshot)
1869 .visibility(Visibility::Restricted)
1870 .audience(vec![AgentDid::new("did:web:agents.example.com:friend")])
1871 .build()
1872 .unwrap();
1873 server.publish_unverified_for_tests(&req).unwrap();
1874
1875 let stranger = AgentDid::new("did:web:agents.example.com:stranger");
1876 let r_anon = server.search(&SearchParams::default(), None).unwrap();
1877 assert!(
1878 r_anon.matches.is_empty(),
1879 "anonymous must not see restricted"
1880 );
1881 let r_stranger = server
1882 .search(&SearchParams::default(), Some(&stranger))
1883 .unwrap();
1884 assert!(r_stranger.matches.is_empty());
1885 let r_owner = server
1886 .search(&SearchParams::default(), Some(&owner))
1887 .unwrap();
1888 assert_eq!(r_owner.matches.len(), 1);
1889 }
1890
1891 #[test]
1895 fn search_private_visible_only_to_producer() {
1896 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1897 let owner = AgentDid::new("did:web:agents.example.com:owner");
1898 let audience_member = AgentDid::new("did:web:agents.example.com:friend");
1899 let p = producer_for(4, owner.as_str());
1900 let req = p
1901 .publish_request()
1902 .title("private note")
1903 .context_type(ContextType::DataSnapshot)
1904 .visibility(Visibility::Private)
1905 .audience(vec![audience_member.clone()])
1906 .build()
1907 .unwrap();
1908 let resp = server.publish_unverified_for_tests(&req).unwrap();
1909
1910 let r_audience = server
1911 .search(&SearchParams::default(), Some(&audience_member))
1912 .unwrap();
1913 assert!(
1914 r_audience.matches.is_empty(),
1915 "audience must NOT see private in search"
1916 );
1917 let r_owner = server
1918 .search(&SearchParams::default(), Some(&owner))
1919 .unwrap();
1920 assert_eq!(
1921 r_owner.matches.len(),
1922 1,
1923 "owner sees their own private context"
1924 );
1925
1926 assert!(server
1928 .retrieve(&resp.ctx_id, Some(&audience_member))
1929 .unwrap()
1930 .is_some());
1931 }
1932
1933 #[cfg(feature = "client")]
1945 #[tokio::test]
1946 async fn publish_verified_rejects_non_did_web_key_id() {
1947 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1948 let p = producer();
1949 let mut req = p
1950 .publish_request()
1951 .title("v1")
1952 .context_type(ContextType::DataSnapshot)
1953 .visibility(Visibility::Public)
1954 .build()
1955 .unwrap();
1956 let did_key = acdp_did::key::did_key_from_ed25519(
1963 &SigningKey::from_bytes(&[9u8; 32]).verifying_key_bytes(),
1964 );
1965 req.signature.key_id = acdp_did::key::did_key_url(&did_key).unwrap();
1966 let resolver = acdp_did::WebResolver::new();
1967 let err = server
1968 .publish_verified(&req, None, &resolver)
1969 .await
1970 .unwrap_err();
1971 match err {
1972 AcdpError::KeyNotAuthorized(msg) => assert!(msg.contains("did:web")),
1973 other => panic!("expected KeyNotAuthorized for non-did:web, got {other:?}"),
1974 }
1975 }
1976
1977 #[cfg(feature = "client")]
1978 #[tokio::test]
1979 async fn publish_verified_rejects_agent_id_keyid_mismatch() {
1980 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
1981 let p = producer();
1982 let mut req = p
1983 .publish_request()
1984 .title("v1")
1985 .context_type(ContextType::DataSnapshot)
1986 .visibility(Visibility::Public)
1987 .build()
1988 .unwrap();
1989 req.signature.key_id = "did:web:other.example.com:agent#key-1".into();
1990 let resolver = acdp_did::WebResolver::new();
1991 let err = server
1992 .publish_verified(&req, None, &resolver)
1993 .await
1994 .unwrap_err();
1995 match err {
1996 AcdpError::KeyNotAuthorized(msg) => assert!(msg.contains("agent_id")),
1997 other => panic!("expected KeyNotAuthorized for agent_id mismatch, got {other:?}"),
1998 }
1999 }
2000
2001 #[cfg(feature = "client")]
2002 #[tokio::test]
2003 async fn publish_verified_rejects_keyid_without_fragment() {
2004 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
2005 let p = producer();
2006 let mut req = p
2007 .publish_request()
2008 .title("v1")
2009 .context_type(ContextType::DataSnapshot)
2010 .visibility(Visibility::Public)
2011 .build()
2012 .unwrap();
2013 req.signature.key_id = "did:web:agents.example.com:test".into(); let resolver = acdp_did::WebResolver::new();
2015 let err = server
2016 .publish_verified(&req, None, &resolver)
2017 .await
2018 .unwrap_err();
2019 assert!(
2022 matches!(
2023 err,
2024 AcdpError::SchemaViolation(_) | AcdpError::KeyResolution(_)
2025 ),
2026 "expected fragment-rejection error, got {err:?}"
2027 );
2028 }
2029
2030 fn caps_with_idempotency() -> CapabilitiesDocument {
2033 let mut c = caps();
2034 c.supports_idempotency_key = true;
2035 c.limits.idempotency_key_ttl_seconds = Some(86_400);
2036 c
2037 }
2038
2039 #[test]
2040 fn idempotency_same_hash_returns_original_response() {
2041 let server = RegistryServer::new(
2042 InMemoryStore::new(),
2043 caps_with_idempotency(),
2044 "registry.example.com",
2045 );
2046 let p = producer();
2047 let req = p
2048 .publish_request()
2049 .title("once")
2050 .context_type(ContextType::DataSnapshot)
2051 .visibility(Visibility::Public)
2052 .build()
2053 .unwrap();
2054 let first = server.publish_unverified_for_tests(&req).unwrap();
2056 let ttl = caps_with_idempotency()
2060 .limits
2061 .idempotency_key_ttl_seconds
2062 .unwrap() as i64;
2063 server
2064 .store()
2065 .idempotency_record(
2066 &req.agent_id,
2067 "k-001",
2068 &req.content_hash,
2069 &first,
2070 chrono::Utc::now() + chrono::Duration::seconds(ttl),
2071 )
2072 .unwrap();
2073 let prior = server
2074 .store()
2075 .idempotency_lookup(&req.agent_id, "k-001")
2076 .unwrap()
2077 .unwrap();
2078 assert_eq!(prior.content_hash, req.content_hash);
2079 assert_eq!(prior.response.ctx_id, first.ctx_id);
2080 }
2081
2082 #[test]
2083 fn idempotency_evicts_after_ttl() {
2084 let store = InMemoryStore::new();
2085 let agent = AgentDid::new("did:web:agents.example.com:test");
2086 let resp = PublishResponse {
2087 registry_receipt: None,
2088 ctx_id: acdp_types::CtxId("acdp://r/12345678-1234-4321-8123-000000000099".into()),
2089 lineage_id: acdp_types::LineageId(
2090 "lin:sha256:9999999999999999999999999999999999999999999999999999999999999999"
2091 .into(),
2092 ),
2093 version: 1,
2094 created_at: chrono::Utc::now(),
2095 status: Status::Active,
2096 };
2097 let past = chrono::Utc::now() - chrono::Duration::seconds(1);
2099 store
2100 .idempotency_record(
2101 &agent,
2102 "expired",
2103 &acdp_types::ContentHash("sha256:0".into()),
2104 &resp,
2105 past,
2106 )
2107 .unwrap();
2108 let prior = store.idempotency_lookup(&agent, "expired").unwrap();
2110 assert!(
2111 prior.is_none(),
2112 "lazy TTL eviction should drop expired record"
2113 );
2114 }
2115
2116 struct AlwaysDeny;
2119 impl crate::registry::RateLimiter for AlwaysDeny {
2120 fn check_publish(&self, agent_id: &AgentDid) -> Result<(), AcdpError> {
2121 Err(AcdpError::RateLimited(format!("blocked: {agent_id}")))
2122 }
2123 }
2124
2125 #[test]
2126 fn rate_limiter_blocks_publish_before_persist() {
2127 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com")
2128 .with_rate_limiter(AlwaysDeny);
2129 let p = producer();
2130 let req = p
2131 .publish_request()
2132 .title("blocked")
2133 .context_type(ContextType::DataSnapshot)
2134 .visibility(Visibility::Public)
2135 .build()
2136 .unwrap();
2137 let err = server.publish_unverified_for_tests(&req).unwrap_err();
2138 assert!(matches!(err, AcdpError::RateLimited(_)));
2139 let resp = server.search(&SearchParams::default(), None).unwrap();
2141 assert!(
2142 resp.matches.is_empty(),
2143 "rate-limited publish must not persist"
2144 );
2145 }
2146
2147 #[test]
2148 fn created_at_is_ms_truncated() {
2149 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
2150 let p = producer();
2151 let req = p
2152 .publish_request()
2153 .title("ms")
2154 .context_type(ContextType::DataSnapshot)
2155 .visibility(Visibility::Public)
2156 .build()
2157 .unwrap();
2158 let resp = server.publish_unverified_for_tests(&req).unwrap();
2159 assert_eq!(
2161 resp.created_at.timestamp_subsec_nanos() % 1_000_000,
2162 0,
2163 "created_at must be millisecond-truncated per RFC-ACDP-0001 §5.3"
2164 );
2165 }
2166
2167 fn did_key_request() -> acdp_types::publish::PublishRequest {
2170 let p = Producer::new_did_key(SigningKey::from_bytes(&[7u8; 32]));
2171 p.publish_request()
2172 .title("did:key publish")
2173 .context_type(ContextType::DataSnapshot)
2174 .visibility(Visibility::Public)
2175 .build()
2176 .unwrap()
2177 }
2178
2179 #[test]
2183 fn did_key_publish_rejected_when_not_advertised() {
2184 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
2185 let err = server
2186 .publish_verified_did_key(&did_key_request(), None)
2187 .unwrap_err();
2188 assert!(
2189 matches!(err, AcdpError::KeyResolution(ref m) if m.contains("supported_did_methods")),
2190 "got {err:?}"
2191 );
2192 }
2193
2194 #[test]
2198 fn did_key_publish_verified_end_to_end() {
2199 let mut c = caps();
2200 c.supported_did_methods.push("did:key".into());
2201 let server = RegistryServer::new(InMemoryStore::new(), c, "registry.example.com");
2202 let req = did_key_request();
2203 let resp = server.publish_verified_did_key(&req, None).unwrap();
2204 assert_eq!(resp.ctx_id.authority(), "registry.example.com");
2205
2206 let mut tampered = did_key_request();
2208 tampered.title = "tampered".into();
2209 let err = server
2210 .publish_verified_did_key(&tampered, None)
2211 .unwrap_err();
2212 assert!(matches!(err, AcdpError::HashMismatch { .. }), "got {err:?}");
2213 }
2214
2215 #[test]
2221 fn receiptless_idempotent_replay_survives_enabling_receipts() {
2222 let mut c = caps();
2223 c.acdp_version = "0.2.0".into();
2224 c.supported_did_methods.push("did:key".into());
2225 c.supports_idempotency_key = true;
2226 c.limits.idempotency_key_ttl_seconds = Some(86_400);
2227 let server = RegistryServer::new(InMemoryStore::new(), c, "registry.example.com")
2228 .with_receipt_signer(
2229 acdp_types::receipt::ReceiptSigner::new(
2230 SigningKey::from_bytes(&[0x11u8; 32]),
2231 "did:web:registry.example.com",
2232 "did:web:registry.example.com#receipt-key-1",
2233 )
2234 .unwrap(),
2235 )
2236 .unwrap();
2237
2238 let req = did_key_request();
2241 let pre_receipts_response = acdp_types::publish::PublishResponse {
2242 ctx_id: CtxId(format!(
2243 "acdp://registry.example.com/{}",
2244 uuid::Uuid::new_v4()
2245 )),
2246 lineage_id: acdp_crypto::derive_lineage_id(&CtxId(
2247 "acdp://registry.example.com/v1".into(),
2248 )),
2249 version: 1,
2250 created_at: acdp_primitives::time::trunc_ms(chrono::Utc::now()),
2251 status: Status::Active,
2252 registry_receipt: None,
2253 };
2254 server
2255 .store()
2256 .idempotency_record(
2257 &req.agent_id,
2258 "pre-receipts-key",
2259 &req.content_hash,
2260 &pre_receipts_response,
2261 chrono::Utc::now() + chrono::Duration::hours(1),
2262 )
2263 .unwrap();
2264
2265 let resp = server
2268 .publish_verified_did_key(&req, Some("pre-receipts-key"))
2269 .expect("replay of a pre-receipts record must succeed");
2270 assert_eq!(resp.ctx_id, pre_receipts_response.ctx_id);
2271 assert!(
2272 resp.registry_receipt.is_none(),
2273 "replay returns the original response verbatim"
2274 );
2275
2276 let p2 = Producer::new_did_key(SigningKey::from_bytes(&[8u8; 32]));
2278 let fresh = p2
2279 .publish_request()
2280 .title("fresh after enabling receipts")
2281 .context_type(ContextType::DataSnapshot)
2282 .visibility(Visibility::Public)
2283 .build()
2284 .unwrap();
2285 let fresh_resp = server.publish_verified_did_key(&fresh, None).unwrap();
2286 assert!(
2287 fresh_resp.registry_receipt.is_some(),
2288 "new inserts on a receipts registry must mint"
2289 );
2290 }
2291
2292 #[test]
2299 fn pinned_verified_publish_mints_receipt_with_correct_fingerprint() {
2300 use base64::{engine::general_purpose::STANDARD, Engine};
2301
2302 let key = SigningKey::from_bytes(&[3u8; 32]);
2303 let verifying_key_bytes = key.verifying_key_bytes();
2304 let pub_b64 = STANDARD.encode(verifying_key_bytes);
2305 let did = "did:web:agents.example.com:pinned-agent";
2306 let p = Producer::new(key, AgentDid::new(did), format!("{did}#key-1"));
2307 let req = p
2308 .publish_request()
2309 .title("pinned publish")
2310 .context_type(ContextType::DataSnapshot)
2311 .visibility(Visibility::Public)
2312 .build()
2313 .unwrap();
2314
2315 let mut c = caps();
2316 c.acdp_version = "0.2.0".into();
2317 let server = RegistryServer::new(InMemoryStore::new(), c, "registry.example.com")
2318 .with_receipt_signer(
2319 acdp_types::receipt::ReceiptSigner::new(
2320 SigningKey::from_bytes(&[0x22u8; 32]),
2321 "did:web:registry.example.com",
2322 "did:web:registry.example.com#receipt-key-1",
2323 )
2324 .unwrap(),
2325 )
2326 .unwrap();
2327
2328 let resp = server
2329 .publish_pinned_verified_in_tenant(&req, None, None, &pub_b64, "ed25519")
2330 .expect("pinned-verified publish must succeed on a receipts registry");
2331 let receipt = resp
2332 .registry_receipt
2333 .expect("a receipts-advertising registry must mint a receipt");
2334 assert_eq!(
2335 receipt["key_fingerprint"].as_str().unwrap(),
2336 acdp_crypto::fingerprint::fingerprint_ed25519(&verifying_key_bytes)
2337 );
2338 }
2339
2340 #[test]
2344 fn pinned_verified_publish_without_receipt_signer_succeeds_with_no_receipt() {
2345 use base64::{engine::general_purpose::STANDARD, Engine};
2346
2347 let key = SigningKey::from_bytes(&[4u8; 32]);
2348 let pub_b64 = STANDARD.encode(key.verifying_key_bytes());
2349 let did = "did:web:agents.example.com:pinned-agent-2";
2350 let p = Producer::new(key, AgentDid::new(did), format!("{did}#key-1"));
2351 let req = p
2352 .publish_request()
2353 .title("pinned publish, no receipts")
2354 .context_type(ContextType::DataSnapshot)
2355 .visibility(Visibility::Public)
2356 .build()
2357 .unwrap();
2358
2359 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
2360 let resp = server
2361 .publish_pinned_verified_in_tenant(&req, None, None, &pub_b64, "ed25519")
2362 .unwrap();
2363 assert!(resp.registry_receipt.is_none());
2364 }
2365
2366 #[test]
2369 fn did_key_publish_path_refuses_did_web() {
2370 let server = RegistryServer::new(InMemoryStore::new(), caps(), "registry.example.com");
2371 let p = producer();
2372 let req = p
2373 .publish_request()
2374 .title("did:web on the offline path")
2375 .context_type(ContextType::DataSnapshot)
2376 .visibility(Visibility::Public)
2377 .build()
2378 .unwrap();
2379 let err = server.publish_verified_did_key(&req, None).unwrap_err();
2380 assert!(
2381 matches!(err, AcdpError::KeyResolution(_)),
2382 "did:web on the offline path must be refused, got {err:?}"
2383 );
2384 }
2385}