acdp 0.2.0

Rust client library for the Agent Context Distribution Protocol (ACDP v0.1.0)
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
//! Cross-registry resolution per RFC-ACDP-0006 (feature = "client").
//!
//! Resolves a `ctx_id` whose authority differs from the registry the
//! consumer is currently talking to. Walks the lineage of `derived_from`
//! references with cycle detection, configurable depth / node / fanout
//! caps, and per-authority caching of the `RegistryClient` and
//! capabilities document.
//!
//! See RFC-ACDP-0006 §4.1 for the seven-step algorithm:
//!   1. Parse URI → authority
//!   2. Fetch the foreign registry's capabilities
//!   3. Verify the registry DID matches `did:web:<authority>`
//!   4. Retrieve the full context
//!   5. Verify content_hash
//!   6. Verify signature via DID resolution
//!   7. Walk `derived_from` references (with cycle/depth/node/fanout/timeout limits)

use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::Mutex;
use std::time::{Duration, Instant};

use crate::client::{ReceiptPolicy, RegistryClient, VerificationPolicy, VerifiedContext};
use crate::did::WebResolver;
use crate::error::AcdpError;
use crate::safe_http::SsrfPolicy;
use crate::types::body::Body;
use crate::types::primitives::CtxId;
use crate::types::CapabilitiesDocument;

/// Per-walk and per-resolve safety options.
///
/// Defaults are tuned for RFC-ACDP-0006 §7.4 / §7.5 — they bound a walk
/// even when the producer fabricates `derived_from` lists pointing into a
/// foreign registry's pathological lineage graph.
#[derive(Debug, Clone)]
pub struct ResolverOptions {
    /// Per-edge maximum depth (default 10).
    pub max_depth: usize,
    /// Total number of contexts the walk may verify (default 100). Acts
    /// as a hard ceiling even when individual hops respect `max_depth`.
    pub max_nodes: usize,
    /// Maximum `derived_from` count permitted on any single context the
    /// walker visits (default 32). A context that lists more parents is
    /// either malformed or hostile — short-circuit before fanning out.
    pub max_fanout: usize,
    /// Wall-clock budget for the entire walk (default 30 s). Wraps
    /// [`CrossRegistryResolver::walk_derived_from`] in `tokio::time::timeout`.
    pub total_timeout: Duration,
    /// How long to cache a foreign registry's capabilities document
    /// before re-fetching (default 5 min). Avoids hammering the foreign
    /// `/.well-known/acdp.json` on every hop.
    pub capabilities_ttl: Duration,
}

impl Default for ResolverOptions {
    fn default() -> Self {
        Self {
            max_depth: 10,
            max_nodes: 100,
            max_fanout: 32,
            total_timeout: Duration::from_secs(30),
            capabilities_ttl: Duration::from_secs(300),
        }
    }
}

/// Resolver for cross-registry references.
///
/// Holds a [`WebResolver`] for DID lookups and caches a [`RegistryClient`]
/// + capabilities document per authority for the lifetime of the resolver.
///
/// The [`SsrfPolicy`] is consulted on every URL the resolver constructs
/// (RFC-ACDP-0006 §7.1, §7.2).
pub struct CrossRegistryResolver {
    did_resolver: WebResolver,
    options: ResolverOptions,
    allowlist: Option<HashSet<String>>,
    ssrf_policy: SsrfPolicy,
    // Per-authority caches. Mutex-guarded for interior mutability across
    // the immutable `&self` API surface; contention is low since
    // authorities are few per walk.
    client_cache: Mutex<HashMap<String, RegistryClient>>,
    /// Per-authority capabilities cache. The `Duration` is the
    /// per-response TTL parsed from `Cache-Control: max-age=N` (capped
    /// at 3600s per RFC-ACDP-0006 §4.2). Replaces an earlier shape
    /// that used the resolver-wide `capabilities_ttl` for every entry,
    /// ignoring the registry's own cache hint (BUG-09).
    caps_cache: Mutex<HashMap<String, (CapabilitiesDocument, Instant, Duration)>>,
}

impl Default for CrossRegistryResolver {
    fn default() -> Self {
        Self::new()
    }
}

impl CrossRegistryResolver {
    /// Build a resolver with default settings: no allowlist, depth 10,
    /// HTTPS-only / no IP literals SSRF policy.
    pub fn new() -> Self {
        Self {
            did_resolver: WebResolver::new(),
            options: ResolverOptions::default(),
            allowlist: None,
            ssrf_policy: SsrfPolicy::default(),
            client_cache: Mutex::new(HashMap::new()),
            caps_cache: Mutex::new(HashMap::new()),
        }
    }

    /// Override the [`SsrfPolicy`] applied to outbound URLs.
    ///
    /// Useful for test environments that need to allow `http://` or
    /// IP-literal hosts. Production deployments SHOULD keep the default.
    pub fn with_ssrf_policy(mut self, policy: SsrfPolicy) -> Self {
        self.ssrf_policy = policy;
        self
    }

    /// Cap the number of `derived_from` hops walked in a single
    /// [`Self::walk_derived_from`] call.
    pub fn with_max_depth(mut self, depth: usize) -> Self {
        self.options.max_depth = depth;
        self
    }

    /// Replace the complete options struct (overrides every individual
    /// `with_*` setter that wasn't already applied).
    pub fn with_options(mut self, options: ResolverOptions) -> Self {
        self.options = options;
        self
    }

    /// Borrow the active options. Useful for tests + telemetry.
    pub fn options(&self) -> &ResolverOptions {
        &self.options
    }

    /// Override the [`WebResolver`] used for DID document lookups.
    ///
    /// Primary use is supplying a `WebResolver::with_root_cert_pem`
    /// instance in tests so a self-signed mock can answer DID-document
    /// requests for `did:web:localhost%3A<port>`. Production callers do
    /// not need this — the default resolver trusts the system CA bundle.
    pub fn with_did_resolver(mut self, resolver: WebResolver) -> Self {
        self.did_resolver = resolver;
        self
    }

    /// Pre-populate the per-authority [`RegistryClient`] cache.
    ///
    /// Primary use is the conformance harness: tests supply a client
    /// whose HTTP layer trusts the in-process TLS server's self-signed
    /// root certificate (via [`RegistryClient::with_root_cert_pem`]), so
    /// the resolver hits the mock instead of attempting a real network
    /// call. The seeded client wins over the lazy
    /// `RegistryClient::new_pinned` constructor that [`Self::resolve`]
    /// would otherwise invoke on first access.
    pub fn seed_client(&self, authority: impl Into<String>, client: RegistryClient) {
        self.client_cache
            .lock()
            .unwrap()
            .insert(authority.into(), client);
    }

    /// Restrict cross-registry resolution to a fixed set of authorities
    /// (lowercase DNS hostnames). When set, any reference outside the
    /// allowlist is rejected with [`AcdpError::CrossRegistryResolutionFailed`].
    pub fn with_allowlist<I, S>(mut self, authorities: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.allowlist = Some(authorities.into_iter().map(Into::into).collect());
        self
    }

    /// Resolve a single cross-registry [`CtxId`] end-to-end.
    ///
    /// Steps 1–6 of RFC-ACDP-0006 §4.1: parse, fetch capabilities,
    /// verify the registry DID *and* its DID document's web binding,
    /// retrieve, recompute hash, verify signature. The [`SsrfPolicy`]
    /// is checked first so a hostile authority cannot drive an
    /// internal-network request.
    pub async fn resolve(&self, ctx_id: &CtxId) -> Result<VerifiedContext, AcdpError> {
        let parsed = CtxId::parse(ctx_id.as_str())?;
        let authority = parsed.authority().to_string();
        self.check_allowlist(&authority)?;

        // RFC-ACDP-0006 §7: SSRF policy on the outbound base URL.
        let base = format!("https://{authority}");
        self.ssrf_policy
            .check_url(&base)
            .map_err(|e| AcdpError::CrossRegistryResolutionFailed(format!("SSRF policy: {e}")))?;

        // Cached client (and capabilities) per authority.
        let registry = self.client_for(&authority, &base).await?;
        let caps = self.capabilities_for(&authority, &registry).await?;

        // Step 3a: capabilities.registry_did MUST be `did:web:<authority>`.
        // BUG-06: percent-encode `:` for host:port authorities so the
        // expected DID round-trips with `authority_to_did_web`.
        let expected_did = crate::did::authority_to_did_web(&authority);
        if caps.registry_did != expected_did {
            return Err(AcdpError::CrossRegistryResolutionFailed(format!(
                "registry DID '{}' does not match expected '{expected_did}'",
                caps.registry_did
            )));
        }

        // Step 3b (RFC-ACDP-0006 §4.1 step 3): resolve the registry's
        // DID document and confirm the web binding matches `<authority>`.
        let registry_doc = self
            .did_resolver
            .resolve(&caps.registry_did)
            .await
            .map_err(|e| {
                AcdpError::CrossRegistryResolutionFailed(format!(
                    "could not resolve registry DID document for '{}': {e}",
                    caps.registry_did
                ))
            })?;
        if registry_doc.id != caps.registry_did {
            return Err(AcdpError::CrossRegistryResolutionFailed(format!(
                "registry DID document `id` '{}' does not match capabilities.registry_did '{}'",
                registry_doc.id, caps.registry_did
            )));
        }

        // Steps 4–6: retrieve + verify. fed-009 / RFC-ACDP-0010 §7+§11:
        // an upstream advertising `acdp-registry-receipts` MUST always
        // serve a receipt — absence is a registry fault (`invalid_receipt`),
        // not a degraded mode — so the policy escalates to `Require` for
        // such upstreams. Receipt-less upstreams proceed under the
        // v0.1.0 trust model (receipt verified only if one is present).
        let policy = if caps.claims_profile(crate::profile::Profile::RegistryReceipts) {
            VerificationPolicy {
                receipts: ReceiptPolicy::Require,
                ..VerificationPolicy::default()
            }
        } else {
            VerificationPolicy::default()
        };
        VerifiedContext::fetch_with_policy(&registry, &self.did_resolver, &parsed, &policy).await
    }

    /// Walk the `derived_from` graph rooted at `body` with cycle detection,
    /// a per-edge depth cap of [`ResolverOptions::max_depth`], a total-
    /// nodes cap of `max_nodes`, a per-context fanout cap of `max_fanout`,
    /// and a wall-clock `total_timeout`. Returns each verified ancestor
    /// (excluding the root). Breadth-first; closer ancestors are returned
    /// first.
    pub async fn walk_derived_from(&self, body: &Body) -> Result<Vec<VerifiedContext>, AcdpError> {
        let total_timeout = self.options.total_timeout;
        let fut = self.walk_derived_from_inner(body);
        match tokio::time::timeout(total_timeout, fut).await {
            Ok(res) => res,
            Err(_) => Err(AcdpError::CrossRegistryResolutionFailed(format!(
                "derived_from walk exceeded total_timeout={:?}",
                total_timeout
            ))),
        }
    }

    async fn walk_derived_from_inner(
        &self,
        body: &Body,
    ) -> Result<Vec<VerifiedContext>, AcdpError> {
        let mut seen: HashSet<String> = HashSet::new();
        seen.insert(body.ctx_id.0.clone());

        if body.derived_from.len() > self.options.max_fanout {
            return Err(AcdpError::CrossRegistryResolutionFailed(format!(
                "root context {} has derived_from fanout {} > max_fanout={}",
                body.ctx_id.0,
                body.derived_from.len(),
                self.options.max_fanout
            )));
        }

        let mut results: Vec<VerifiedContext> = Vec::new();
        let mut frontier: VecDeque<(CtxId, usize)> = body
            .derived_from
            .iter()
            .map(|c| (c.clone(), 1usize))
            .collect();

        while let Some((next, depth)) = frontier.pop_front() {
            if !seen.insert(next.0.clone()) {
                continue; // cycle
            }
            if depth > self.options.max_depth {
                return Err(AcdpError::CrossRegistryResolutionFailed(format!(
                    "derived_from walk exceeded max_depth={} at {}",
                    self.options.max_depth, next.0
                )));
            }
            if results.len() >= self.options.max_nodes {
                return Err(AcdpError::CrossRegistryResolutionFailed(format!(
                    "derived_from walk exceeded max_nodes={} (last attempted: {})",
                    self.options.max_nodes, next.0
                )));
            }
            let verified = self.resolve(&next).await?;
            let parents = &verified.body().derived_from;
            if parents.len() > self.options.max_fanout {
                return Err(AcdpError::CrossRegistryResolutionFailed(format!(
                    "context {} has derived_from fanout {} > max_fanout={}",
                    next.0,
                    parents.len(),
                    self.options.max_fanout
                )));
            }
            for parent in parents {
                if !seen.contains(parent.as_str()) {
                    frontier.push_back((parent.clone(), depth + 1));
                }
            }
            results.push(verified);
        }
        Ok(results)
    }

    fn check_allowlist(&self, authority: &str) -> Result<(), AcdpError> {
        if let Some(list) = &self.allowlist {
            if !list.contains(authority) {
                return Err(AcdpError::CrossRegistryResolutionFailed(format!(
                    "authority '{authority}' is not on the resolver allowlist"
                )));
            }
        }
        Ok(())
    }

    /// Return a cached `RegistryClient` for the authority, building one
    /// on first use. Reuse across hops avoids per-hop reqwest
    /// connection-pool churn.
    ///
    /// SEC-01: the client is built with [`RegistryClient::new_pinned`],
    /// which resolves the authority's DNS up-front, filters every
    /// resolved IP through the resolver's [`SsrfPolicy`], and pins the
    /// connection to that address. Without pinning a hostile `ctx_id`
    /// authority (e.g. `internal-host.example.com` resolving to
    /// `10.0.0.1` or `169.254.169.254`) would slip past the URL-syntax
    /// `check_url` gate and reach an internal target. The seeded test
    /// path ([`Self::seed_client`]) bypasses this constructor.
    async fn client_for(&self, authority: &str, base: &str) -> Result<RegistryClient, AcdpError> {
        {
            let cache = self.client_cache.lock().unwrap();
            if let Some(c) = cache.get(authority) {
                return Ok(c.clone());
            }
        }
        // Build with full DNS pinning before taking the cache lock —
        // `new_pinned` is async and the cache mutex must not be held
        // across the await.
        let client = RegistryClient::new_pinned(base, &self.ssrf_policy).await?;
        let mut cache = self.client_cache.lock().unwrap();
        Ok(cache.entry(authority.to_string()).or_insert(client).clone())
    }

    /// Return the cached capabilities for `authority`, fetching when
    /// the entry is missing or its per-response TTL has elapsed.
    ///
    /// BUG-09: TTL comes from the response's `Cache-Control: max-age=N`
    /// (clamped to `[1s, ResolverOptions::capabilities_ttl]` so the
    /// resolver-wide ceiling still applies) rather than a fixed value.
    /// A registry serving `Cache-Control: max-age=60` is honored; one
    /// serving no `Cache-Control` falls back to the
    /// [`RegistryClient::capabilities_with_ttl`] default (300s).
    async fn capabilities_for(
        &self,
        authority: &str,
        registry: &RegistryClient,
    ) -> Result<CapabilitiesDocument, AcdpError> {
        // Fast path: cache hit + within per-response TTL.
        {
            let cache = self.caps_cache.lock().unwrap();
            if let Some((caps, fetched_at, ttl)) = cache.get(authority) {
                if fetched_at.elapsed() < *ttl {
                    return Ok(caps.clone());
                }
            }
        }
        let (caps, response_ttl) = registry
            .capabilities_with_ttl()
            .await
            .map_err(|e| match e {
                AcdpError::Http(_) | AcdpError::KeyResolutionUnreachable(_) => {
                    AcdpError::CrossRegistryResolutionFailed(format!(
                        "could not reach registry '{authority}': {e}"
                    ))
                }
                other => other,
            })?;
        // Clamp to the resolver-wide ceiling so a registry advertising
        // an absurd `max-age` can't pin a stale doc indefinitely.
        let ttl = response_ttl.min(self.options.capabilities_ttl);
        let mut cache = self.caps_cache.lock().unwrap();
        cache.insert(authority.to_string(), (caps.clone(), Instant::now(), ttl));
        Ok(caps)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn allowlist_rejects_outside_authorities() {
        let resolver =
            CrossRegistryResolver::new().with_allowlist(["registry.example.com".to_string()]);
        let err = resolver.check_allowlist("evil.com").unwrap_err();
        assert!(matches!(err, AcdpError::CrossRegistryResolutionFailed(_)));
        resolver.check_allowlist("registry.example.com").unwrap();
    }

    #[test]
    fn options_default_values_match_doc() {
        let o = ResolverOptions::default();
        assert_eq!(o.max_depth, 10);
        assert_eq!(o.max_nodes, 100);
        assert_eq!(o.max_fanout, 32);
        assert_eq!(o.total_timeout, Duration::from_secs(30));
        assert_eq!(o.capabilities_ttl, Duration::from_secs(300));
    }

    #[test]
    fn with_options_replaces_full_struct() {
        let r = CrossRegistryResolver::new().with_options(ResolverOptions {
            max_depth: 3,
            max_nodes: 7,
            max_fanout: 2,
            total_timeout: Duration::from_secs(5),
            capabilities_ttl: Duration::from_secs(60),
        });
        assert_eq!(r.options().max_depth, 3);
        assert_eq!(r.options().max_nodes, 7);
        assert_eq!(r.options().max_fanout, 2);
    }

    #[test]
    fn cycle_detection_short_circuits() {
        let _resolver = CrossRegistryResolver::new();
        let mut seen: HashSet<String> = HashSet::new();
        let id = "acdp://r/12345678-1234-4321-8123-123456781234".to_string();
        assert!(seen.insert(id.clone()));
        assert!(!seen.insert(id));
    }
}