Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Remote @context loading + caching + pinned core contexts.
3 :
4 : use crate::context::Context;
5 : use antares_model::{NgsiError, TenantId};
6 : use serde_json::Value;
7 : use std::collections::HashMap;
8 : use std::sync::Arc;
9 : use tokio::sync::RwLock;
10 : // std Instant panics on wasm32; web-time is the std re-export
11 : // natively and performance.now() in the browser.
12 : #[cfg(not(target_arch = "wasm32"))]
13 : use std::time::Instant;
14 : #[cfg(target_arch = "wasm32")]
15 : use web_time::Instant;
16 : // moka's clock panics on wasm32 (std Instant); the browser build swaps
17 : // in the FIFO minicache behind the same call surface.
18 : #[cfg(target_arch = "wasm32")]
19 : use crate::minicache::Cache as BoundedCache;
20 : #[cfg(not(target_arch = "wasm32"))]
21 : use moka::sync::Cache as BoundedCache;
22 :
23 : /// Core context versions served from the build, never the network
24 : /// (uri.etsi.org serves an HTML landing page to plain HTTP clients).
25 : static PINNED: &[(&str, &str)] = &[
26 : (
27 : "https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.3.jsonld",
28 : include_str!("../contexts/core-v1.6.jsonld"),
29 : ),
30 : (
31 : "https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.4.jsonld",
32 : include_str!("../contexts/core-v1.6.jsonld"),
33 : ),
34 : (
35 : "https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.5.jsonld",
36 : include_str!("../contexts/core-v1.6.jsonld"),
37 : ),
38 : (
39 : "https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.6.jsonld",
40 : include_str!("../contexts/core-v1.6.jsonld"),
41 : ),
42 : (
43 : "https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.7.jsonld",
44 : include_str!("../contexts/core-v1.7.jsonld"),
45 : ),
46 : (
47 : "https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.8.jsonld",
48 : include_str!("../contexts/core-v1.8.jsonld"),
49 : ),
50 : (
51 : "https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.9.jsonld",
52 : include_str!("../contexts/core-v1.9.jsonld"),
53 : ),
54 : ];
55 :
56 : /// 4.4: "The NGSI-LD Core @context is publicly available at
57 : /// `https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.9.jsonld` and
58 : /// shall contain all the terms as mandated by annex B." It is what this broker
59 : /// advertises (the Link header of a `application/json` answer, the `@context`
60 : /// of an `ld+json` one, the context a forwarded request carries) and the
61 : /// document merged last when a request names no @context of its own. Older
62 : /// core versions stay in `PINNED` because a client may still reference one.
63 : pub const CORE_CONTEXT: &str = "https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.9.jsonld";
64 :
65 : /// Usage bookkeeping for one externally-referenced @context URL (5.13.3.5:
66 : /// localId, createdAt, numberOfHits, lastUsage of "Cached" entries).
67 : #[derive(Clone, Debug)]
68 : pub struct CtxUsage {
69 : /// The @context URL as the client referenced it.
70 : pub url: String,
71 : /// Broker-generated `localId` for the Cached entry.
72 : pub local_id: String,
73 : /// First reference (`createdAt`).
74 : pub created_at: String,
75 : /// Most recent reference (`lastUsage`).
76 : pub last_usage: String,
77 : /// Number of references (`numberOfHits`).
78 : pub hits: u64,
79 : }
80 :
81 : /// Name resolution runs before the HTTP client exists, so none of its
82 : /// timeouts cover it: an unresponsive resolver would hold the request path
83 : /// open indefinitely. A lookup that does not answer within this bound is a
84 : /// DENIAL — the policy never passes a destination it could not check.
85 : const DNS_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2);
86 :
87 : /// Outbound deadlines stretch 10× when the test binary runs under a
88 : /// sanitizer (ANTARES_TEST_SANITIZER, set by the strict workflow):
89 : /// ThreadSanitizer slows every thread, and hundreds of tests sharing one
90 : /// runner pushed loopback fetches past their limits. Production is 1.
91 29570 : pub fn slow_factor() -> u64 {
92 29570 : if std::env::var_os("ANTARES_TEST_SANITIZER").is_some() {
93 0 : 10
94 : } else {
95 29570 : 1
96 : }
97 29570 : }
98 :
99 : /// Egress policy hook for @context fetches: scheme allowlist is
100 : /// enforced in `fetch`; this adds the private-range deny (loopback,
101 : /// RFC 1918, link-local incl. the 169.254.169.254 metadata range, ULA).
102 : /// Private egress is ALLOWED by default (notifications must reach private
103 : /// nets out of the box — dev boxes, compose stacks and the ETSI/IOP mocks
104 : /// all live there); `ANTARES_EGRESS_ALLOW_PRIVATE=false`
105 : /// turns the deny on for internet-exposed deployments.
106 : /// The DNS-pinning resolver and redirect cap that enforce it on the wire
107 : /// are `PolicyResolver` / `client_builder` below.
108 : #[derive(Clone, Copy, Debug)]
109 : pub struct EgressPolicy {
110 : /// Whether fetches to private/loopback/link-local ranges are allowed.
111 : pub allow_private: bool,
112 : }
113 :
114 : /// Programmatic stand-in for `ANTARES_EGRESS_ALLOW_PRIVATE` — wasm32 has
115 : /// NO process environment (`std::env::var` always errs there), so the
116 : /// browser/Node embedder sets this before constructing the broker.
117 : static ALLOW_PRIVATE_OVERRIDE: std::sync::atomic::AtomicBool =
118 : std::sync::atomic::AtomicBool::new(false);
119 :
120 : /// Grant egress to private/loopback ranges for policies created AFTER this
121 : /// call. The wasm constructor path — wasm32 has no process environment — and
122 : /// the way a test grants itself loopback: an atomic store is the same switch
123 : /// as the environment variable with no write a concurrent reader can land in
124 : /// the middle of. Native deployments use `ANTARES_EGRESS_ALLOW_PRIVATE`.
125 498 : pub fn allow_private_egress(v: bool) {
126 498 : ALLOW_PRIVATE_OVERRIDE.store(v, std::sync::atomic::Ordering::Relaxed);
127 498 : }
128 :
129 : impl EgressPolicy {
130 : /// Policy from `ANTARES_EGRESS_ALLOW_PRIVATE` or the programmatic
131 : /// override. One of several `from_env` constructors in the workspace
132 : /// (the notifier's retry policy is another): each reads its own keys for
133 : /// its own type, so the shared name is the convention, not a copy.
134 8442 : pub fn from_env() -> Self {
135 : Self {
136 8442 : allow_private: Self::allow_private_from(
137 8442 : std::env::var("ANTARES_EGRESS_ALLOW_PRIVATE")
138 8442 : .ok()
139 8442 : .as_deref(),
140 0 : ) || ALLOW_PRIVATE_OVERRIDE.load(std::sync::atomic::Ordering::Relaxed),
141 : }
142 8442 : }
143 :
144 : /// Read the switch tolerantly: a security control that only understands
145 : /// one spelling hands the operator the opposite of the intent when the
146 : /// value is `FALSE` or carries stray whitespace.
147 8466 : fn allow_private_from(v: Option<&str>) -> bool {
148 8466 : v.is_none_or(|v| {
149 28 : let v = v.trim();
150 28 : !(v.eq_ignore_ascii_case("false") || v == "0")
151 28 : })
152 8466 : }
153 :
154 : /// The IPv4 destination an IPv6 address stands for. Four spellings
155 : /// reach one and the same host: `::ffff:a.b.c.d` (IPv4-mapped),
156 : /// `::a.b.c.d` (IPv4-compatible, RFC 4291), `64:ff9b::a.b.c.d` (the
157 : /// well-known NAT64 prefix, RFC 6052, which an IPv6-only deployment
158 : /// translates straight back to IPv4) and `2002:a.b.c.d::/48` (6to4,
159 : /// RFC 3056, where the two segments after the prefix ARE the IPv4
160 : /// address a 6to4 tunnel forwards to, whatever the rest of the address
161 : /// holds). Both classifiers below unwrap through this one function, so a
162 : /// range denied in one spelling is denied in every spelling.
163 297 : fn embedded_v4(v6: std::net::Ipv6Addr) -> Option<std::net::Ipv4Addr> {
164 297 : let s = v6.segments();
165 297 : let v4 = |hi: u16, lo: u16| std::net::Ipv4Addr::from((u32::from(hi) << 16) | u32::from(lo));
166 297 : if s[0] == 0x0064 && s[1] == 0xff9b && s[2..6] == [0, 0, 0, 0] {
167 18 : return Some(v4(s[6], s[7]));
168 279 : }
169 279 : if s[0] == 0x2002 {
170 40 : return Some(v4(s[1], s[2]));
171 239 : }
172 239 : v6.to_ipv4_mapped().or_else(|| v6.to_ipv4())
173 297 : }
174 :
175 : /// The cloud instance-metadata endpoints — the IPv4 link-local range
176 : /// (169.254.0.0/16, RFC 3927) shared by AWS, Azure, GCP and OCI,
177 : /// `100.100.100.200`, which sits in carrier-grade NAT rather than
178 : /// link-local, the IMDS-over-IPv6 ULA `fd00:ec2::254`, and every IPv6
179 : /// spelling of the IPv4 ones. Refused whatever `allow_private` says: no
180 : /// development box, compose stack or conformance mock lives there, so
181 : /// denying it costs nothing, while reaching it from a client-supplied
182 : /// @context URL or notification endpoint is the classic credential-theft
183 : /// SSRF.
184 1683 : pub fn ip_is_metadata(ip: std::net::IpAddr) -> bool {
185 1683 : match ip {
186 1402 : std::net::IpAddr::V4(v4) => v4.is_link_local() || v4.octets() == [100, 100, 100, 200],
187 281 : std::net::IpAddr::V6(v6) => {
188 281 : v6.segments() == [0xfd00, 0x0ec2, 0, 0, 0, 0, 0, 0x254]
189 263 : || Self::embedded_v4(v6)
190 263 : .is_some_and(|v4| Self::ip_is_metadata(std::net::IpAddr::V4(v4)))
191 : }
192 : }
193 1683 : }
194 :
195 : /// The ranges the private-egress deny covers. IPv4: loopback, RFC 1918,
196 : /// link-local, `0.0.0.0/8`, carrier-grade NAT, the IETF assignment and
197 : /// benchmarking blocks, and everything reserved above `240.0.0.0`.
198 : /// IPv6: loopback, unspecified, `fc00::/7` unique-local and `fe80::/10`
199 : /// link-local, with an embedded IPv4 destination judged as its IPv4
200 : /// self. Public because every client-supplied destination is judged by
201 : /// this one classifier — the @context fetch here and the MQTT endpoint
202 : /// in `antares-notifier` — so a range added to it cannot be missing from
203 : /// one of the bindings.
204 124 : pub fn ip_is_private(ip: std::net::IpAddr) -> bool {
205 124 : match ip {
206 90 : std::net::IpAddr::V4(v4) => {
207 90 : let o = v4.octets();
208 90 : v4.is_loopback()
209 58 : || v4.is_private()
210 40 : || v4.is_link_local()
211 : // 0.0.0.0/8 "this network", which a Linux stack routes to
212 : // the local host; subsumes the unspecified address
213 40 : || o[0] == 0
214 : // 100.64.0.0/10 carrier-grade NAT (RFC 6598), where a
215 : // cloud provider's internal services live
216 24 : || (o[0] == 100 && (64..128).contains(&o[1]))
217 : // 192.0.0.0/24 IETF protocol assignments (RFC 6890)
218 18 : || (o[0] == 192 && o[1] == 0 && o[2] == 0)
219 : // 198.18.0.0/15 benchmarking (RFC 2544)
220 16 : || (o[0] == 198 && (o[1] == 18 || o[1] == 19))
221 : // 240.0.0.0/4 reserved; subsumes the broadcast address
222 14 : || o[0] >= 240
223 : }
224 34 : std::net::IpAddr::V6(v6) => {
225 : // an IPv4 destination in IPv6 spelling is the same host —
226 : // judge it as its IPv4 self, or ::ffff:127.0.0.1 and
227 : // 64:ff9b::a9fe:a9fe slip past the IPv6 checks
228 34 : if Self::embedded_v4(v6)
229 34 : .is_some_and(|v4| Self::ip_is_private(std::net::IpAddr::V4(v4)))
230 : {
231 28 : return true;
232 6 : }
233 6 : v6.is_loopback()
234 6 : || v6.is_unspecified()
235 : // fc00::/7 unique-local + fe80::/10 link-local
236 6 : || (v6.segments()[0] & 0xfe00) == 0xfc00
237 6 : || (v6.segments()[0] & 0xffc0) == 0xfe80
238 : }
239 : }
240 124 : }
241 :
242 : /// Judge one destination against the policy. The instance-metadata
243 : /// ranges are refused whatever the switch says; the private ranges are
244 : /// refused only under `allow_private: false`, which ADR-0010 makes the
245 : /// internet-facing posture rather than the default. A host given as a
246 : /// name is resolved once, and any private address in the answer denies
247 : /// the fetch.
248 5287 : pub async fn check_host(&self, host: &str, port: u16) -> Result<(), String> {
249 5287 : self.check_host_within(host, port, DNS_TIMEOUT * slow_factor() as u32)
250 5287 : .await
251 5287 : }
252 :
253 5289 : async fn check_host_within(
254 5289 : &self,
255 5289 : host: &str,
256 5289 : port: u16,
257 5289 : dns_timeout: std::time::Duration,
258 5289 : ) -> Result<(), String> {
259 : // What this function can judge: a host given as a literal address,
260 : // in any spelling. A metadata address is refused before the
261 : // private-egress switch is consulted, so a deployment that allows
262 : // private egress (the default) cannot be steered at its own instance
263 : // credentials by IP. A host given as a NAME is only resolved here
264 : // when private egress is denied; with it allowed the name passes and
265 : // the verdict on what it resolves to belongs to the transport, which
266 : // judges the addresses it is about to dial: `PolicyResolver` for
267 : // every reqwest client, `checked_addr` for MQTT. Both drop a
268 : // metadata address whatever the switch says. A binding that dials
269 : // without such a filter is not covered by this check alone.
270 5289 : if let Ok(ip) = host.trim_matches(['[', ']']).parse::<std::net::IpAddr>() {
271 1176 : if Self::ip_is_metadata(ip) {
272 90 : return Err(format!("egress to {ip} denied (instance metadata)"));
273 1086 : }
274 4113 : }
275 5199 : if self.allow_private {
276 5125 : return Ok(());
277 74 : }
278 74 : if host.eq_ignore_ascii_case("localhost") {
279 2 : return Err(format!("egress to {host} denied (private range)"));
280 72 : }
281 72 : if let Ok(ip) = host.trim_matches(['[', ']']).parse::<std::net::IpAddr>() {
282 68 : if Self::ip_is_private(ip) {
283 56 : return Err(format!("egress to {ip} denied (private range)"));
284 12 : }
285 12 : return Ok(());
286 4 : }
287 : #[cfg(not(target_arch = "wasm32"))]
288 : {
289 : // The lookup runs under its own deadline and a lookup that does
290 : // not answer is a DENIAL: a destination the policy could not
291 : // check is never allowed through, and the request path never
292 : // waits on the resolver.
293 4 : let addrs = tokio::time::timeout(dns_timeout, tokio::net::lookup_host((host, port)))
294 4 : .await
295 4 : .map_err(|_| format!("resolving {host}: timed out"))?
296 2 : .map_err(|e| format!("resolving {host}: {e}"))?;
297 2 : for a in addrs {
298 2 : if Self::ip_is_private(a.ip()) {
299 : // The denial reaches the client verbatim in the RFC 7807
300 : // `detail`; naming the resolved address would turn the
301 : // request parameter into an internal-DNS oracle.
302 2 : return Err(format!("egress to {host} denied (private range)"));
303 0 : }
304 : }
305 : }
306 : // wasm32: a page cannot resolve DNS — the browser does, and its
307 : // same-origin/CORS machinery is the egress boundary there.
308 : #[cfg(target_arch = "wasm32")]
309 : let _ = (port, dns_timeout);
310 0 : Ok(())
311 5289 : }
312 : }
313 :
314 : /// axum handlers require `Send` futures and axum state requires
315 : /// `Send + Sync`, but reqwest's wasm client and futures are neither — the
316 : /// browser build is single-threaded, so `send_wrapper` bridges the gap
317 : /// soundly (it still panics at runtime on any actual cross-thread use).
318 : /// Natively these are the identity types.
319 : #[cfg(not(target_arch = "wasm32"))]
320 : pub type HttpClient = reqwest::Client;
321 : #[cfg(target_arch = "wasm32")]
322 : #[allow(missing_docs)] // documented on the native arm above
323 : pub type HttpClient = send_wrapper::SendWrapper<reqwest::Client>;
324 :
325 : /// Wrap a reqwest client as [`HttpClient`] (identity natively).
326 11632 : pub fn wrap_client(c: reqwest::Client) -> HttpClient {
327 : #[cfg(not(target_arch = "wasm32"))]
328 : {
329 11632 : c
330 : }
331 : #[cfg(target_arch = "wasm32")]
332 : {
333 : send_wrapper::SendWrapper::new(c)
334 : }
335 11632 : }
336 :
337 : /// Run one whole HTTP interaction (build → send → read body) as a unit
338 : /// whose future is Send on every target. Native: the identity. The inputs
339 : /// must move INTO the future and only Send data may come out.
340 : #[cfg(not(target_arch = "wasm32"))]
341 5123 : pub fn http_interaction<F: std::future::Future>(fut: F) -> F {
342 5123 : fut
343 5123 : }
344 : #[cfg(target_arch = "wasm32")]
345 : #[allow(missing_docs)] // documented on the native arm above
346 : pub fn http_interaction<F: std::future::Future>(fut: F) -> send_wrapper::SendWrapper<F> {
347 : send_wrapper::SendWrapper::new(fut)
348 : }
349 :
350 : /// The recursion box for `merge_entry` — Send on every target: the only
351 : /// un-Send piece (reqwest's wasm fetch) is already fenced inside
352 : /// `http_interaction`, so the box itself can stay Send and the axum handler
353 : /// futures above it keep their required Send bound.
354 : type BoxFut<'a, T> = std::pin::Pin<Box<dyn std::future::Future<Output = T> + Send + 'a>>;
355 :
356 : /// Redirect cap: a fetch may not be bounced more than this many times.
357 : /// Each hop is a fresh destination the policy has to clear, so the cap is what
358 : /// keeps an open redirector from walking us into a private range.
359 : pub const MAX_REDIRECTS: usize = 3;
360 :
361 : /// DNS pinning. `check_host` resolves a name to decide whether egress is
362 : /// allowed, but reqwest would resolve it *again* at connect time — a window in
363 : /// which the answer can change (DNS rebinding). Installing the policy as the
364 : /// client's resolver closes it: the addresses the connector dials are the ones
365 : /// this filter passed, so the check and the connect see the same answer by
366 : /// construction. Redirect hops go through it too.
367 : #[cfg(not(target_arch = "wasm32"))]
368 : #[derive(Debug)]
369 : pub struct PolicyResolver(EgressPolicy);
370 :
371 : #[cfg(not(target_arch = "wasm32"))]
372 : impl reqwest::dns::Resolve for PolicyResolver {
373 55 : fn resolve(&self, name: reqwest::dns::Name) -> reqwest::dns::Resolving {
374 55 : let allow_private = self.0.allow_private;
375 55 : Box::pin(async move {
376 55 : let host = name.as_str().to_owned();
377 55 : let addrs = tokio::net::lookup_host((host.as_str(), 0)).await?;
378 55 : let kept: Vec<std::net::SocketAddr> = addrs
379 326 : .filter(|a| !EgressPolicy::ip_is_metadata(a.ip()))
380 326 : .filter(|a| allow_private || !EgressPolicy::ip_is_private(a.ip()))
381 55 : .collect();
382 55 : if kept.is_empty() {
383 2 : return Err(format!("egress to {host} denied (private range)").into());
384 53 : }
385 53 : Ok(Box::new(kept.into_iter()) as reqwest::dns::Addrs)
386 55 : })
387 55 : }
388 : }
389 :
390 : /// One rustls crypto provider per process, installed before the first
391 : /// client is built. reqwest is compiled provider-less (`rustls-no-provider`)
392 : /// because its own `rustls` feature would pull aws-lc-rs beside ring, and a
393 : /// provider-less client PANICS at build time with none installed. ring is
394 : /// the workspace's only provider, so a second call here would be the same
395 : /// choice; the result is discarded because another crate installing the same
396 : /// provider first is success, not a conflict.
397 : ///
398 : /// `client_builder` calls it, so every client this workspace builds is
399 : /// covered. It is public because a DEPENDENCY can build a client too — the
400 : /// OTLP exporter does, from inside `opentelemetry-http` — and a binary must
401 : /// therefore install the provider before it wires anything up. The browser
402 : /// owns TLS behind `fetch`, so on wasm32 this is a no-op by design.
403 : #[cfg(not(target_arch = "wasm32"))]
404 11707 : pub fn install_crypto_provider() {
405 : static ONCE: std::sync::OnceLock<()> = std::sync::OnceLock::new();
406 11707 : ONCE.get_or_init(|| {
407 263 : let _ = rustls::crypto::ring::default_provider().install_default();
408 263 : });
409 11707 : }
410 :
411 : #[cfg(target_arch = "wasm32")]
412 : #[allow(missing_docs)] // documented on the native arm above
413 : pub fn install_crypto_provider() {}
414 :
415 : /// The one outbound-client constructor: every reqwest client in the
416 : /// broker — @context fetches, notifications, federation forwards — is built
417 : /// from this, so the policy cannot be forgotten at a call site. Timeouts stay
418 : /// the caller's choice; the security-relevant settings do not.
419 : ///
420 : /// `ANTARES_EXTRA_CA_FILE`: optional PEM bundle of ADDITIONAL trust anchors
421 : /// (private CAs, corporate proxies — and servers that ship an incomplete
422 : /// chain, as forge.etsi.org does). Verification itself is never
423 : /// disableable; this only widens what it trusts, per deployment.
424 : /// Read once per builder call — the wiring constructs clients at startup.
425 : #[cfg(not(target_arch = "wasm32"))]
426 11650 : pub fn client_builder(policy: EgressPolicy) -> reqwest::ClientBuilder {
427 : // SSRF: `PolicyResolver` only fires for HOSTNAME targets — reqwest
428 : // dials IP-LITERAL URLs directly, so a `302 Location: http://169.254.169.254/`
429 : // would skip the egress check on every hop. The custom redirect policy
430 : // re-checks each hop's URL: an IP literal in a private range is refused,
431 : // hostnames still clear through the resolver at connect. Hop count capped.
432 11650 : let allow_private = policy.allow_private;
433 11650 : let redirect = reqwest::redirect::Policy::custom(move |attempt| {
434 : // previous() includes the initial URL, so `> MAX_REDIRECTS` matches
435 : // Policy::limited(MAX_REDIRECTS): 1 initial request + MAX_REDIRECTS hops.
436 12 : if attempt.previous().len() > MAX_REDIRECTS {
437 : // same shape as Policy::limited: a redirect error, is_redirect()==true
438 2 : return attempt.error(format!("exceeded {MAX_REDIRECTS} redirects"));
439 10 : }
440 : // `Url` hands back an IPv6 host bracketed (`[::1]`), which
441 : // `IpAddr::from_str` rejects — trim as `check_host` does, or every
442 : // IPv6 literal hop skips the check.
443 10 : if let Ok(ip) = attempt
444 10 : .url()
445 10 : .host_str()
446 10 : .unwrap_or("")
447 10 : .trim_matches(['[', ']'])
448 10 : .parse::<std::net::IpAddr>()
449 : {
450 10 : if EgressPolicy::ip_is_metadata(ip)
451 10 : || (!allow_private && EgressPolicy::ip_is_private(ip))
452 : {
453 : // stop (don't follow) — the caller sees a non-2xx and fails the
454 : // fetch, but we never connected to the private target.
455 4 : return attempt.stop();
456 6 : }
457 0 : }
458 6 : attempt.follow()
459 12 : });
460 11650 : install_crypto_provider();
461 11650 : let mut b = reqwest::Client::builder()
462 11650 : .redirect(redirect)
463 11650 : .dns_resolver(std::sync::Arc::new(PolicyResolver(policy)));
464 11650 : if let Ok(path) = std::env::var("ANTARES_EXTRA_CA_FILE") {
465 0 : match std::fs::read(&path) {
466 0 : Ok(pem) => match reqwest::Certificate::from_pem_bundle(&pem) {
467 0 : Ok(certs) => {
468 0 : for c in certs {
469 0 : b = b.add_root_certificate(c);
470 0 : }
471 : }
472 : // once at startup; this crate carries no tracing dep
473 0 : Err(e) => eprintln!("ANTARES_EXTRA_CA_FILE {path}: not a PEM bundle ({e})"),
474 : },
475 0 : Err(e) => eprintln!("ANTARES_EXTRA_CA_FILE {path}: unreadable ({e})"),
476 : }
477 11650 : }
478 11650 : b
479 11650 : }
480 :
481 : /// wasm32: the browser owns TLS trust, redirects and name resolution —
482 : /// reqwest's wasm `ClientBuilder` exposes none of those knobs, and the page's
483 : /// CORS sandbox is the egress boundary. The policy still gates URLs via
484 : /// `check_host` before any fetch.
485 : #[cfg(target_arch = "wasm32")]
486 : pub fn client_builder(_policy: EgressPolicy) -> reqwest::ClientBuilder {
487 : reqwest::Client::builder()
488 : }
489 :
490 : /// Client timeouts are native knobs; the browser's fetch has no
491 : /// client-level equivalent, so on wasm32 this is a no-op by design.
492 : #[cfg(not(target_arch = "wasm32"))]
493 11632 : pub fn with_timeouts(
494 11632 : b: reqwest::ClientBuilder,
495 11632 : connect: std::time::Duration,
496 11632 : total: std::time::Duration,
497 11632 : ) -> reqwest::ClientBuilder {
498 11632 : b.connect_timeout(connect).timeout(total)
499 11632 : }
500 :
501 : #[cfg(target_arch = "wasm32")]
502 : #[allow(missing_docs)] // documented on the native arm above
503 : pub fn with_timeouts(
504 : b: reqwest::ClientBuilder,
505 : _connect: std::time::Duration,
506 : _total: std::time::Duration,
507 : ) -> reqwest::ClientBuilder {
508 : b
509 : }
510 :
511 : /// How long a context fetch waits before its one retry. A connection that
512 : /// fails at once — a resolver or a route not answering yet, the first
513 : /// outbound request of a freshly started container — tends to fail again at
514 : /// once, and recovers within a fraction of a second rather than within the
515 : /// microseconds an immediate retry leaves it. The pause is paid only after a
516 : /// connection already carried no response.
517 : const RETRY_PAUSE: std::time::Duration = std::time::Duration::from_millis(500);
518 :
519 : /// Wait `d` on whichever timer the target has.
520 : #[cfg(not(target_arch = "wasm32"))]
521 22 : async fn pause(d: std::time::Duration) {
522 22 : tokio::time::sleep(d).await;
523 22 : }
524 :
525 : #[cfg(target_arch = "wasm32")]
526 : async fn pause(d: std::time::Duration) {
527 : gloo_timers::future::TimeoutFuture::new(u32::try_from(d.as_millis()).unwrap_or(u32::MAX)).await;
528 : }
529 :
530 : /// Hard wall-clock bound for an outbound interaction. Native clients carry
531 : /// their timeouts at construction, so this passes through; on wasm32 the
532 : /// browser fetch has no client-level timeout (and reqwest's AbortController
533 : /// timer does not arm inside a dedicated worker), so an unresolved fetch
534 : /// would pend FOREVER — and a pending context fetch holds its resolve
535 : /// permit, which eventually stalls ALL context resolution (a stopped context
536 : /// server once froze a whole Robot run for over an hour this way, every later
537 : /// fetch queued behind the leaked permits). `None` = deadline exceeded.
538 : #[cfg(not(target_arch = "wasm32"))]
539 5125 : pub async fn io_deadline<T>(fut: impl std::future::Future<Output = T>, _ms: u32) -> Option<T> {
540 5125 : Some(fut.await)
541 5047 : }
542 :
543 : #[cfg(target_arch = "wasm32")]
544 : #[allow(missing_docs)] // documented on the native arm above
545 : pub async fn io_deadline<T>(fut: impl std::future::Future<Output = T>, ms: u32) -> Option<T> {
546 : use futures_util::future::{select, Either};
547 : use futures_util::pin_mut;
548 : let t = gloo_timers::future::TimeoutFuture::new(ms);
549 : pin_mut!(fut);
550 : match select(fut, t).await {
551 : Either::Left((v, _)) => Some(v),
552 : Either::Right(_) => None,
553 : }
554 : }
555 :
556 : /// Ceiling on a header-supplied cache lifetime: one year. The values are
557 : /// remote input; unclamped, `Instant::now() + duration` overflows (and
558 : /// panics) on a hostile max-age or Expires.
559 : const MAX_CONTEXT_TTL: std::time::Duration = std::time::Duration::from_secs(31_536_000);
560 :
561 : /// 6.3.16: cache lifetime of a downloaded @context comes from its response
562 : /// headers. `None` = no explicit lifetime (cache until evicted/reloaded).
563 238 : fn ttl_from_headers(
564 238 : cache_control: Option<&str>,
565 238 : expires: Option<&str>,
566 238 : ) -> Option<std::time::Duration> {
567 238 : if let Some(cc) = cache_control {
568 36 : let cc = cc.to_ascii_lowercase();
569 36 : if cc.contains("no-store") || cc.contains("no-cache") {
570 2 : return Some(std::time::Duration::ZERO);
571 34 : }
572 : // 6.3.16: "a max-age or s-maxage response directive"; the broker is a
573 : // shared cache, so s-maxage takes precedence when both are present
574 : // (RFC 7234 5.2.2.9).
575 64 : for prefix in ["s-maxage=", "max-age="] {
576 64 : if let Some(v) = cc
577 64 : .split(',')
578 72 : .filter_map(|d| d.trim().strip_prefix(prefix))
579 64 : .next()
580 : {
581 34 : if let Ok(secs) = v.trim().parse::<u64>() {
582 34 : return Some(std::time::Duration::from_secs(secs).min(MAX_CONTEXT_TTL));
583 0 : }
584 30 : }
585 : }
586 202 : }
587 202 : if let Some(exp) = expires {
588 : // HTTP-date (RFC 7231); an unparsable or past Expires means "stale".
589 6 : let when = chrono::DateTime::parse_from_rfc2822(exp).ok()?;
590 6 : let delta = when.with_timezone(&chrono::Utc) - chrono::Utc::now();
591 6 : return Some(
592 6 : delta
593 6 : .to_std()
594 6 : .unwrap_or(std::time::Duration::ZERO)
595 6 : .min(MAX_CONTEXT_TTL),
596 6 : );
597 196 : }
598 196 : None
599 238 : }
600 :
601 : /// @context responses above this size are refused.
602 : pub(crate) const MAX_CONTEXT_BYTES: usize = 5 * 1024 * 1024;
603 :
604 : /// Cap on usage-registry entries (client-supplied URLs); past it, adding a
605 : /// new URL evicts the least recently used entry.
606 : const MAX_USAGE_ENTRIES: usize = 4096;
607 :
608 : /// Fetch-count cap per @context resolution — a hostile context tree must
609 : /// not turn one request into an unbounded crawl. Checked BEFORE each
610 : /// fetch, so at most this many URLs are ever contacted. Public because
611 : /// `/q/health` publishes the caps a request runs under, and a second
612 : /// constant carrying the same number is one that can drift from the one
613 : /// actually enforced.
614 : pub const MAX_CONTEXT_URLS: usize = 32;
615 :
616 : /// The merged-context cache is keyed by the SERIALIZED user @context, which
617 : /// an `application/ld+json` body may carry inline up to the body cap — 256
618 : /// multi-megabyte keys (plus the term maps built from them) are no memory
619 : /// bound at all. Past this length the merge is simply not cached: no network
620 : /// is involved, the fetched documents stay warm, and the attacker's lever
621 : /// disappears.
622 : const MAX_MERGED_KEY_BYTES: usize = 8 * 1024;
623 :
624 : /// Byte budget of the fetched-document cache. Entry count alone is not a
625 : /// memory bound when one entry may be MAX_CONTEXT_BYTES.
626 : const MAX_FETCHED_CACHE_BYTES: u64 = 16 * 1024 * 1024;
627 :
628 : /// Entry ceiling, charged through the byte budget: every entry costs at
629 : /// least MAX_FETCHED_CACHE_BYTES/MAX_FETCHED_ENTRIES of it, so a flood of
630 : /// tiny documents cannot turn a byte budget into an unbounded map.
631 : const MAX_FETCHED_ENTRIES: u64 = 256;
632 :
633 : /// Terms the merged-context cache may hold across all its entries. A merged
634 : /// Context costs memory in proportion to its term map (the map itself, its
635 : /// compaction inverse and its prefix index), and an @context document may
636 : /// spend its 5 MiB budget on hundreds of thousands of short mappings — so an
637 : /// entry ceiling alone bounds nothing. 256 entries of an ordinary vocabulary
638 : /// (the core context defines 184 terms; a large domain one runs to tens of
639 : /// thousands) fit inside this comfortably.
640 : const MAX_MERGED_CACHE_TERMS: u64 = 2_000_000;
641 :
642 : /// Entry ceiling of the merged-context cache, charged through the term
643 : /// budget exactly as the fetched cache charges its byte budget.
644 : const MAX_MERGED_ENTRIES: u64 = 256;
645 :
646 : /// The merged-context cache, bounded by TERMS held rather than by entries.
647 : #[cfg(not(target_arch = "wasm32"))]
648 5270 : fn merged_cache() -> BoundedCache<String, Arc<Context>> {
649 5270 : let floor = (MAX_MERGED_CACHE_TERMS / MAX_MERGED_ENTRIES) as u32;
650 5270 : BoundedCache::builder()
651 5270 : .max_capacity(MAX_MERGED_CACHE_TERMS)
652 5270 : .weigher(move |_key: &String, ctx: &Arc<Context>| {
653 550 : u32::try_from(ctx.term_count())
654 550 : .unwrap_or(u32::MAX)
655 550 : .max(floor)
656 550 : })
657 5270 : .build()
658 5270 : }
659 :
660 : /// wasm32: the FIFO minicache carries no weigher, and a browser tab's
661 : /// @context set is tiny — the entry bound is the bound there.
662 : #[cfg(target_arch = "wasm32")]
663 : fn merged_cache() -> BoundedCache<String, Arc<Context>> {
664 : BoundedCache::new(MAX_MERGED_ENTRIES)
665 : }
666 :
667 : /// The fetched-document cache, bounded by BYTES: the weight of an entry is
668 : /// the size of the document it holds (floored, see above).
669 : #[cfg(not(target_arch = "wasm32"))]
670 5268 : fn fetched_cache() -> BoundedCache<String, FetchedDoc> {
671 : // what one entry costs of the budget at minimum
672 5268 : let floor = (MAX_FETCHED_CACHE_BYTES / MAX_FETCHED_ENTRIES) as u32;
673 5268 : BoundedCache::builder()
674 5268 : .max_capacity(MAX_FETCHED_CACHE_BYTES)
675 5268 : .weigher(move |_url: &String, doc: &FetchedDoc| {
676 1328 : u32::try_from(doc.value.to_string().len())
677 1328 : .unwrap_or(u32::MAX)
678 1328 : .max(floor)
679 1328 : })
680 5268 : .build()
681 5268 : }
682 :
683 : /// wasm32: the FIFO minicache carries no weigher, and a browser tab's
684 : /// @context set is tiny — the entry bound is the bound there.
685 : #[cfg(target_arch = "wasm32")]
686 : fn fetched_cache() -> BoundedCache<String, FetchedDoc> {
687 : BoundedCache::new(MAX_FETCHED_ENTRIES)
688 : }
689 :
690 : #[derive(Clone)]
691 : struct FetchedDoc {
692 : value: Arc<Value>,
693 : /// 6.3.16 expiry deadline; `None` = cache until evicted.
694 : stale_at: Option<Instant>,
695 : /// The Tenant this document belongs to, for the locally stored kinds
696 : /// (5.13.1 "Hosted": "@contexts that are explicitly added by users";
697 : /// "ImplicitlyCreated": created as a side effect of an operation). 5.5.10:
698 : /// "If a Tenant is specified for an NGSI-LD operation, the operation
699 : /// shall only be applied to information related to the specified
700 : /// Tenant" — so those mappings expand that Tenant's payloads only.
701 : /// `None` = a "Cached" copy of a document the broker downloaded from a
702 : /// public URL, which belongs to no Tenant and is shared by all of them.
703 : owner: Option<TenantId>,
704 : }
705 :
706 : impl FetchedDoc {
707 : /// Does this document resolve for `tenant`? A document owned by no Tenant
708 : /// resolves for everyone; one owned by a Tenant resolves for that Tenant
709 : /// alone, and for a resolution with no Tenant in scope it does not resolve
710 : /// at all. The store answers the same question about the row this document
711 : /// was read from (ADR-0021), so a `None` that saw every owner here would
712 : /// be a rule the two layers disagree on.
713 222 : fn serves(&self, tenant: Option<&TenantId>) -> bool {
714 222 : match (&self.owner, tenant) {
715 112 : (None, _) => true,
716 0 : (Some(_), None) => false,
717 110 : (Some(owner), Some(t)) => owner == t,
718 : }
719 222 : }
720 :
721 : /// 6.3.16 lifetime reached.
722 170 : fn is_stale(&self) -> bool {
723 170 : self.stale_at.is_some_and(|t| Instant::now() >= t)
724 170 : }
725 : }
726 :
727 : /// The @context loader: fetches, caches and merges @contexts under the
728 : /// egress policy, with the core context pinned.
729 : pub struct Loader {
730 : http: HttpClient,
731 : policy: EgressPolicy,
732 : /// URL → parsed `@context` member of the fetched document (+ 6.3.16 TTL
733 : /// and, for locally stored @contexts, the owning Tenant).
734 : /// Bounded LRU — every cache has a max size.
735 : fetched: BoundedCache<String, FetchedDoc>,
736 : /// cache key (serialized user context) → merged+frozen Context (the
737 : /// parsed-context LRU — the centerpiece, size-capped at 256).
738 : merged: BoundedCache<String, Arc<Context>>,
739 : /// Core context, pre-merged and PINNED outside the LRU (never evicted).
740 : core_only: Arc<Context>,
741 : /// URL → usage stats for every external @context referenced by requests
742 : /// (5.13 Cached-entry bookkeeping). Client-supplied URLs must never grow
743 : /// state without limit: capped at `MAX_USAGE_ENTRIES`, and admitting a
744 : /// new URL past the cap evicts the entry with the oldest lastUsage.
745 : usage: RwLock<HashMap<String, CtxUsage>>,
746 : /// merged-cache key → every URL that resolution touched (so cache hits
747 : /// still bump numberOfHits for nested references).
748 : merged_urls: BoundedCache<String, Arc<Vec<String>>>,
749 : /// Bounded concurrency on cold context FETCHES (one permit per network
750 : /// fetch, not per resolution) — a burst of exotic-context requests can't
751 : /// blow the JSON working-set budget.
752 : resolve_permits: tokio::sync::Semaphore,
753 : /// Write-through: freshly fetched remote contexts are handed to this
754 : /// hook (the broker persists them as kind='Cached' rows) so the cache
755 : /// survives a restart. Set once at wiring; None in tests.
756 : cache_writer: std::sync::RwLock<Option<CacheWriter>>,
757 : /// Shared-store hit counter: bump the persisted row on every counted
758 : /// use; a missing row reports a cross-instance delete. `None` in
759 : /// compositions without a store (bare loader tests).
760 : usage_bump: std::sync::RwLock<Option<UsageBump>>,
761 : /// Store-backed lookup for the @contexts this broker hosts. Set once at
762 : /// wiring; None in tests.
763 : local_lookup: std::sync::RwLock<Option<LocalLookup>>,
764 : }
765 :
766 : /// Request header marking a broker-internal @context fetch (this loader
767 : /// resolving a URL, possibly through the fleet's own LB). The serve endpoint
768 : /// skips its serve-hit bump for these — the resolving instance counts the
769 : /// use itself (5.13.3.5, one client use = one hit).
770 : pub const INTERNAL_FETCH_HEADER: &str = "x-antares-ctx-fetch";
771 :
772 : /// What a store-backed hook hands back. Boxed because the store behind it
773 : /// is asynchronous and the hook is held as a trait object.
774 : pub type HookFuture<'a, T> = std::pin::Pin<Box<dyn std::future::Future<Output = T> + Send + 'a>>;
775 :
776 : /// (the Tenant the resolution acts for, url, parsed `@context` value) —
777 : /// called on every fresh remote fetch. The row it writes is `Cached` and so
778 : /// belongs to no Tenant, but reaching the store to write it is a call the
779 : /// store answers per Tenant (ADR-0021).
780 : pub type CacheWriter = std::sync::Arc<
781 : dyn for<'a> Fn(Option<&'a TenantId>, &'a str, &'a Value) -> HookFuture<'a, ()> + Send + Sync,
782 : >;
783 : /// (Tenant, url) -> "the shared row still exists" (after bumping its hit
784 : /// counter). A Hosted row is only bumped by the Tenant that owns it, which
785 : /// is also the only Tenant whose resolutions reach it.
786 : pub type UsageBump = std::sync::Arc<
787 : dyn for<'a> Fn(Option<&'a TenantId>, &'a str) -> HookFuture<'a, bool> + Send + Sync,
788 : >;
789 : /// (Tenant, url) -> the `@context` value of the row this broker HOSTS under
790 : /// it, with the Tenant that owns it (5.13.1 Hosted/ImplicitlyCreated;
791 : /// `None` = owned by no Tenant). The store is where a broker-local @context
792 : /// comes from — see [`Loader::set_local_lookup`].
793 : pub type LocalLookup = std::sync::Arc<
794 : dyn for<'a> Fn(
795 : Option<&'a TenantId>,
796 : &'a str,
797 : ) -> HookFuture<'a, Option<(Option<TenantId>, Value)>>
798 : + Send
799 : + Sync,
800 : >;
801 :
802 : impl Default for Loader {
803 0 : fn default() -> Self {
804 0 : Self::new()
805 0 : }
806 : }
807 :
808 : impl Loader {
809 : /// A loader with the policy from the environment and default timeouts.
810 5238 : pub fn new() -> Self {
811 5238 : Self::with_policy(EgressPolicy::from_env())
812 5238 : }
813 :
814 : /// A loader with an explicit policy over a freshly built HTTP client.
815 : #[allow(clippy::expect_used)]
816 5268 : pub fn with_policy(policy: EgressPolicy) -> Self {
817 : // TLS backend initialisation is the only failure reqwest reports
818 : // here; without it no @context can ever be fetched.
819 5268 : let client = with_timeouts(
820 5268 : client_builder(policy),
821 5268 : std::time::Duration::from_secs(5 * slow_factor()),
822 5268 : std::time::Duration::from_secs(10 * slow_factor()),
823 : )
824 5268 : .build()
825 5268 : .expect("reqwest client");
826 5268 : Self::with_client(policy, client)
827 5268 : }
828 : }
829 :
830 : /// The pinned core `@context`, parsed and frozen, with no loader, client or
831 : /// cache behind it: the value every `Loader` pins, and what a test of
832 : /// expansion or matching needs on its own.
833 : #[allow(clippy::expect_used)]
834 5436 : pub fn core_context() -> Context {
835 5436 : let mut core = Context::default();
836 : // PINNED holds CORE_CONTEXT and every entry parses: pinned by
837 : // `every_pinned_context_parses_and_carries_an_at_context`
838 5436 : merge_context_value(&mut core, &pinned(CORE_CONTEXT).expect("pinned core"));
839 5436 : core.freeze();
840 5436 : core.source = Value::String(CORE_CONTEXT.to_owned());
841 5436 : core
842 5436 : }
843 :
844 : impl Loader {
845 : /// A loader over the caller's own HTTP client — a gateway with its own
846 : /// proxy, allowlist or TLS setup fetches @contexts through it. `policy`
847 : /// still clears every URL before the fetch (`check_host`), but the
848 : /// transport belongs to the caller: the DNS pin (`PolicyResolver`) and
849 : /// the per-hop redirect cap live in `client_builder`, so a client built
850 : /// any other way resolves names unfiltered and follows redirects under
851 : /// reqwest's own default. Start from `client_builder(policy)` and add
852 : /// the timeouts, proxy and trust anchors to it to keep both. Every
853 : /// cache is per instance, nothing here is process-global.
854 5268 : pub fn with_client(policy: EgressPolicy, client: reqwest::Client) -> Self {
855 5268 : let core = core_context();
856 5268 : Self {
857 5268 : http: wrap_client(client),
858 5268 : policy,
859 5268 : fetched: fetched_cache(),
860 5268 : merged: merged_cache(),
861 5268 : core_only: Arc::new(core),
862 5268 : usage: RwLock::new(HashMap::new()),
863 5268 : merged_urls: BoundedCache::new(MAX_MERGED_ENTRIES),
864 5268 : resolve_permits: tokio::sync::Semaphore::new(32),
865 5268 : cache_writer: std::sync::RwLock::new(None),
866 5268 : usage_bump: std::sync::RwLock::new(None),
867 5268 : local_lookup: std::sync::RwLock::new(None),
868 5268 : }
869 5268 : }
870 :
871 : /// Install the hook that persists a fetched @context (url, document).
872 3182 : pub fn set_cache_writer(&self, w: CacheWriter) {
873 3182 : *self
874 3182 : .cache_writer
875 3182 : .write()
876 3182 : .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(w);
877 3182 : }
878 :
879 : /// Wire the store as the source of the @contexts this broker HOSTS
880 : /// (5.13.1 Hosted and ImplicitlyCreated). The in-process copy those
881 : /// resources leave behind is a CACHE: it is lost on a restart (only
882 : /// `Cached` rows are preloaded) and evictable from the bounded document
883 : /// cache at any time. Without this hook such a miss becomes an outbound
884 : /// GET of a URL the broker minted from a request's `Host` header — a
885 : /// client can then name the host its own @context is fetched from, and
886 : /// with it the term mappings that expand its Tenant's payloads. The row
887 : /// carries the owning Tenant, so 5.5.10 still decides who resolves it.
888 3182 : pub fn set_local_lookup(&self, f: LocalLookup) {
889 3182 : *self
890 3182 : .local_lookup
891 3182 : .write()
892 3182 : .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(f);
893 3182 : }
894 :
895 : /// Wire the shared-store usage bump (5.13.3.5): called on every
896 : /// counted use of a URL. Returns whether the shared row still exists —
897 : /// `false` means another instance deleted the @context, and this
898 : /// instance must drop its warm copies so the delete is honoured here.
899 3182 : pub fn set_usage_bump(&self, f: UsageBump) {
900 3182 : *self
901 3182 : .usage_bump
902 3182 : .write()
903 3182 : .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(f);
904 3182 : }
905 :
906 : /// Boot preload: re-seed a Cached entry persisted by the writer —
907 : /// the parsed doc goes into the fetch cache, the bookkeeping identity
908 : /// (localId/createdAt) into the usage registry, so 5.13 listings look
909 : /// the same across a restart.
910 0 : pub async fn seed_cached(&self, url: &str, local_id: &str, created_at: &str, ctx_value: Value) {
911 0 : self.fetched.insert(
912 0 : url.to_owned(),
913 0 : FetchedDoc {
914 0 : value: Arc::new(ctx_value),
915 0 : stale_at: None,
916 0 : // 5.13.1 "Cached": a copy of a public document, no Tenant
917 0 : owner: None,
918 0 : },
919 : );
920 0 : self.usage.write().await.insert(
921 0 : url.to_owned(),
922 0 : CtxUsage {
923 0 : url: url.to_owned(),
924 0 : local_id: local_id.to_owned(),
925 0 : created_at: created_at.to_owned(),
926 0 : last_usage: created_at.to_owned(),
927 0 : hits: 0,
928 0 : },
929 : );
930 0 : }
931 :
932 9790 : fn now() -> String {
933 9790 : chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
934 9790 : }
935 :
936 : /// Bump usage stats (numberOfHits / lastUsage, 5.13.3.5) for one URL —
937 : /// in this instance's registry AND, via the usage_bump hook, in the
938 : /// shared store row (per-instance counters split-brain behind a
939 : /// load balancer). Returns true when the hook reported the row GONE
940 : /// (deleted through another instance): local copies are evicted so the
941 : /// next resolution refetches and re-creates the entry.
942 9790 : pub async fn bump_url(&self, tenant: Option<&TenantId>, url: &str) -> bool {
943 9790 : let now = Self::now();
944 : {
945 9790 : let mut map = self.usage.write().await;
946 9790 : if let Some(u) = map.get_mut(url) {
947 1358 : u.hits += 1;
948 1358 : u.last_usage = now;
949 1358 : } else {
950 : // hold the size bound: admitting a new URL past the cap
951 : // evicts the entry with the oldest lastUsage (RFC 3339
952 : // strings order chronologically)
953 8432 : if map.len() >= MAX_USAGE_ENTRIES {
954 8 : if let Some(oldest) = map
955 8 : .values()
956 32760 : .min_by(|a, b| a.last_usage.cmp(&b.last_usage))
957 8 : .map(|u| u.url.clone())
958 8 : {
959 8 : map.remove(&oldest);
960 8 : }
961 8424 : }
962 8432 : map.insert(
963 8432 : url.to_owned(),
964 8432 : CtxUsage {
965 8432 : url: url.to_owned(),
966 8432 : // deterministic (uuid5 of the URL): the same identity
967 8432 : // names this entry in the usage registry, the persisted
968 8432 : // Cached row (write-through) and across restarts — an
969 8432 : // API delete can therefore always find the row (5.13.5).
970 8432 : local_id: uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_URL, url.as_bytes())
971 8432 : .to_string(),
972 8432 : created_at: now.clone(),
973 8432 : last_usage: now,
974 8432 : hits: 1,
975 8432 : },
976 8432 : );
977 : }
978 : }
979 : // the hook is cloned out of the lock before it is awaited: a
980 : // std guard held across an await is not Send
981 9790 : let bump = self
982 9790 : .usage_bump
983 9790 : .read()
984 9790 : .unwrap_or_else(std::sync::PoisonError::into_inner)
985 9790 : .clone();
986 9790 : let row_exists = match bump {
987 1542 : Some(f) => f(tenant, url).await,
988 8248 : None => true,
989 : };
990 9790 : if row_exists {
991 9786 : return false;
992 4 : }
993 4 : self.usage.write().await.remove(url);
994 4 : self.evict(url).await;
995 4 : true
996 9790 : }
997 :
998 : /// Usage entries of every external @context referenced so far.
999 100 : pub async fn usage_list(&self) -> Vec<CtxUsage> {
1000 100 : self.usage.read().await.values().cloned().collect()
1001 100 : }
1002 :
1003 : /// Find a usage entry by original URL or by its generated localId.
1004 12 : pub async fn usage_get(&self, id: &str) -> Option<CtxUsage> {
1005 12 : let map = self.usage.read().await;
1006 12 : map.get(id)
1007 12 : .or_else(|| map.values().find(|u| u.local_id == id))
1008 12 : .cloned()
1009 12 : }
1010 :
1011 : /// Drop a URL's usage entry and evict it from the caches.
1012 204 : pub async fn usage_remove(&self, url: &str) {
1013 204 : self.usage.write().await.remove(url);
1014 204 : self.evict(url).await;
1015 204 : }
1016 :
1017 : /// 5.13.5.4 Delete and Reload: re-download a Cached @context from its
1018 : /// original URL, replacing the stored copy only on success. Any error —
1019 : /// download failure or invalid content per 5.5.4 — is
1020 : /// LdContextNotAvailable and "the operation ends without removing the
1021 : /// existing @context".
1022 22 : pub async fn refetch(&self, url: &str) -> Result<(), NgsiError> {
1023 22 : let old = self.fetched.get(url);
1024 22 : self.fetched.invalidate(url); // force a network fetch
1025 22 : match self.fetch(url, None).await {
1026 : Ok(_) => {
1027 : // merged contexts built on the old copy are stale
1028 6 : self.invalidate_merged_using(url);
1029 6 : Ok(())
1030 : }
1031 16 : Err(e) => {
1032 16 : if let Some(old) = old {
1033 16 : self.fetched.insert(url.to_owned(), old);
1034 16 : }
1035 16 : Err(match e {
1036 2 : NgsiError::BadRequestData(m) => NgsiError::LdContextNotAvailable(m),
1037 14 : other => other,
1038 : })
1039 : }
1040 : }
1041 22 : }
1042 :
1043 : /// Core-only context (no user @context supplied).
1044 27790 : pub fn core(&self) -> Arc<Context> {
1045 27790 : Arc::clone(&self.core_only)
1046 27790 : }
1047 :
1048 : /// Resolve a user-supplied `@context` value (string URL, object, or array)
1049 : /// into a merged Context with the core context merged last. No Tenant in
1050 : /// scope: locally stored @contexts of every Tenant resolve, so a
1051 : /// resolution serving a request must use `resolve_for` instead (5.5.10).
1052 84 : pub async fn resolve(&self, user: &Value) -> Result<Arc<Context>, NgsiError> {
1053 84 : self.resolve_counted(None, user, true).await
1054 84 : }
1055 :
1056 : /// Resolve WITHOUT counting usage hits — for broker-internal resolutions
1057 : /// (notification building), which are not client @context usage (053_08).
1058 6 : pub async fn resolve_quiet(&self, user: &Value) -> Result<Arc<Context>, NgsiError> {
1059 6 : self.resolve_counted(None, user, false).await
1060 6 : }
1061 :
1062 : /// 5.5.10: resolve within one Tenant — "the operation shall only be
1063 : /// applied to information related to the specified Tenant", so an
1064 : /// @context another Tenant stored locally (5.13.1) does not resolve here.
1065 5588 : pub async fn resolve_for(
1066 5588 : &self,
1067 5588 : tenant: &TenantId,
1068 5588 : user: &Value,
1069 5588 : ) -> Result<Arc<Context>, NgsiError> {
1070 5588 : self.resolve_counted(Some(tenant), user, true).await
1071 5588 : }
1072 :
1073 : /// `resolve_for` without counting usage hits.
1074 695 : pub async fn resolve_quiet_for(
1075 695 : &self,
1076 695 : tenant: &TenantId,
1077 695 : user: &Value,
1078 695 : ) -> Result<Arc<Context>, NgsiError> {
1079 695 : self.resolve_counted(Some(tenant), user, false).await
1080 695 : }
1081 :
1082 6373 : async fn resolve_counted(
1083 6373 : &self,
1084 6373 : tenant: Option<&TenantId>,
1085 6373 : user: &Value,
1086 6373 : count: bool,
1087 6373 : ) -> Result<Arc<Context>, NgsiError> {
1088 6373 : let key = user.to_string();
1089 : // urls already counted on the merged-hit path — the fallthrough
1090 : // rebuild below must not bump them a second time.
1091 6373 : let mut counted: Vec<String> = Vec::new();
1092 6373 : if let Some(hit) = self
1093 6373 : .merged
1094 6373 : .get(&key)
1095 6373 : .filter(|_| self.merged_hit_is_usable(&key, tenant))
1096 : {
1097 1703 : if count {
1098 : // cache hit: bump every URL this context resolution involves.
1099 : // A bump that finds the shared row GONE means another
1100 : // instance deleted this @context — do NOT serve the warm
1101 : // copy; fall through and rebuild (refetch re-creates it).
1102 1263 : let urls = self.merged_urls.get(&key);
1103 1263 : let mut deleted_elsewhere = false;
1104 1299 : for url in urls.iter().flat_map(|u| u.iter()) {
1105 1299 : if self.bump_url(tenant, url).await {
1106 4 : deleted_elsewhere = true;
1107 1295 : } else {
1108 1295 : counted.push(url.clone());
1109 1295 : }
1110 : }
1111 1263 : if !deleted_elsewhere {
1112 1259 : return Ok(hit);
1113 4 : }
1114 : } else {
1115 440 : return Ok(hit);
1116 : }
1117 4670 : }
1118 4674 : let mut ctx = Context::default();
1119 4674 : let urls = std::sync::Mutex::new(Vec::new());
1120 4674 : self.merge_entry(&mut ctx, user, 0, &urls, None, tenant)
1121 4674 : .await?;
1122 546 : let urls = urls.into_inner().unwrap_or_default();
1123 546 : if count {
1124 285 : for url in &urls {
1125 249 : if counted.contains(url) {
1126 0 : continue; // already bumped on the merged-hit path
1127 249 : }
1128 : // only after successful resolution. A bump that reports the
1129 : // shared row GONE (deleted through another instance) while
1130 : // the fetch above was served from this instance's warm doc
1131 : // cache means the write-through never ran — a counted use
1132 : // re-creates the entry (5.13.5.4): refetch (bump_url just
1133 : // evicted the warm copy) so the row exists, then count on it.
1134 249 : if self.bump_url(tenant, url).await && self.fetch(url, tenant).await.is_ok() {
1135 0 : let _ = self.bump_url(tenant, url).await;
1136 249 : }
1137 : }
1138 261 : }
1139 : // Core context last: its (protected) terms win — CIM 009 4.4.
1140 : // CORE_CONTEXT is a PINNED entry and every entry parses: pinned by
1141 : // `every_pinned_context_parses_and_carries_an_at_context`.
1142 : #[allow(clippy::expect_used)]
1143 546 : let core = pinned(CORE_CONTEXT).expect("pinned core");
1144 546 : merge_context_value(&mut ctx, &core);
1145 546 : ctx.freeze();
1146 546 : ctx.source = user.clone();
1147 546 : let arc = Arc::new(ctx);
1148 546 : if key.len() <= MAX_MERGED_KEY_BYTES {
1149 530 : self.merged_urls.insert(key.clone(), Arc::new(urls));
1150 530 : self.merged.insert(key, Arc::clone(&arc));
1151 530 : }
1152 546 : Ok(arc)
1153 6373 : }
1154 :
1155 : /// A merged-context cache hit is usable only while every document it was
1156 : /// built from is still fresh (6.3.16: "implementations shall periodically
1157 : /// invalidate the "Cached" @contexts according to the headers mentioned
1158 : /// above" — a merged context is only as fresh as its sources) and still
1159 : /// resolves for this Tenant (5.5.10): the cache is keyed by the user
1160 : /// @context alone, which two Tenants can send verbatim, so a merge built
1161 : /// from one Tenant's locally stored @context must not be handed to
1162 : /// another.
1163 1727 : fn merged_hit_is_usable(&self, key: &str, tenant: Option<&TenantId>) -> bool {
1164 1727 : match self.merged_urls.get(key) {
1165 : // the documents behind this entry are unknown: it can be shown
1166 : // neither fresh nor in-Tenant, so it is rebuilt
1167 0 : None => false,
1168 1749 : Some(urls) => urls.iter().all(|url| match self.fetched.get(url) {
1169 92 : Some(doc) => doc.serves(tenant) && !doc.is_stale(),
1170 : // Dropped from the document cache (it is a bounded LRU, and
1171 : // a locally stored @context is as evictable as any other):
1172 : // its OWNER is now unknown, and unknown is not public, so the
1173 : // entry is rebuilt — a rebuild re-reads the ownership and
1174 : // refuses what this Tenant may not have. A pinned core
1175 : // context has no cache entry by design and belongs to no
1176 : // Tenant, so it never forces one.
1177 1657 : None => Self::is_pinned_core(url),
1178 1749 : }),
1179 : }
1180 1727 : }
1181 :
1182 8782 : fn merge_entry<'a>(
1183 8782 : &'a self,
1184 8782 : ctx: &'a mut Context,
1185 8782 : entry: &'a Value,
1186 8782 : depth: usize,
1187 8782 : urls: &'a std::sync::Mutex<Vec<String>>,
1188 8782 : // JSON-LD 1.1 (section 3.1): a relative context IRI inside a fetched context
1189 8782 : // document resolves against THAT document's URL. The ETSI compound
1190 8782 : // context references "ngsi-ld-test-suite.jsonld" relatively — without
1191 8782 : // this every request using it dies with LdContextNotAvailable.
1192 8782 : base: Option<std::sync::Arc<String>>,
1193 8782 : tenant: Option<&'a TenantId>,
1194 8782 : ) -> BoxFut<'a, Result<(), NgsiError>> {
1195 8782 : Box::pin(async move {
1196 : // 5.5.6: an @context that "is invalid" is BadRequestData; 504
1197 : // LdContextNotAvailable is reserved for one that "is not
1198 : // available". Both caps below are reached from client-supplied
1199 : // structure alone, so a client must not be able to mint gateway
1200 : // errors on demand.
1201 8782 : if depth > 8 {
1202 2 : return Err(NgsiError::BadRequestData(
1203 2 : "@context nesting too deep".into(),
1204 2 : ));
1205 8780 : }
1206 8780 : match entry {
1207 3356 : Value::Array(items) => {
1208 3484 : for item in items {
1209 3484 : self.merge_entry(ctx, item, depth + 1, urls, base.clone(), tenant)
1210 3484 : .await?;
1211 : }
1212 24 : Ok(())
1213 : }
1214 4746 : Value::String(url) => {
1215 4746 : let resolved: String =
1216 4746 : if url.starts_with("http://") || url.starts_with("https://") {
1217 4742 : url.clone()
1218 4 : } else if let Some(b) = base.as_deref() {
1219 0 : reqwest::Url::parse(b)
1220 0 : .and_then(|b| b.join(url))
1221 0 : .map(String::from)
1222 0 : .map_err(|e| {
1223 0 : NgsiError::LdContextNotAvailable(format!(
1224 0 : "cannot resolve @context URL {url} against {b}: {e}"
1225 0 : ))
1226 0 : })?
1227 : } else {
1228 4 : url.clone()
1229 : };
1230 : // Cap enforced before the network is touched: once the
1231 : // resolution has already fetched MAX_CONTEXT_URLS
1232 : // documents, the next reference fails instead of
1233 : // extending the crawl. Poisoned lock fails closed.
1234 4746 : let fetched_so_far = urls.lock().map(|u| u.len()).unwrap_or(usize::MAX);
1235 4746 : if fetched_so_far >= MAX_CONTEXT_URLS {
1236 2 : return Err(NgsiError::BadRequestData(format!(
1237 2 : "@context resolution exceeds {MAX_CONTEXT_URLS} referenced URLs"
1238 2 : )));
1239 4744 : }
1240 4744 : let doc = self.fetch(&resolved, tenant).await?;
1241 624 : if let Ok(mut u) = urls.lock() {
1242 624 : u.push(resolved.clone());
1243 624 : }
1244 624 : self.merge_entry(
1245 624 : ctx,
1246 624 : &doc,
1247 624 : depth + 1,
1248 624 : urls,
1249 624 : Some(std::sync::Arc::new(resolved)),
1250 624 : tenant,
1251 624 : )
1252 624 : .await
1253 : }
1254 : // 5.5.7's Scoped Context prohibition binds the user
1255 : // @context. A document fetched from a pinned core URL is
1256 : // a Core @context, not client input, so it merges under
1257 : // the Core rule; everything else is the user's.
1258 678 : Value::Object(obj) => match base.as_deref() {
1259 606 : Some(b) if Self::is_pinned_core(b) => ctx.merge_core_object(obj),
1260 362 : _ => ctx.merge_object(obj),
1261 : },
1262 0 : Value::Null => Ok(()),
1263 0 : _ => Err(NgsiError::BadRequestData("invalid @context entry".into())),
1264 : }
1265 8782 : })
1266 8782 : }
1267 :
1268 : /// Is `url` one of the built-in (pinned) core context URLs?
1269 4399 : pub fn is_pinned_core(url: &str) -> bool {
1270 4399 : pinned(url).is_some()
1271 4399 : }
1272 :
1273 : /// Fetch a remote context document, returning its `@context` member.
1274 : /// 5.13.1: Cached @contexts are invalidated per the protocol's explicit
1275 : /// expiration indications — cache hits honour the 6.3.16 lifetime, stale
1276 : /// entries are re-fetched, and a changed body invalidates the
1277 : /// merged-context cache.
1278 4766 : async fn fetch(&self, url: &str, tenant: Option<&TenantId>) -> Result<Arc<Value>, NgsiError> {
1279 4766 : if let Some(v) = pinned(url) {
1280 320 : return Ok(Arc::new(v));
1281 4446 : }
1282 4446 : let err = |m: String| NgsiError::LdContextNotAvailable(m);
1283 4446 : let mut stale_value: Option<Arc<Value>> = None;
1284 4446 : if let Some(hit) = self.fetched.get(url) {
1285 118 : if !hit.serves(tenant) {
1286 : // 5.5.10 + 5.13.1: this URL names an @context another Tenant
1287 : // stored locally, so for this Tenant it does not exist — and
1288 : // fetching it would only reach the same entry back through
1289 : // the broker's own (Tenant-gated) serve endpoint.
1290 20 : return Err(err(format!("@context {url} is not available")));
1291 98 : }
1292 98 : if hit.is_stale() {
1293 2 : stale_value = Some(Arc::clone(&hit.value));
1294 2 : } else {
1295 96 : return Ok(hit.value);
1296 : }
1297 4328 : }
1298 : // A URL this broker hosts is served from its row, never fetched:
1299 : // the warm copy is only a cache (see `set_local_lookup`).
1300 4330 : let lookup = self
1301 4330 : .local_lookup
1302 4330 : .read()
1303 4330 : .unwrap_or_else(std::sync::PoisonError::into_inner)
1304 4330 : .clone();
1305 4330 : let hosted = match lookup {
1306 4238 : Some(f) => f(tenant, url).await,
1307 92 : None => None,
1308 : };
1309 4330 : if let Some((owner, value)) = hosted {
1310 12 : let doc = FetchedDoc {
1311 12 : value: Arc::new(value),
1312 12 : stale_at: None, // hosted locally: no 6.3.16 lifetime
1313 12 : owner,
1314 12 : };
1315 12 : if !doc.serves(tenant) {
1316 0 : return Err(err(format!("@context {url} is not available")));
1317 12 : }
1318 12 : let arc = Arc::clone(&doc.value);
1319 12 : self.fetched.insert(url.to_owned(), doc);
1320 12 : return Ok(arc);
1321 4318 : }
1322 4318 : if !url.starts_with("http://") && !url.starts_with("https://") {
1323 4 : return Err(err(format!("unsupported @context URL: {url}")));
1324 4314 : }
1325 : // SSRF hook: deny private destinations unless configured.
1326 4314 : let parsed = reqwest::Url::parse(url).map_err(|e| err(format!("bad URL {url}: {e}")))?;
1327 4314 : let host = parsed.host_str().unwrap_or_default().to_owned();
1328 4314 : let port = parsed.port_or_known_default().unwrap_or(443);
1329 4314 : self.policy
1330 4314 : .check_host(&host, port)
1331 4314 : .await
1332 4314 : .map_err(|e| err(format!("fetching {url}: {e}")))?;
1333 : // Bounded concurrency on cold fetching. The permit covers this ONE
1334 : // network fetch and is released before the crawl recurses into the
1335 : // document's own references — held across a whole recursive
1336 : // resolution instead, a handful of slow context trees would stall
1337 : // every cold resolution in the process.
1338 : // `acquire` fails only on a closed semaphore, which nothing closes.
1339 : // If that ever changed, proceeding without the permit costs the
1340 : // concurrency bound and not the fetch, so it is not worth a panic.
1341 4314 : let _permit = self.resolve_permits.acquire().await.ok();
1342 : // The whole HTTP interaction is one Send unit (http_interaction);
1343 : // only Send data (ttl + bytes) crosses back out.
1344 4336 : let send = || {
1345 4336 : self.http
1346 4336 : .get(url)
1347 4336 : .header("Accept", "application/ld+json, application/json")
1348 : // marks this as a broker-internal context resolution: the
1349 : // serving instance must not add a serve-hit on top of this
1350 : // instance's own bump_url (053_08 fleet double-count)
1351 4336 : .header(INTERNAL_FETCH_HEADER, "1")
1352 4336 : .send()
1353 4336 : };
1354 4314 : let interact = async {
1355 : // 5.5.6 turns an @context that "is not available" into
1356 : // LdContextNotAvailable, and one connection that carried no
1357 : // response has not established that: a refused or dropped
1358 : // connection is asked once more, after `RETRY_PAUSE`. Only the
1359 : // send is repeated — a response that did arrive is the answer
1360 : // whatever its status, and a read that failed part-way already
1361 : // cost the bytes. The client timeout and the redirect cap are
1362 : // answers in themselves and are not retried, and both attempts
1363 : // share the deadline below.
1364 4314 : let resp = match send().await {
1365 4292 : Ok(r) => r,
1366 22 : Err(e) if e.is_timeout() || e.is_redirect() => {
1367 0 : return Err(err(format!("fetching {url}: {e}")));
1368 : }
1369 : Err(_) => {
1370 22 : pause(RETRY_PAUSE).await;
1371 22 : send()
1372 22 : .await
1373 22 : .map_err(|e| err(format!("fetching {url}: {e}")))?
1374 : }
1375 : };
1376 4296 : if !resp.status().is_success() {
1377 4078 : return Err(err(format!("fetching {url}: HTTP {}", resp.status())));
1378 218 : }
1379 218 : let ttl = ttl_from_headers(
1380 218 : resp.headers()
1381 218 : .get("cache-control")
1382 218 : .and_then(|v| v.to_str().ok()),
1383 218 : resp.headers().get("expires").and_then(|v| v.to_str().ok()),
1384 : );
1385 : // Bounded response size (504 LdContextNotAvailable on breach).
1386 218 : if resp
1387 218 : .content_length()
1388 218 : .is_some_and(|l| l as usize > MAX_CONTEXT_BYTES)
1389 : {
1390 0 : return Err(err(format!("{url}: @context document too large")));
1391 218 : }
1392 : // A declared Content-Length is advisory only — a chunked body
1393 : // has none. Natively the body is accumulated chunk by chunk and
1394 : // refused the moment it would pass the cap, so an oversized
1395 : // response is never buffered in full first.
1396 : #[cfg(not(target_arch = "wasm32"))]
1397 216 : let bytes = {
1398 218 : let mut resp = resp;
1399 218 : let mut buf: Vec<u8> = Vec::new();
1400 716 : while let Some(chunk) = resp
1401 716 : .chunk()
1402 716 : .await
1403 716 : .map_err(|e| err(format!("reading {url}: {e}")))?
1404 : {
1405 500 : if buf.len() + chunk.len() > MAX_CONTEXT_BYTES {
1406 2 : return Err(err(format!("{url}: @context document too large")));
1407 498 : }
1408 498 : buf.extend_from_slice(&chunk);
1409 : }
1410 216 : buf
1411 : };
1412 : // wasm: the browser fetch hands over the body whole; the
1413 : // post-read size check below still applies.
1414 : #[cfg(target_arch = "wasm32")]
1415 : let bytes = resp
1416 : .bytes()
1417 : .await
1418 : .map_err(|e| err(format!("reading {url}: {e}")))?
1419 : .to_vec();
1420 216 : Ok((ttl, bytes))
1421 4314 : };
1422 4314 : let (ttl, bytes) = http_interaction(async {
1423 4314 : match io_deadline(interact, 10_000).await {
1424 4314 : Some(r) => r,
1425 0 : None => Err(err(format!("fetching {url}: deadline exceeded"))),
1426 : }
1427 4314 : })
1428 4314 : .await?;
1429 216 : if bytes.len() > MAX_CONTEXT_BYTES {
1430 0 : return Err(err(format!("{url}: @context document too large")));
1431 216 : }
1432 : // 5.5.6: unavailability is LdContextNotAvailable, but a RETRIEVED
1433 : // remote @context whose content is invalid is BadRequestData.
1434 216 : let doc: Value = serde_json::from_slice(&bytes)
1435 216 : .map_err(|e| NgsiError::BadRequestData(format!("{url} is not a JSON document: {e}")))?;
1436 210 : let ctx_val = doc
1437 210 : .get("@context")
1438 210 : .cloned()
1439 210 : .ok_or_else(|| NgsiError::BadRequestData(format!("{url} has no @context member")))?;
1440 202 : let arc = Arc::new(ctx_val);
1441 202 : if stale_value.is_some_and(|old| *old != *arc) {
1442 2 : // Refreshed content differs: merged contexts built on the old
1443 2 : // copy are invalid.
1444 2 : self.invalidate_merged_using(url);
1445 200 : }
1446 202 : self.fetched.insert(
1447 202 : url.to_owned(),
1448 : FetchedDoc {
1449 202 : value: Arc::clone(&arc),
1450 202 : stale_at: ttl.map(|d| Instant::now() + d),
1451 : // 5.13.1 "Cached": downloaded from a public URL, no Tenant
1452 202 : owner: None,
1453 : },
1454 : );
1455 : // Write-through: persist what was just fetched.
1456 202 : let writer = self
1457 202 : .cache_writer
1458 202 : .read()
1459 202 : .unwrap_or_else(std::sync::PoisonError::into_inner)
1460 202 : .clone();
1461 202 : if let Some(w) = writer {
1462 126 : w(tenant, url, &arc).await;
1463 76 : }
1464 202 : Ok(arc)
1465 4766 : }
1466 :
1467 : /// Insert a locally-hosted context (jsonldContexts API) so later
1468 : /// resolutions of `url` need no network round-trip. Bound to no Tenant:
1469 : /// every resolution sees it.
1470 882 : pub async fn put_local(&self, url: String, context_value: Value) {
1471 882 : self.insert_local(None, url, context_value)
1472 882 : }
1473 :
1474 : /// The same for an @context a client stored THROUGH a Tenant (5.13.1
1475 : /// "Hosted"/"ImplicitlyCreated"): per 5.5.10 those mappings apply to that
1476 : /// Tenant's operations only, so another Tenant naming the same URL
1477 : /// resolves nothing.
1478 216 : pub async fn put_local_for(&self, tenant: &TenantId, url: String, context_value: Value) {
1479 216 : self.insert_local(Some(tenant.clone()), url, context_value)
1480 216 : }
1481 :
1482 1098 : fn insert_local(&self, owner: Option<TenantId>, url: String, context_value: Value) {
1483 1098 : self.fetched.insert(
1484 1098 : url.clone(),
1485 1098 : FetchedDoc {
1486 1098 : value: Arc::new(context_value),
1487 1098 : stale_at: None, // hosted locally: no 6.3.16 lifetime
1488 1098 : owner,
1489 1098 : },
1490 : );
1491 1098 : self.invalidate_merged_using(&url);
1492 1098 : }
1493 :
1494 : /// Drop the merged contexts built from `url` — and only those. One added,
1495 : /// reloaded or deleted @context must not throw away every Tenant's
1496 : /// parsed contexts and make them all re-fetch the world; a merge whose
1497 : /// sources are unknown is dropped, since it cannot be shown unaffected.
1498 1314 : fn invalidate_merged_using(&self, url: &str) {
1499 1314 : let stale: Vec<String> = self
1500 1314 : .merged
1501 1314 : .iter()
1502 1904 : .filter(|(key, _)| match self.merged_urls.get(key.as_str()) {
1503 2560 : Some(urls) => urls.iter().any(|u| u == url),
1504 0 : None => true,
1505 1580 : })
1506 1314 : .map(|(key, _)| (*key).clone())
1507 1314 : .collect();
1508 1314 : for key in stale {
1509 40 : self.merged.invalidate(&key);
1510 40 : self.merged_urls.invalidate(&key);
1511 40 : }
1512 1314 : }
1513 :
1514 : /// Cache occupancy (entries per cache). Feeds /q/health, and is what
1515 : /// the security regression tests assert the cache size caps against.
1516 12 : pub fn cache_stats(&self) -> serde_json::Value {
1517 12 : self.fetched.run_pending_tasks();
1518 12 : self.merged.run_pending_tasks();
1519 12 : self.merged_urls.run_pending_tasks();
1520 12 : serde_json::json!({
1521 12 : "fetched": self.fetched.entry_count(),
1522 12 : "merged": self.merged.entry_count(),
1523 12 : "mergedUrls": self.merged_urls.entry_count(),
1524 : })
1525 12 : }
1526 :
1527 : /// Drop the fetched document for `url` and every merged context using it.
1528 208 : pub async fn evict(&self, url: &str) {
1529 208 : self.fetched.invalidate(url);
1530 208 : self.invalidate_merged_using(url);
1531 208 : }
1532 : }
1533 :
1534 : /// The `@context` of a compiled-in document, or `None` for a URL that is not
1535 : /// pinned. The bodies are `include_str!`-ed at build time, so a body that
1536 : /// does not parse or carries no `@context` is a build defect rather than
1537 : /// anything a request can cause —
1538 : /// `every_pinned_context_parses_and_carries_an_at_context` fails on it.
1539 15167 : fn pinned(url: &str) -> Option<Value> {
1540 105397 : let (_, body) = PINNED.iter().find(|(u, _)| *u == url)?;
1541 9851 : let doc: Value = serde_json::from_str(body).ok()?;
1542 9851 : doc.get("@context").cloned()
1543 15167 : }
1544 :
1545 : /// Merge a pre-parsed Core @context value (4.4: merged last, so it wins).
1546 5982 : fn merge_context_value(ctx: &mut Context, v: &Value) {
1547 5982 : match v {
1548 5982 : Value::Object(o) => {
1549 5982 : let _ = ctx.merge_core_object(o);
1550 5982 : }
1551 0 : Value::Array(items) => {
1552 0 : for i in items {
1553 0 : merge_context_value(ctx, i);
1554 0 : }
1555 : }
1556 0 : _ => {}
1557 : }
1558 5982 : }
1559 :
1560 : #[cfg(test)]
1561 : mod tests {
1562 : use super::*;
1563 :
1564 : /// `core_context` and the merge tail unwrap `pinned(CORE_CONTEXT)`,
1565 : /// because a broker without the Core @context cannot expand anything and
1566 : /// has no useful degraded mode. That is only safe while the compiled-in
1567 : /// table really does carry a parseable `@context` for every URL in it —
1568 : /// which is a property of the build, so it is checked here rather than
1569 : /// left to a comment. A pinned document that stops parsing fails this
1570 : /// test instead of the first request after deploy.
1571 : #[test]
1572 2 : fn every_pinned_context_parses_and_carries_an_at_context() {
1573 2 : assert!(!PINNED.is_empty(), "the pinned table is empty");
1574 14 : for (url, _) in PINNED {
1575 14 : assert!(
1576 14 : pinned(url).is_some(),
1577 : "pinned document {url} does not parse, or carries no @context"
1578 : );
1579 : }
1580 2 : let core = pinned(CORE_CONTEXT).expect("the Core @context is pinned");
1581 2 : assert!(
1582 2 : core.is_object() || core.is_array() || core.is_string(),
1583 : "the Core @context is not a usable @context value: {core}"
1584 : );
1585 2 : assert!(
1586 14 : PINNED.iter().any(|(u, _)| *u == CORE_CONTEXT),
1587 : "CORE_CONTEXT must be in the pinned table by URL, not merely resolvable"
1588 : );
1589 2 : }
1590 :
1591 : /// 4.4: "the Core @context is protected and shall remain immutable and
1592 : /// invariant during expansion or compaction of terms. […] implementations
1593 : /// shall consider the Core @context as if it were in the last position of
1594 : /// the @context array." A user context redefining a core term must not
1595 : /// win, while its own new terms still apply.
1596 : #[tokio::test]
1597 2 : async fn core_terms_are_protected_from_user_redefinition() {
1598 2 : let loader = Loader::new();
1599 2 : let user = serde_json::json!({
1600 2 : "Property": "https://evil.example/Property",
1601 2 : "observedAt": "https://evil.example/observedAt",
1602 2 : "speed": "https://example.org/speed"
1603 : });
1604 2 : let ctx = loader.resolve(&user).await.expect("resolve");
1605 2 : assert_eq!(
1606 2 : ctx.expand_key("Property"),
1607 : "https://uri.etsi.org/ngsi-ld/Property"
1608 : );
1609 2 : assert_eq!(
1610 2 : ctx.expand_key("observedAt"),
1611 : "https://uri.etsi.org/ngsi-ld/observedAt"
1612 : );
1613 2 : assert_eq!(ctx.expand_key("speed"), "https://example.org/speed");
1614 2 : }
1615 :
1616 : /// The resolver is the enforcement point, so a name that
1617 : /// resolves into a private range must fail at DNS time — that is what
1618 : /// makes a rebinding answer between check and connect harmless.
1619 : #[tokio::test]
1620 2 : async fn policy_resolver_filters_private_answers() {
1621 : use reqwest::dns::Resolve;
1622 : use std::str::FromStr;
1623 2 : let deny = PolicyResolver(EgressPolicy {
1624 2 : allow_private: false,
1625 2 : });
1626 2 : let name = reqwest::dns::Name::from_str("localhost").expect("name");
1627 2 : assert!(deny.resolve(name).await.is_err());
1628 :
1629 2 : let allow = PolicyResolver(EgressPolicy {
1630 2 : allow_private: true,
1631 2 : });
1632 2 : let name = reqwest::dns::Name::from_str("localhost").expect("name");
1633 2 : let addrs = allow.resolve(name).await.expect("allowed");
1634 2 : assert!(addrs.count() > 0);
1635 2 : }
1636 :
1637 : /// 5.5.6: "When a remote JSON-LD @context referenced by an incoming
1638 : /// request is not available … LdContextNotAvailable. If the remote
1639 : /// JSON-LD @context is invalid … BadRequestData." Unreachable → 503/504
1640 : /// class; fetched-but-invalid content (not JSON, or no @context member)
1641 : /// → BadRequestData.
1642 : #[tokio::test]
1643 2 : async fn clause_5_5_6_unavailable_vs_invalid_remote_context() {
1644 4 : let serve = |body: &'static str| async move {
1645 4 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
1646 4 : .await
1647 4 : .expect("bind");
1648 4 : let addr = listener.local_addr().expect("addr");
1649 4 : tokio::spawn(async move {
1650 8 : while let Ok((mut sock, _)) = listener.accept().await {
1651 : use tokio::io::AsyncWriteExt;
1652 4 : let resp = format!(
1653 : "HTTP/1.1 200 OK\r\nContent-Type: application/ld+json\r\nContent-Length: {}\r\n\r\n{body}",
1654 4 : body.len()
1655 : );
1656 4 : let _ = sock.write_all(resp.as_bytes()).await;
1657 4 : let _ = sock.flush().await;
1658 : }
1659 0 : });
1660 4 : addr
1661 8 : };
1662 2 : let loader = Loader::with_policy(EgressPolicy {
1663 2 : allow_private: true,
1664 2 : });
1665 : // unreachable → LdContextNotAvailable
1666 2 : let err = loader
1667 2 : .resolve(&Value::String("http://127.0.0.1:9/ctx.jsonld".into()))
1668 2 : .await
1669 2 : .expect_err("unreachable context");
1670 2 : assert!(
1671 2 : matches!(err, NgsiError::LdContextNotAvailable(_)),
1672 : "unavailable → LdContextNotAvailable, got {err:?}"
1673 : );
1674 : // served but not a JSON document → BadRequestData
1675 2 : let addr = serve("this is { not json").await;
1676 2 : let err = loader
1677 2 : .resolve(&Value::String(format!("http://{addr}/ctx.jsonld")))
1678 2 : .await
1679 2 : .expect_err("non-JSON context document");
1680 2 : assert!(
1681 2 : matches!(err, NgsiError::BadRequestData(_)),
1682 : "invalid (non-JSON) → BadRequestData, got {err:?}"
1683 : );
1684 : // served JSON without an @context member → BadRequestData
1685 2 : let addr = serve(r#"{"note": "no context here"}"#).await;
1686 2 : let err = loader
1687 2 : .resolve(&Value::String(format!("http://{addr}/ctx.jsonld")))
1688 2 : .await
1689 2 : .expect_err("JSON without @context member");
1690 2 : assert!(
1691 2 : matches!(err, NgsiError::BadRequestData(_)),
1692 2 : "invalid (no @context member) → BadRequestData, got {err:?}"
1693 2 : );
1694 2 : }
1695 :
1696 : /// reqwest is compiled provider-less, and a client built while no rustls
1697 : /// crypto provider is installed does not error — it PANICS inside
1698 : /// `build()`. The one constructor installs ring first, so every client in
1699 : /// the broker is buildable; a feature change that drops the install fails
1700 : /// here instead of at the first outbound request in production.
1701 : #[test]
1702 2 : fn a_client_builds_with_no_crypto_provider_installed_beforehand() {
1703 2 : assert!(
1704 2 : client_builder(EgressPolicy {
1705 2 : allow_private: true,
1706 2 : })
1707 2 : .build()
1708 2 : .is_ok(),
1709 : "the outbound-client constructor must install a crypto provider"
1710 : );
1711 2 : }
1712 :
1713 : /// The redirect cap is only real if it is installed on the client an open
1714 : /// redirector actually talks to — so bounce one against a server that
1715 : /// always redirects to itself and assert the client gives up.
1716 : #[tokio::test]
1717 2 : async fn client_builder_caps_redirects() {
1718 2 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
1719 2 : .await
1720 2 : .expect("bind");
1721 2 : let addr = listener.local_addr().expect("addr");
1722 2 : let hops = Arc::new(std::sync::atomic::AtomicUsize::new(0));
1723 2 : let seen = hops.clone();
1724 2 : tokio::spawn(async move {
1725 : loop {
1726 10 : let Ok((mut sock, _)) = listener.accept().await else {
1727 0 : return;
1728 : };
1729 8 : seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
1730 8 : let resp = format!(
1731 : "HTTP/1.1 302 Found\r\nLocation: http://{addr}/loop\r\nContent-Length: 0\r\n\r\n"
1732 : );
1733 : use tokio::io::AsyncWriteExt;
1734 8 : let _ = sock.write_all(resp.as_bytes()).await;
1735 8 : let _ = sock.flush().await;
1736 : }
1737 0 : });
1738 :
1739 2 : let client = client_builder(EgressPolicy {
1740 2 : allow_private: true,
1741 2 : })
1742 2 : .timeout(std::time::Duration::from_secs(5))
1743 2 : .build()
1744 2 : .expect("client");
1745 2 : let err = client
1746 2 : .get(format!("http://{addr}/start"))
1747 2 : .send()
1748 2 : .await
1749 2 : .expect_err("redirect loop must not be followed forever");
1750 2 : assert!(err.is_redirect(), "gave up for the redirect-cap reason");
1751 2 : assert_eq!(
1752 2 : hops.load(std::sync::atomic::Ordering::SeqCst),
1753 2 : MAX_REDIRECTS + 1,
1754 2 : "one initial request plus MAX_REDIRECTS hops"
1755 2 : );
1756 2 : }
1757 :
1758 : // SSRF: a redirect to a private IP LITERAL is refused per hop even
1759 : // though reqwest's DNS PolicyResolver never sees IP literals. With
1760 : // allow_private=false the redirect target (127.0.0.1) must not be followed.
1761 : #[tokio::test]
1762 2 : async fn redirect_to_private_ip_literal_is_blocked() {
1763 : use std::sync::atomic::{AtomicUsize, Ordering};
1764 : use std::sync::Arc;
1765 2 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
1766 2 : .await
1767 2 : .expect("bind");
1768 2 : let addr = listener.local_addr().expect("addr");
1769 2 : let hits = Arc::new(AtomicUsize::new(0));
1770 2 : let seen = hits.clone();
1771 2 : tokio::spawn(async move {
1772 4 : while let Ok((mut sock, _)) = listener.accept().await {
1773 2 : seen.fetch_add(1, Ordering::SeqCst);
1774 : // redirect to a private IP literal (self)
1775 2 : let resp = format!(
1776 : "HTTP/1.1 302 Found\r\nLocation: http://{addr}/internal\r\nContent-Length: 0\r\n\r\n"
1777 : );
1778 : use tokio::io::AsyncWriteExt;
1779 2 : let _ = sock.write_all(resp.as_bytes()).await;
1780 2 : let _ = sock.flush().await;
1781 : }
1782 0 : });
1783 2 : let client = client_builder(EgressPolicy {
1784 2 : allow_private: false,
1785 2 : })
1786 2 : .timeout(std::time::Duration::from_secs(5))
1787 2 : .build()
1788 2 : .expect("client");
1789 : // the initial IP-literal request connects (resolver never runs for it),
1790 : // gets the 302, and the policy STOPS instead of following to the private
1791 : // hop — so the server is hit exactly once and we get the 3xx back.
1792 2 : let resp = client
1793 2 : .get(format!("http://{addr}/start"))
1794 2 : .send()
1795 2 : .await
1796 2 : .expect("stop returns the 3xx, not an error");
1797 2 : assert_eq!(resp.status().as_u16(), 302, "redirect was not followed");
1798 2 : assert_eq!(
1799 2 : hits.load(Ordering::SeqCst),
1800 2 : 1,
1801 2 : "only the initial request; the private-IP hop was refused"
1802 2 : );
1803 2 : }
1804 :
1805 : // SSRF: the same refusal for the IPv6 spelling of a literal. `Url`
1806 : // returns an IPv6 host BRACKETED (`[::1]`), which `IpAddr::from_str`
1807 : // rejects, so a hop check that parses `host_str()` as it comes waves the
1808 : // whole IPv6 literal space through — including `[::ffff:169.254.169.254]`.
1809 : // The redirect target here is a second server on IPv6 loopback: it must
1810 : // never be reached.
1811 : #[tokio::test]
1812 2 : async fn redirect_to_private_ipv6_literal_is_blocked() {
1813 : use std::sync::atomic::{AtomicUsize, Ordering};
1814 : use std::sync::Arc;
1815 2 : let target = tokio::net::TcpListener::bind("[::1]:0")
1816 2 : .await
1817 2 : .expect("bind v6");
1818 2 : let target_addr = target.local_addr().expect("addr");
1819 2 : let reached = Arc::new(AtomicUsize::new(0));
1820 2 : let hit = reached.clone();
1821 2 : tokio::spawn(async move {
1822 2 : while let Ok((mut sock, _)) = target.accept().await {
1823 0 : hit.fetch_add(1, Ordering::SeqCst);
1824 : use tokio::io::AsyncWriteExt;
1825 0 : let _ = sock
1826 0 : .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
1827 0 : .await;
1828 : }
1829 0 : });
1830 :
1831 2 : let entry = tokio::net::TcpListener::bind("127.0.0.1:0")
1832 2 : .await
1833 2 : .expect("bind");
1834 2 : let entry_addr = entry.local_addr().expect("addr");
1835 2 : tokio::spawn(async move {
1836 4 : while let Ok((mut sock, _)) = entry.accept().await {
1837 2 : let resp = format!(
1838 : "HTTP/1.1 302 Found\r\nLocation: http://[::1]:{}/internal\r\nContent-Length: 0\r\n\r\n",
1839 2 : target_addr.port()
1840 : );
1841 : use tokio::io::AsyncWriteExt;
1842 2 : let _ = sock.write_all(resp.as_bytes()).await;
1843 2 : let _ = sock.flush().await;
1844 : }
1845 0 : });
1846 :
1847 2 : let client = client_builder(EgressPolicy {
1848 2 : allow_private: false,
1849 2 : })
1850 2 : .timeout(std::time::Duration::from_secs(5))
1851 2 : .build()
1852 2 : .expect("client");
1853 2 : let resp = client
1854 2 : .get(format!("http://{entry_addr}/start"))
1855 2 : .send()
1856 2 : .await
1857 2 : .expect("stop returns the 3xx, not an error");
1858 2 : assert_eq!(resp.status().as_u16(), 302, "redirect was not followed");
1859 2 : assert_eq!(
1860 2 : reached.load(Ordering::SeqCst),
1861 2 : 0,
1862 2 : "the IPv6-literal hop was followed into loopback"
1863 2 : );
1864 2 : }
1865 :
1866 : #[tokio::test]
1867 2 : async fn core_context_has_ngsi_terms() {
1868 2 : let l = Loader::new();
1869 2 : let c = l.core();
1870 2 : assert_eq!(
1871 2 : c.expand_key("location"),
1872 : "https://uri.etsi.org/ngsi-ld/location"
1873 : );
1874 2 : assert_eq!(
1875 2 : c.expand_key("unknownTerm"),
1876 : "https://uri.etsi.org/ngsi-ld/default-context/unknownTerm"
1877 : );
1878 2 : assert_eq!(
1879 2 : c.compact_iri("https://uri.etsi.org/ngsi-ld/location"),
1880 2 : "location"
1881 2 : );
1882 2 : }
1883 :
1884 : #[test]
1885 2 : fn cache_lifetime_from_headers() {
1886 : // 6.3.16: Cache-Control wins, no-store/no-cache = immediately stale,
1887 : // Expires as the fallback, neither = cache until evicted.
1888 2 : assert_eq!(
1889 2 : ttl_from_headers(Some("max-age=60"), None),
1890 2 : Some(std::time::Duration::from_secs(60))
1891 : );
1892 2 : assert_eq!(
1893 2 : ttl_from_headers(Some("public, max-age=5, immutable"), None),
1894 2 : Some(std::time::Duration::from_secs(5))
1895 : );
1896 2 : assert_eq!(
1897 2 : ttl_from_headers(Some("no-store"), None),
1898 : Some(std::time::Duration::ZERO)
1899 : );
1900 : // 6.3.16 names "a max-age or s-maxage response directive"; the broker
1901 : // is a shared cache, so s-maxage wins over max-age when both appear.
1902 2 : assert_eq!(
1903 2 : ttl_from_headers(Some("s-maxage=120"), None),
1904 2 : Some(std::time::Duration::from_secs(120))
1905 : );
1906 2 : assert_eq!(
1907 2 : ttl_from_headers(Some("max-age=60, s-maxage=120"), None),
1908 2 : Some(std::time::Duration::from_secs(120))
1909 : );
1910 2 : assert_eq!(ttl_from_headers(None, None), None);
1911 2 : let past = ttl_from_headers(None, Some("Tue, 01 Jan 2019 00:00:00 GMT"));
1912 2 : assert_eq!(
1913 : past,
1914 : Some(std::time::Duration::ZERO),
1915 : "past Expires = stale"
1916 : );
1917 2 : let future = ttl_from_headers(None, Some("Fri, 01 Jan 2100 00:00:00 GMT"));
1918 2 : assert!(future.expect("parsed") > std::time::Duration::from_secs(3600));
1919 : // Header values are remote input; an unclamped lifetime overflows
1920 : // Instant arithmetic when added to now(). Both paths clamp to a year.
1921 2 : let year = std::time::Duration::from_secs(31_536_000);
1922 2 : assert_eq!(
1923 2 : ttl_from_headers(Some("max-age=18446744073709551615"), None),
1924 2 : Some(year),
1925 : "huge max-age clamps to one year"
1926 : );
1927 2 : assert_eq!(
1928 2 : ttl_from_headers(None, Some("Fri, 01 Jan 2100 00:00:00 GMT")),
1929 2 : Some(year),
1930 : "far-future Expires clamps to one year"
1931 : );
1932 2 : }
1933 :
1934 : #[tokio::test]
1935 2 : async fn egress_policy_denies_private_ranges() {
1936 2 : let deny = EgressPolicy {
1937 2 : allow_private: false,
1938 2 : };
1939 22 : for host in [
1940 2 : "127.0.0.1",
1941 2 : "10.1.2.3",
1942 2 : "192.168.0.9",
1943 2 : "172.16.5.5",
1944 2 : "169.254.169.254",
1945 2 : "localhost",
1946 2 : "::1",
1947 2 : "0.0.0.0",
1948 2 : // IPv4-mapped IPv6 forms of private targets must not slip
1949 2 : // past the v6 arm — same destinations, different spelling.
1950 2 : "::ffff:127.0.0.1",
1951 2 : "::ffff:169.254.169.254",
1952 2 : "::ffff:10.1.2.3",
1953 2 : ] {
1954 22 : assert!(
1955 22 : deny.check_host(host, 80).await.is_err(),
1956 : "{host} must be denied"
1957 : );
1958 : }
1959 2 : assert!(
1960 2 : deny.check_host("93.184.216.34", 443).await.is_ok(),
1961 : "public IP allowed"
1962 : );
1963 2 : assert!(
1964 2 : deny.check_host("::ffff:8.8.8.8", 443).await.is_ok(),
1965 : "IPv4-mapped public IP allowed"
1966 : );
1967 2 : let allow = EgressPolicy {
1968 2 : allow_private: true,
1969 2 : };
1970 2 : assert!(allow.check_host("127.0.0.1", 80).await.is_ok());
1971 : // ...but the instance-metadata range is refused even then: allowing
1972 : // private egress is a development convenience, handing out cloud
1973 : // credentials is not part of it.
1974 8 : for host in [
1975 2 : "169.254.169.254",
1976 2 : "169.254.170.2",
1977 2 : "::ffff:169.254.169.254",
1978 2 : "::169.254.169.254",
1979 2 : ] {
1980 8 : let err = allow
1981 8 : .check_host(host, 80)
1982 8 : .await
1983 8 : .expect_err("{host} must be denied whatever the private-egress setting");
1984 8 : assert!(err.contains("metadata"), "{host}: {err}");
1985 2 : }
1986 2 : }
1987 :
1988 : /// 5.13.5.4 Delete and Reload: on reload the broker re-downloads BEFORE
1989 : /// removing — a failed or invalid download raises LdContextNotAvailable
1990 : /// and "the operation ends without removing the existing @context"; a
1991 : /// successful download replaces it.
1992 : #[tokio::test]
1993 2 : async fn clause_5_13_5_4_reload_keeps_existing_on_failure_replaces_on_success() {
1994 : // switchable mock: Some(body) → 200 with that body, None → 500
1995 2 : let body = Arc::new(std::sync::Mutex::new(Some(
1996 2 : r#"{"@context":{"speed":"https://a.example/speed"}}"#.to_string(),
1997 2 : )));
1998 2 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
1999 2 : .await
2000 2 : .expect("bind");
2001 2 : let addr = listener.local_addr().expect("addr");
2002 2 : let served = body.clone();
2003 2 : tokio::spawn(async move {
2004 10 : while let Ok((mut sock, _)) = listener.accept().await {
2005 : use tokio::io::AsyncWriteExt;
2006 8 : let b = served.lock().expect("lock").clone();
2007 8 : let resp = match b {
2008 6 : Some(b) => format!(
2009 : "HTTP/1.1 200 OK\r\nContent-Type: application/ld+json\r\nContent-Length: {}\r\n\r\n{b}",
2010 6 : b.len()
2011 : ),
2012 2 : None => "HTTP/1.1 500 Internal Server Error\r\nContent-Length: 0\r\n\r\n"
2013 2 : .to_string(),
2014 : };
2015 8 : let _ = sock.write_all(resp.as_bytes()).await;
2016 : }
2017 0 : });
2018 2 : let url = format!("http://{addr}/ctx.jsonld");
2019 2 : let loader = Loader::with_policy(EgressPolicy {
2020 2 : allow_private: true,
2021 2 : });
2022 2 : let ctx = loader
2023 2 : .resolve(&Value::String(url.clone()))
2024 2 : .await
2025 2 : .expect("initial fetch");
2026 2 : assert_eq!(ctx.expand_key("speed"), "https://a.example/speed");
2027 :
2028 : // download fails → LdContextNotAvailable, the existing copy stays
2029 : // usable even through a fresh (uncached) resolution shape
2030 2 : *body.lock().expect("lock") = None;
2031 2 : let err = loader.refetch(&url).await.expect_err("failed reload");
2032 2 : assert!(
2033 2 : matches!(err, NgsiError::LdContextNotAvailable(_)),
2034 : "download failure → LdContextNotAvailable, got {err:?}"
2035 : );
2036 2 : let ctx = loader
2037 2 : .resolve(&serde_json::json!([url.clone()]))
2038 2 : .await
2039 2 : .expect("existing copy kept after failed reload");
2040 2 : assert_eq!(ctx.expand_key("speed"), "https://a.example/speed");
2041 :
2042 : // invalid content → LdContextNotAvailable (5.13.5.4 — not the 5.5.6
2043 : // BadRequestData used outside reload), existing copy kept
2044 2 : *body.lock().expect("lock") = Some(r#"{"note":"no @context member"}"#.to_string());
2045 2 : let err = loader.refetch(&url).await.expect_err("invalid reload");
2046 2 : assert!(
2047 2 : matches!(err, NgsiError::LdContextNotAvailable(_)),
2048 : "invalid content → LdContextNotAvailable, got {err:?}"
2049 : );
2050 2 : let ctx = loader
2051 2 : .resolve(&Value::String(url.clone()))
2052 2 : .await
2053 2 : .expect("existing copy kept after invalid reload");
2054 2 : assert_eq!(ctx.expand_key("speed"), "https://a.example/speed");
2055 :
2056 : // success → "the existing @context is replaced with the newly
2057 : // downloaded one"
2058 2 : *body.lock().expect("lock") =
2059 2 : Some(r#"{"@context":{"speed":"https://b.example/speed"}}"#.to_string());
2060 2 : loader.refetch(&url).await.expect("successful reload");
2061 2 : let ctx = loader
2062 2 : .resolve(&Value::String(url))
2063 2 : .await
2064 2 : .expect("resolve after reload");
2065 2 : assert_eq!(ctx.expand_key("speed"), "https://b.example/speed");
2066 2 : }
2067 :
2068 : /// The response-size cap must trip WHILE the body is being read, not
2069 : /// after the whole thing was buffered: a chunked response (no
2070 : /// Content-Length) that never terminates would otherwise be
2071 : /// accumulated in full before the check. The server here streams 8 MiB
2072 : /// and closes without the final 0-chunk — the resolve must fail with
2073 : /// the size error (cap hit mid-read), never with a read/decode error.
2074 : #[tokio::test]
2075 2 : async fn oversized_chunked_context_is_refused_at_the_cap() {
2076 2 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
2077 2 : .await
2078 2 : .expect("bind");
2079 2 : let addr = listener.local_addr().expect("addr");
2080 2 : tokio::spawn(async move {
2081 2 : while let Ok((mut sock, _)) = listener.accept().await {
2082 : use tokio::io::{AsyncReadExt, AsyncWriteExt};
2083 : // drain the request head: closing a socket with unread data
2084 : // pending sends RST, which discards body bytes the client
2085 : // has not consumed yet and turns the test nondeterministic
2086 2 : let mut reqbuf = vec![0u8; 4096];
2087 2 : let _ = sock.read(&mut reqbuf).await;
2088 2 : if sock
2089 2 : .write_all(
2090 2 : b"HTTP/1.1 200 OK\r\nContent-Type: application/ld+json\r\nTransfer-Encoding: chunked\r\n\r\n",
2091 : )
2092 2 : .await
2093 2 : .is_err()
2094 : {
2095 0 : continue;
2096 2 : }
2097 2 : let chunk = vec![b'x'; 64 * 1024];
2098 2 : let head = format!("{:x}\r\n", chunk.len());
2099 2 : for _ in 0..128 {
2100 : // stop early once the client hangs up (cap tripped)
2101 224 : if sock.write_all(head.as_bytes()).await.is_err()
2102 224 : || sock.write_all(&chunk).await.is_err()
2103 222 : || sock.write_all(b"\r\n").await.is_err()
2104 : {
2105 0 : break;
2106 222 : }
2107 : }
2108 : // no terminating 0-chunk: the connection just closes
2109 : }
2110 0 : });
2111 2 : let loader = Loader::with_policy(EgressPolicy {
2112 2 : allow_private: true,
2113 2 : });
2114 2 : let err = loader
2115 2 : .resolve(&Value::String(format!("http://{addr}/big.jsonld")))
2116 2 : .await
2117 2 : .expect_err("oversized chunked body must be refused");
2118 2 : let msg = format!("{err:?}");
2119 2 : assert!(
2120 2 : msg.contains("too large"),
2121 2 : "cap must fire during the read, got {msg}"
2122 2 : );
2123 2 : }
2124 :
2125 : /// 5.5.6: an @context is fetched again when the first connection carried
2126 : /// no response at all. "Not available" is a statement about the document,
2127 : /// and a connection dropped before a status line has not established it —
2128 : /// the server here closes the first connection without writing a byte and
2129 : /// answers the second, so the resolve returns the document. Exactly one
2130 : /// extra attempt: a server that is really down must not be hammered.
2131 : #[tokio::test]
2132 2 : async fn a_dropped_connection_is_retried_before_the_context_is_unavailable() {
2133 : use std::sync::atomic::{AtomicUsize, Ordering};
2134 2 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
2135 2 : .await
2136 2 : .expect("bind");
2137 2 : let addr = listener.local_addr().expect("addr");
2138 2 : let hits = Arc::new(AtomicUsize::new(0));
2139 2 : let seen = hits.clone();
2140 2 : tokio::spawn(async move {
2141 6 : while let Ok((mut sock, _)) = listener.accept().await {
2142 : use tokio::io::{AsyncReadExt, AsyncWriteExt};
2143 4 : if seen.fetch_add(1, Ordering::SeqCst) == 0 {
2144 : // close before any response: reqwest reports a send
2145 : // error, the shape a dropped connection takes
2146 2 : drop(sock);
2147 2 : continue;
2148 2 : }
2149 2 : let mut buf = vec![0u8; 4096];
2150 2 : let _ = sock.read(&mut buf).await;
2151 2 : let body = r#"{"@context":{"a":"https://example.org/a"}}"#;
2152 2 : let resp = format!(
2153 : "HTTP/1.1 200 OK\r\nContent-Type: application/ld+json\r\nConnection: close\r\nContent-Length: {}\r\n\r\n{body}",
2154 2 : body.len()
2155 : );
2156 2 : let _ = sock.write_all(resp.as_bytes()).await;
2157 2 : let _ = sock.flush().await;
2158 : }
2159 0 : });
2160 2 : let loader = Loader::with_policy(EgressPolicy {
2161 2 : allow_private: true,
2162 2 : });
2163 2 : loader
2164 2 : .resolve(&Value::String(format!("http://{addr}/c.jsonld")))
2165 2 : .await
2166 2 : .expect("one dropped connection is not an unavailable @context");
2167 2 : assert_eq!(
2168 2 : hits.load(Ordering::SeqCst),
2169 2 : 2,
2170 2 : "the drop is retried once and no more"
2171 2 : );
2172 2 : }
2173 :
2174 : /// A connection that fails at once tends to fail again at once: a
2175 : /// resolver or a route that is not answering yet recovers within a
2176 : /// fraction of a second, not within the microseconds an immediate retry
2177 : /// leaves it. The second attempt therefore starts no sooner than
2178 : /// `RETRY_PAUSE` after the first connection was dropped.
2179 : #[tokio::test]
2180 2 : async fn the_second_attempt_waits_before_it_connects() {
2181 2 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
2182 2 : .await
2183 2 : .expect("bind");
2184 2 : let addr = listener.local_addr().expect("addr");
2185 2 : let accepted = Arc::new(std::sync::Mutex::new(Vec::<std::time::Instant>::new()));
2186 2 : let seen = accepted.clone();
2187 2 : tokio::spawn(async move {
2188 6 : while let Ok((mut sock, _)) = listener.accept().await {
2189 : use tokio::io::{AsyncReadExt, AsyncWriteExt};
2190 4 : let first = {
2191 4 : let mut at = seen.lock().expect("lock");
2192 4 : at.push(std::time::Instant::now());
2193 4 : at.len() == 1
2194 : };
2195 4 : if first {
2196 2 : drop(sock);
2197 2 : continue;
2198 2 : }
2199 2 : let mut buf = vec![0u8; 4096];
2200 2 : let _ = sock.read(&mut buf).await;
2201 2 : let body = r#"{"@context":{"a":"https://example.org/a"}}"#;
2202 2 : let resp = format!(
2203 : "HTTP/1.1 200 OK\r\nContent-Type: application/ld+json\r\nConnection: close\r\nContent-Length: {}\r\n\r\n{body}",
2204 2 : body.len()
2205 : );
2206 2 : let _ = sock.write_all(resp.as_bytes()).await;
2207 2 : let _ = sock.flush().await;
2208 : }
2209 0 : });
2210 2 : let loader = Loader::with_policy(EgressPolicy {
2211 2 : allow_private: true,
2212 2 : });
2213 2 : loader
2214 2 : .resolve(&Value::String(format!("http://{addr}/c.jsonld")))
2215 2 : .await
2216 2 : .expect("the second attempt is answered");
2217 2 : let at = accepted.lock().expect("lock").clone();
2218 2 : assert_eq!(at.len(), 2, "one retry: {at:?}");
2219 2 : assert!(
2220 2 : at[1] - at[0] >= RETRY_PAUSE,
2221 2 : "the retry connected {:?} after the drop, sooner than {RETRY_PAUSE:?}",
2222 2 : at[1] - at[0]
2223 2 : );
2224 2 : }
2225 :
2226 : /// The retry is bounded: a host that refuses every connection is reported
2227 : /// as unavailable after the second attempt, never tried a third time.
2228 : #[tokio::test]
2229 2 : async fn a_host_that_never_answers_is_unavailable_after_one_retry() {
2230 : use std::sync::atomic::{AtomicUsize, Ordering};
2231 2 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
2232 2 : .await
2233 2 : .expect("bind");
2234 2 : let addr = listener.local_addr().expect("addr");
2235 2 : let hits = Arc::new(AtomicUsize::new(0));
2236 2 : let seen = hits.clone();
2237 2 : tokio::spawn(async move {
2238 6 : while let Ok((sock, _)) = listener.accept().await {
2239 4 : seen.fetch_add(1, Ordering::SeqCst);
2240 4 : drop(sock);
2241 4 : }
2242 0 : });
2243 2 : let loader = Loader::with_policy(EgressPolicy {
2244 2 : allow_private: true,
2245 2 : });
2246 2 : let e = loader
2247 2 : .resolve(&Value::String(format!("http://{addr}/c.jsonld")))
2248 2 : .await
2249 2 : .expect_err("a host that answers nothing is unavailable");
2250 2 : assert!(
2251 2 : matches!(e, NgsiError::LdContextNotAvailable(_)),
2252 : "5.5.6: an unreachable @context is LdContextNotAvailable, got {e:?}"
2253 : );
2254 2 : assert_eq!(
2255 2 : hits.load(Ordering::SeqCst),
2256 2 : 2,
2257 2 : "two attempts in total, not a loop"
2258 2 : );
2259 2 : }
2260 :
2261 : /// The per-resolution fetch cap must stop the crawl BEFORE the network
2262 : /// is hit past the limit — a hostile context listing many siblings
2263 : /// must not trigger them all and only then be rejected.
2264 : #[tokio::test]
2265 2 : async fn fetch_cap_stops_crawl_before_the_limit_is_passed() {
2266 : use std::sync::atomic::{AtomicUsize, Ordering};
2267 2 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
2268 2 : .await
2269 2 : .expect("bind");
2270 2 : let addr = listener.local_addr().expect("addr");
2271 2 : let hits = Arc::new(AtomicUsize::new(0));
2272 2 : let seen = hits.clone();
2273 2 : tokio::spawn(async move {
2274 66 : while let Ok((mut sock, _)) = listener.accept().await {
2275 64 : seen.fetch_add(1, Ordering::SeqCst);
2276 : use tokio::io::{AsyncReadExt, AsyncWriteExt};
2277 64 : let mut buf = vec![0u8; 4096];
2278 64 : let n = sock.read(&mut buf).await.unwrap_or(0);
2279 64 : let req = String::from_utf8_lossy(&buf[..n]).into_owned();
2280 64 : let body = if req.starts_with("GET /root") {
2281 2 : let children: Vec<String> = (0..40)
2282 80 : .map(|i| format!("\"http://{addr}/c{i}.jsonld\""))
2283 2 : .collect();
2284 2 : format!("{{\"@context\":[{}]}}", children.join(","))
2285 : } else {
2286 62 : r#"{"@context":{"a":"https://example.org/a"}}"#.to_string()
2287 : };
2288 : // Connection: close → one request per connection, so the
2289 : // accept counter equals the number of fetches made.
2290 64 : let resp = format!(
2291 : "HTTP/1.1 200 OK\r\nContent-Type: application/ld+json\r\nConnection: close\r\nContent-Length: {}\r\n\r\n{body}",
2292 64 : body.len()
2293 : );
2294 64 : let _ = sock.write_all(resp.as_bytes()).await;
2295 64 : let _ = sock.flush().await;
2296 : }
2297 0 : });
2298 2 : let loader = Loader::with_policy(EgressPolicy {
2299 2 : allow_private: true,
2300 2 : });
2301 2 : let err = loader
2302 2 : .resolve(&Value::String(format!("http://{addr}/root.jsonld")))
2303 2 : .await
2304 2 : .expect_err("a 40-URL crawl must be rejected");
2305 2 : assert!(
2306 2 : matches!(err, NgsiError::BadRequestData(_)),
2307 : "5.5.6: a client-supplied cap breach is invalid input, got {err:?}"
2308 : );
2309 2 : let n = hits.load(Ordering::SeqCst);
2310 2 : assert!(n <= 33, "crawl must stop at the cap, made {n} fetches");
2311 2 : }
2312 :
2313 : /// The usage registry records client-supplied URLs — it must hold a
2314 : /// hard size bound, not grow by one entry per distinct URL forever.
2315 : #[tokio::test]
2316 2 : async fn usage_registry_is_bounded() {
2317 2 : let loader = Loader::with_policy(EgressPolicy {
2318 2 : allow_private: true,
2319 2 : });
2320 8200 : for i in 0..4100 {
2321 8200 : let _ = loader
2322 8200 : .bump_url(None, &format!("https://ctx.example/{i}.jsonld"))
2323 8200 : .await;
2324 : }
2325 2 : let list = loader.usage_list().await;
2326 2 : assert!(
2327 2 : list.len() <= 4096,
2328 : "usage registry must stay bounded, got {} entries",
2329 0 : list.len()
2330 : );
2331 : // eviction must sacrifice old entries, never the one just added
2332 2 : assert!(
2333 2181 : list.iter().any(|u| u.url.ends_with("/4099.jsonld")),
2334 2 : "the most recently used entry must survive eviction"
2335 2 : );
2336 2 : }
2337 :
2338 : /// Name resolution is on the request path and outside every client
2339 : /// timeout: a resolver that never answers must not hold the caller, and
2340 : /// the unanswered lookup must DENY rather than let the fetch through.
2341 : #[tokio::test]
2342 2 : async fn unanswered_dns_lookup_denies_instead_of_hanging() {
2343 2 : let deny = EgressPolicy {
2344 2 : allow_private: false,
2345 2 : };
2346 2 : let started = std::time::Instant::now();
2347 2 : let err = deny
2348 2 : .check_host_within(
2349 2 : "ctx.example.invalid",
2350 2 : 443,
2351 2 : std::time::Duration::from_millis(0),
2352 2 : )
2353 2 : .await
2354 2 : .expect_err("an unanswered lookup must be denied, never allowed");
2355 2 : assert!(err.contains("timed out"), "denial names the timeout: {err}");
2356 2 : assert!(
2357 2 : started.elapsed() < std::time::Duration::from_secs(1),
2358 2 : "the caller must not wait on the resolver"
2359 2 : );
2360 2 : }
2361 :
2362 : /// The fetch cache is bounded in BYTES, not just in entries: one entry
2363 : /// may be MAX_CONTEXT_BYTES, so an entry-only bound is no memory bound.
2364 : #[tokio::test]
2365 2 : async fn fetch_cache_holds_a_byte_budget() {
2366 2 : let loader = Loader::with_policy(EgressPolicy {
2367 2 : allow_private: true,
2368 2 : });
2369 2 : let mib = 1024 * 1024;
2370 2 : let doc = Value::String("x".repeat(mib));
2371 80 : for i in 0..40 {
2372 80 : loader
2373 80 : .put_local(format!("https://ctx.example/{i}.jsonld"), doc.clone())
2374 80 : .await;
2375 : }
2376 2 : let entries = loader.cache_stats()["fetched"].as_u64().expect("count");
2377 2 : assert!(
2378 2 : entries <= MAX_FETCHED_CACHE_BYTES / mib as u64,
2379 2 : "byte budget breached: {entries} entries of 1 MiB"
2380 2 : );
2381 2 : }
2382 :
2383 : /// 6.3.16: "implementations shall periodically invalidate the "Cached"
2384 : /// @contexts according to the headers mentioned above." A repeat
2385 : /// resolution of the same @context value is served from the merged cache,
2386 : /// so the lifetime has to be enforced THERE too or a max-age=0 document is
2387 : /// frozen for the process lifetime.
2388 : #[tokio::test]
2389 2 : async fn merged_cache_honours_the_context_lifetime() {
2390 2 : let body = Arc::new(std::sync::Mutex::new(
2391 2 : r#"{"@context":{"speed":"https://a.example/speed"}}"#.to_string(),
2392 : ));
2393 2 : let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
2394 2 : .await
2395 2 : .expect("bind");
2396 2 : let addr = listener.local_addr().expect("addr");
2397 2 : let served = body.clone();
2398 2 : tokio::spawn(async move {
2399 6 : while let Ok((mut sock, _)) = listener.accept().await {
2400 : use tokio::io::{AsyncReadExt, AsyncWriteExt};
2401 4 : let mut buf = vec![0u8; 4096];
2402 4 : let _ = sock.read(&mut buf).await;
2403 4 : let b = served.lock().map(|b| b.clone()).unwrap_or_default();
2404 4 : let resp = format!(
2405 : "HTTP/1.1 200 OK\r\nContent-Type: application/ld+json\r\nCache-Control: max-age=0\r\nConnection: close\r\nContent-Length: {}\r\n\r\n{b}",
2406 4 : b.len()
2407 : );
2408 4 : let _ = sock.write_all(resp.as_bytes()).await;
2409 4 : let _ = sock.flush().await;
2410 : }
2411 0 : });
2412 2 : let loader = Loader::with_policy(EgressPolicy {
2413 2 : allow_private: true,
2414 2 : });
2415 2 : let url = Value::String(format!("http://{addr}/ctx.jsonld"));
2416 2 : let ctx = loader.resolve(&url).await.expect("first resolve");
2417 2 : assert_eq!(ctx.expand_key("speed"), "https://a.example/speed");
2418 :
2419 2 : *body.lock().expect("lock") =
2420 2 : r#"{"@context":{"speed":"https://b.example/speed"}}"#.to_string();
2421 2 : let ctx = loader.resolve(&url).await.expect("second resolve");
2422 2 : assert_eq!(
2423 2 : ctx.expand_key("speed"),
2424 2 : "https://b.example/speed",
2425 2 : "an expired @context must be re-resolved, not served from the merged cache"
2426 2 : );
2427 2 : }
2428 :
2429 : /// The merged cache is keyed by the SERIALIZED user @context, which an
2430 : /// `application/ld+json` body may carry inline up to the body cap — an
2431 : /// entry-only bound is no memory bound when one key is megabytes.
2432 : #[tokio::test]
2433 2 : async fn merged_cache_refuses_oversized_keys() {
2434 2 : let loader = Loader::with_policy(EgressPolicy {
2435 2 : allow_private: true,
2436 2 : });
2437 2 : let mut big = serde_json::Map::new();
2438 4000 : for i in 0..2000 {
2439 4000 : big.insert(
2440 4000 : format!("term{i:06}"),
2441 4000 : Value::String(format!("https://ex.example/{i:06}")),
2442 4000 : );
2443 4000 : }
2444 2 : let user = Value::Object(big);
2445 2 : assert!(user.to_string().len() > MAX_MERGED_KEY_BYTES);
2446 2 : for _ in 0..8 {
2447 16 : loader.resolve(&user).await.expect("resolve");
2448 : }
2449 2 : let stats = loader.cache_stats();
2450 2 : assert_eq!(
2451 2 : stats["merged"].as_u64().expect("count"),
2452 : 0,
2453 : "an oversized inline @context must not be cached: {stats}"
2454 : );
2455 : // negative: a small inline @context still is.
2456 2 : loader
2457 2 : .resolve(&serde_json::json!({"a": "https://ex.example/a"}))
2458 2 : .await
2459 2 : .expect("resolve");
2460 2 : assert_eq!(loader.cache_stats()["merged"].as_u64().expect("count"), 1);
2461 2 : }
2462 :
2463 : /// The doc comment on `ip_is_metadata` promises the metadata range is
2464 : /// "Refused whatever `allow_private` says" — that has to include the
2465 : /// native IPv6 spelling of the AWS IMDS endpoint, which is a ULA and would
2466 : /// otherwise be waved through by the default private-egress setting.
2467 : #[tokio::test]
2468 2 : async fn metadata_endpoint_is_denied_in_every_spelling() {
2469 2 : let allow = EgressPolicy {
2470 2 : allow_private: true,
2471 2 : };
2472 10 : for host in [
2473 2 : "169.254.169.254",
2474 2 : "::ffff:169.254.169.254",
2475 2 : "::169.254.169.254",
2476 2 : "fd00:ec2::254",
2477 2 : "[fd00:ec2::254]",
2478 2 : ] {
2479 10 : let err = allow
2480 10 : .check_host(host, 80)
2481 10 : .await
2482 10 : .expect_err("the metadata endpoint must be denied");
2483 10 : assert!(err.contains("metadata"), "{host}: {err}");
2484 2 : }
2485 2 : }
2486 :
2487 : /// The metadata deny is unconditional, so it has to cover every
2488 : /// provider's endpoint and not only the `169.254.169.254` the big four
2489 : /// share. `100.100.100.200` sits in carrier-grade NAT, which the default
2490 : /// `allow_private: true` posture does not deny, and an IPv6-only
2491 : /// deployment reaches the link-local endpoint through the NAT64 prefix.
2492 : #[tokio::test]
2493 2 : async fn metadata_deny_covers_the_cgnat_and_nat64_spellings() {
2494 2 : let allow = EgressPolicy {
2495 2 : allow_private: true,
2496 2 : };
2497 6 : for host in [
2498 2 : "100.100.100.200",
2499 2 : "[::ffff:100.100.100.200]",
2500 2 : // 64:ff9b::169.254.169.254
2501 2 : "[64:ff9b::a9fe:a9fe]",
2502 2 : ] {
2503 6 : let err = allow
2504 6 : .check_host(host, 80)
2505 6 : .await
2506 6 : .expect_err("the metadata endpoint must be denied");
2507 6 : assert!(err.contains("metadata"), "{host}: {err}");
2508 2 : }
2509 2 : }
2510 :
2511 : /// The merged-context cache is charged by the TERMS an entry holds, not
2512 : /// by counting entries: an @context document may spend its whole byte
2513 : /// budget on short mappings, and a merged Context costs memory in
2514 : /// proportion to its term map, its compaction inverse and its prefix
2515 : /// index. An entry ceiling alone would let 256 such documents pin
2516 : /// hundreds of megabytes.
2517 : #[cfg(not(target_arch = "wasm32"))]
2518 : #[test]
2519 2 : fn the_merged_cache_is_bounded_by_terms_not_by_entries() {
2520 20 : let big = |n: usize| {
2521 20 : let mut m = serde_json::Map::new();
2522 562466 : for i in 0..n {
2523 562466 : m.insert(
2524 562466 : format!("t{i:06}"),
2525 562466 : Value::String(format!("http://a.example/{i}")),
2526 562466 : );
2527 562466 : }
2528 20 : let mut c = Context::default();
2529 20 : c.merge_object(&m).expect("merge");
2530 20 : c.freeze();
2531 20 : Arc::new(c)
2532 20 : };
2533 2 : let cache = merged_cache();
2534 : // one entry far under the floor still costs the floor, so a flood of
2535 : // tiny contexts cannot turn the term budget into an unbounded map
2536 2 : cache.insert("small".into(), big(1));
2537 2 : cache.run_pending_tasks();
2538 2 : let floor = MAX_MERGED_CACHE_TERMS / MAX_MERGED_ENTRIES;
2539 2 : assert_eq!(
2540 2 : cache.weighted_size(),
2541 : floor,
2542 : "a small entry costs the floor"
2543 : );
2544 : // and a large one costs what it holds
2545 2 : let n = (floor as usize) * 4;
2546 2 : cache.insert("large".into(), big(n));
2547 2 : cache.run_pending_tasks();
2548 2 : assert_eq!(
2549 2 : cache.weighted_size(),
2550 2 : floor + n as u64,
2551 : "a large entry is charged its term count"
2552 : );
2553 : // past the budget the cache evicts rather than growing
2554 16 : for i in 0..8 {
2555 16 : cache.insert(format!("f{i}"), big(n));
2556 16 : }
2557 2 : cache.run_pending_tasks();
2558 2 : assert!(
2559 2 : cache.weighted_size() <= MAX_MERGED_CACHE_TERMS,
2560 : "weighted size {} passed the budget",
2561 0 : cache.weighted_size()
2562 : );
2563 2 : }
2564 :
2565 : /// 6to4 (IETF RFC 3056) is the fourth spelling that carries an IPv4
2566 : /// destination inside an IPv6 address: `2002:V4ADDR::/48`, where the two
2567 : /// segments after the prefix ARE the target IPv4 address. A host with a
2568 : /// 6to4 tunnel routes `[2002:a9fe:a9fe::]` to 169.254.169.254, so the
2569 : /// unconditional metadata deny has to unwrap it like the other three.
2570 : #[tokio::test]
2571 2 : async fn metadata_and_private_denies_cover_the_6to4_spelling() {
2572 2 : let allow = EgressPolicy {
2573 2 : allow_private: true,
2574 2 : };
2575 6 : for host in [
2576 2 : // 2002:169.254.169.254::
2577 2 : "[2002:a9fe:a9fe::]",
2578 2 : "[2002:a9fe:a9fe:1:2:3:4:5]",
2579 2 : // 2002:100.100.100.200::
2580 2 : "[2002:6464:64c8::]",
2581 2 : ] {
2582 6 : let err = allow
2583 6 : .check_host(host, 80)
2584 6 : .await
2585 6 : .expect_err("the metadata endpoint must be denied");
2586 6 : assert!(err.contains("metadata"), "{host}: {err}");
2587 : }
2588 2 : let deny = EgressPolicy {
2589 2 : allow_private: false,
2590 2 : };
2591 6 : for host in [
2592 2 : // 2002:127.0.0.1::
2593 2 : "[2002:7f00:1::]",
2594 2 : // 2002:10.1.1.1::
2595 2 : "[2002:a01:101::]",
2596 2 : // 2002:192.168.0.1::
2597 2 : "[2002:c0a8:1::]",
2598 2 : ] {
2599 6 : let err = deny
2600 6 : .check_host(host, 80)
2601 6 : .await
2602 6 : .expect_err("a 6to4 address standing for a private IPv4 must be denied");
2603 6 : assert!(err.contains("denied"), "{host}: {err}");
2604 2 : }
2605 2 : // the prefix alone denies nothing: 2002:5db8:d822:: stands for the
2606 2 : // public 93.184.216.34 and stays reachable
2607 2 : deny.check_host("[2002:5db8:d822::]", 80)
2608 2 : .await
2609 2 : .expect("a 6to4 address for a public IPv4 is public");
2610 2 : }
2611 :
2612 : /// ADR-0010 makes `allow_private: false` the internet-facing posture, so
2613 : /// the classifier has to cover the ranges an internal service actually
2614 : /// sits on: carrier-grade NAT (RFC 6598), `0.0.0.0/8`, which a Linux
2615 : /// stack routes to the local host, the IETF assignment and benchmarking
2616 : /// blocks, the reserved space above `240.0.0.0`, and the NAT64 prefix
2617 : /// (RFC 6052), which translates straight back to an IPv4 target.
2618 : #[tokio::test]
2619 2 : async fn private_deny_covers_the_ranges_internal_services_sit_on() {
2620 2 : let deny = EgressPolicy {
2621 2 : allow_private: false,
2622 2 : };
2623 16 : for host in [
2624 2 : "100.64.0.1",
2625 2 : "100.127.255.254",
2626 2 : "0.1.2.3",
2627 2 : "192.0.0.8",
2628 2 : "198.18.0.1",
2629 2 : "240.0.0.1",
2630 2 : "[::ffff:100.64.0.1]",
2631 2 : // 64:ff9b::10.1.1.1
2632 2 : "[64:ff9b::a01:101]",
2633 2 : ] {
2634 16 : let err = deny
2635 16 : .check_host(host, 80)
2636 16 : .await
2637 16 : .expect_err("an internal-range destination must be denied");
2638 16 : assert!(err.contains("denied"), "{host}: {err}");
2639 2 : }
2640 2 : // the widening stops at the public internet: 100.128.0.0 is the first
2641 2 : // address above carrier-grade NAT and stays reachable.
2642 6 : for host in ["93.184.216.34", "100.128.0.1", "[2606:2800:220::1]"] {
2643 6 : deny.check_host(host, 80)
2644 6 : .await
2645 6 : .unwrap_or_else(|e| panic!("{host} is public and must pass: {e}"));
2646 2 : }
2647 2 : }
2648 :
2649 : /// The denial text is returned to the client verbatim in the RFC 7807
2650 : /// `detail`, so naming the address a hostname resolved to turns the
2651 : /// request parameter into an internal-DNS oracle.
2652 : #[tokio::test]
2653 2 : async fn private_range_denial_does_not_name_the_resolved_address() {
2654 2 : let deny = EgressPolicy {
2655 2 : allow_private: false,
2656 2 : };
2657 : // A NAME that resolves privately takes the resolver path — the one
2658 : // that used to embed the answer. Where the name does not resolve the
2659 : // message is the lookup error, which leaks nothing.
2660 2 : let err = deny
2661 2 : .check_host("ip6-localhost", 80)
2662 2 : .await
2663 2 : .expect_err("a name resolving into a private range is denied");
2664 2 : assert!(
2665 2 : !err.contains("::1") && !err.contains("127.0.0.1"),
2666 : "leaked the resolved address: {err}"
2667 : );
2668 4 : for host in ["10.1.2.3", "127.0.0.1"] {
2669 2 : // an IP LITERAL is the client's own input — echoing it back leaks
2670 2 : // nothing it did not already know.
2671 4 : let err = deny.check_host(host, 80).await.expect_err("denied");
2672 4 : assert!(err.contains("private range"), "{host}: {err}");
2673 2 : }
2674 2 : }
2675 :
2676 : /// 5.5.6 assigns 504 LdContextNotAvailable to a remote @context that "is
2677 : /// not available" and BadRequestData to one that "is invalid". Nested
2678 : /// arrays and an over-long reference tree are entirely client-supplied and
2679 : /// touch no network, so they are the invalid case — a client must not be
2680 : /// able to mint gateway errors on demand.
2681 : #[tokio::test]
2682 2 : async fn client_side_context_caps_are_bad_request_not_gateway_errors() {
2683 2 : let loader = Loader::with_policy(EgressPolicy {
2684 2 : allow_private: true,
2685 2 : });
2686 2 : let mut nested = serde_json::json!(["x"]);
2687 24 : for _ in 0..12 {
2688 24 : nested = Value::Array(vec![nested]);
2689 24 : }
2690 2 : let err = loader
2691 2 : .resolve(&nested)
2692 2 : .await
2693 2 : .expect_err("a too-deep @context must be rejected");
2694 2 : assert!(
2695 2 : matches!(err, NgsiError::BadRequestData(_)),
2696 2 : "an over-nested @context is invalid input, got {err:?}"
2697 2 : );
2698 2 : }
2699 :
2700 : /// A security switch that only understands one spelling silently gives
2701 : /// the operator the opposite of the intent.
2702 : #[test]
2703 2 : fn egress_switch_ignores_case_and_whitespace() {
2704 12 : for v in ["false", "FALSE", "False", " false ", "0", " 0\t"] {
2705 12 : assert!(
2706 12 : !EgressPolicy::allow_private_from(Some(v)),
2707 : "{v:?} must turn the private-egress deny ON"
2708 : );
2709 : }
2710 8 : for v in ["true", "1", "", "yes"] {
2711 8 : assert!(
2712 8 : EgressPolicy::allow_private_from(Some(v)),
2713 : "{v:?} must leave private egress allowed"
2714 : );
2715 : }
2716 2 : assert!(
2717 2 : EgressPolicy::allow_private_from(None),
2718 : "unset means allowed"
2719 : );
2720 2 : }
2721 :
2722 : /// The published posture is part of the switch. SECURITY.md told a reader
2723 : /// private ranges were denied by default while this function allows them,
2724 : /// which reads as an SSRF guard that is on when it is off — and an
2725 : /// operator auditing the broker from that page would never set the
2726 : /// variable. The default lives in one place, here, so the two documents
2727 : /// that state it are checked against it here too.
2728 : #[test]
2729 2 : fn the_documented_egress_default_is_the_compiled_one() {
2730 2 : let root = concat!(env!("CARGO_MANIFEST_DIR"), "/../..");
2731 2 : let book = std::fs::read_to_string(format!("{root}/docs/src/configuration.md"))
2732 2 : .expect("the configuration chapter");
2733 2 : let row = book
2734 2 : .lines()
2735 96 : .find(|l| l.starts_with("| `ANTARES_EGRESS_ALLOW_PRIVATE` |"))
2736 2 : .expect("the variable has no row in the configuration table");
2737 2 : let stated = row
2738 2 : .split('|')
2739 2 : .nth(2)
2740 2 : .expect("the default column")
2741 2 : .trim()
2742 2 : .trim_matches('`');
2743 2 : assert_eq!(
2744 2 : stated.parse::<bool>().ok(),
2745 2 : Some(EgressPolicy::allow_private_from(None)),
2746 : "the chapter states {stated:?} as the default"
2747 : );
2748 :
2749 2 : let policy =
2750 2 : std::fs::read_to_string(format!("{root}/SECURITY.md")).expect("the security policy");
2751 2 : assert!(
2752 2 : !policy.contains("private-range deny by default"),
2753 : "SECURITY.md claims a deny-by-default this switch does not implement"
2754 : );
2755 2 : assert!(
2756 2 : policy.contains("ANTARES_EGRESS_ALLOW_PRIVATE=false"),
2757 : "SECURITY.md must name the switch that turns the deny on"
2758 : );
2759 2 : }
2760 :
2761 : /// A locally hosted @context is stored "for the Tenant" that added it
2762 : /// (5.13.1, 5.13.2.4) and 5.5.10 makes the Tenant the boundary an
2763 : /// operation applies within: another Tenant naming the same URL must not
2764 : /// have its payload expanded by those mappings. The URL is on a dead
2765 : /// port, so a resolution that succeeds can only have come from the local
2766 : /// entry — and for a foreign Tenant the @context is simply not available
2767 : /// (5.5.6).
2768 : #[tokio::test]
2769 2 : async fn clause_5_13_1_hosted_context_is_private_to_its_tenant() {
2770 2 : let loader = Loader::with_policy(EgressPolicy {
2771 2 : allow_private: true,
2772 2 : });
2773 2 : let alpha = TenantId::new("alpha").expect("tenant");
2774 2 : let beta = TenantId::new("beta").expect("tenant");
2775 2 : let url = Value::String(
2776 2 : "http://127.0.0.1:9/ngsi-ld/v1/jsonldContexts/2f2e1a00-0000-4000-8000-000000000001"
2777 2 : .to_owned(),
2778 2 : );
2779 2 : loader
2780 2 : .put_local_for(
2781 2 : &alpha,
2782 2 : url.as_str().expect("url").to_owned(),
2783 2 : serde_json::json!({"secret": "https://alpha.example/secret"}),
2784 2 : )
2785 2 : .await;
2786 :
2787 2 : let ctx = loader
2788 2 : .resolve_for(&alpha, &url)
2789 2 : .await
2790 2 : .expect("the owning Tenant resolves its own @context");
2791 2 : assert_eq!(ctx.expand_key("secret"), "https://alpha.example/secret");
2792 :
2793 2 : let err = loader
2794 2 : .resolve_for(&beta, &url)
2795 2 : .await
2796 2 : .expect_err("another Tenant must not resolve a @context it does not own");
2797 2 : assert!(
2798 2 : matches!(err, NgsiError::LdContextNotAvailable(_)),
2799 : "a foreign Hosted @context is not available, got {err:?}"
2800 : );
2801 : // and nothing of the owner's mappings reaches the other Tenant by way
2802 : // of the merged-context cache either
2803 2 : let ctx = loader
2804 2 : .resolve_for(
2805 2 : &beta,
2806 2 : &serde_json::json!({"other": "https://beta.example/other"}),
2807 2 : )
2808 2 : .await
2809 2 : .expect("resolve");
2810 2 : assert_eq!(
2811 2 : ctx.expand_key("secret"),
2812 2 : "https://uri.etsi.org/ngsi-ld/default-context/secret",
2813 2 : "the owner's term mapping must not expand another Tenant's payload"
2814 2 : );
2815 2 : }
2816 :
2817 : /// The merged cache is keyed by the user @context alone, so the Tenant
2818 : /// gate on a hit is the ONLY thing keeping one Tenant's locally stored
2819 : /// mappings out of another Tenant's resolution — and it reads the
2820 : /// ownership off the document cache, a bounded LRU that evicts under
2821 : /// load. A source document that is no longer there has unknown
2822 : /// ownership, which is not the same as public: the hit must be rebuilt,
2823 : /// not handed over.
2824 : #[tokio::test]
2825 2 : async fn an_evicted_source_document_does_not_open_a_merged_hit_to_another_tenant() {
2826 2 : let loader = Loader::with_policy(EgressPolicy {
2827 2 : allow_private: true,
2828 2 : });
2829 2 : let alpha = TenantId::new("alpha").expect("tenant");
2830 2 : let beta = TenantId::new("beta").expect("tenant");
2831 : // dead port: nothing can be re-fetched, so anything served can only
2832 : // have come from the caches under test
2833 2 : let url =
2834 2 : "http://127.0.0.1:9/ngsi-ld/v1/jsonldContexts/3f3e1a00-0000-4000-8000-000000000003";
2835 2 : let user = Value::String(url.to_owned());
2836 2 : loader
2837 2 : .put_local_for(
2838 2 : &alpha,
2839 2 : url.to_owned(),
2840 2 : serde_json::json!({"secret": "https://alpha.example/secret"}),
2841 2 : )
2842 2 : .await;
2843 2 : let ctx = loader
2844 2 : .resolve_for(&alpha, &user)
2845 2 : .await
2846 2 : .expect("the owning Tenant resolves its own @context");
2847 2 : assert_eq!(ctx.expand_key("secret"), "https://alpha.example/secret");
2848 :
2849 : // the document leaves the bounded cache; the merged entry it fed
2850 : // stays, as an LRU eviction leaves it
2851 2 : loader.fetched.invalidate(url);
2852 :
2853 2 : match loader.resolve_for(&beta, &user).await {
2854 2 : Err(_) => {}
2855 2 : Ok(ctx) => assert_ne!(
2856 2 : ctx.expand_key("secret"),
2857 2 : "https://alpha.example/secret",
2858 2 : "another Tenant's mappings were served from the merged cache"
2859 2 : ),
2860 2 : }
2861 2 : }
2862 :
2863 : /// Adding one @context must not throw away every Tenant's merged
2864 : /// contexts: the merged entries built FROM the written document are
2865 : /// dropped (a rewritten @context is never served stale) and the rest —
2866 : /// another Tenant's warm context included — stay.
2867 : #[tokio::test]
2868 2 : async fn hosted_context_write_keeps_unrelated_merged_contexts() {
2869 2 : let loader = Loader::with_policy(EgressPolicy {
2870 2 : allow_private: true,
2871 2 : });
2872 2 : let alpha = TenantId::new("alpha").expect("tenant");
2873 2 : let beta = TenantId::new("beta").expect("tenant");
2874 2 : let base = "http://127.0.0.1:9/ngsi-ld/v1/jsonldContexts";
2875 2 : let a_url = format!("{base}/aaaa1111-0000-4000-8000-000000000001");
2876 2 : let b_url = format!("{base}/bbbb2222-0000-4000-8000-000000000002");
2877 2 : loader
2878 2 : .put_local_for(
2879 2 : &alpha,
2880 2 : a_url.clone(),
2881 2 : serde_json::json!({"speed": "https://a.example/v1"}),
2882 2 : )
2883 2 : .await;
2884 2 : loader
2885 2 : .put_local_for(
2886 2 : &beta,
2887 2 : b_url.clone(),
2888 2 : serde_json::json!({"level": "https://b.example/level"}),
2889 2 : )
2890 2 : .await;
2891 2 : loader
2892 2 : .resolve_for(&alpha, &Value::String(a_url.clone()))
2893 2 : .await
2894 2 : .expect("alpha resolves its own @context");
2895 2 : loader
2896 2 : .resolve_for(&beta, &Value::String(b_url.clone()))
2897 2 : .await
2898 2 : .expect("beta resolves its own @context");
2899 2 : assert_eq!(loader.cache_stats()["merged"].as_u64(), Some(2));
2900 :
2901 : // alpha adds an UNRELATED @context: no merged context was built from
2902 : // it, so nothing may be discarded
2903 2 : loader
2904 2 : .put_local_for(
2905 2 : &alpha,
2906 2 : format!("{base}/cccc3333-0000-4000-8000-000000000003"),
2907 2 : serde_json::json!({"other": "https://a.example/other"}),
2908 2 : )
2909 2 : .await;
2910 2 : assert_eq!(
2911 2 : loader.cache_stats()["merged"].as_u64(),
2912 : Some(2),
2913 : "one Tenant's @context write must not flush another Tenant's merged context"
2914 : );
2915 :
2916 : // correctness first: rewriting a document a merged context WAS built
2917 : // from drops that entry, so the new mappings are the ones served
2918 2 : loader
2919 2 : .put_local_for(
2920 2 : &alpha,
2921 2 : a_url.clone(),
2922 2 : serde_json::json!({"speed": "https://a.example/v2"}),
2923 2 : )
2924 2 : .await;
2925 2 : let ctx = loader
2926 2 : .resolve_for(&alpha, &Value::String(a_url))
2927 2 : .await
2928 2 : .expect("resolve after rewrite");
2929 2 : assert_eq!(
2930 2 : ctx.expand_key("speed"),
2931 : "https://a.example/v2",
2932 : "a rewritten @context must never be served from the merged cache"
2933 : );
2934 2 : let ctx = loader
2935 2 : .resolve_for(&beta, &Value::String(b_url))
2936 2 : .await
2937 2 : .expect("beta's merged context survived");
2938 2 : assert_eq!(ctx.expand_key("level"), "https://b.example/level");
2939 2 : }
2940 :
2941 : #[tokio::test]
2942 2 : async fn pinned_versions_resolve_without_network() {
2943 2 : let l = Loader::new();
2944 14 : for v in ["1.3", "1.4", "1.5", "1.6", "1.7", "1.8", "1.9"] {
2945 14 : let url = format!("https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v{v}.jsonld");
2946 14 : let ctx = l.resolve(&Value::String(url)).await.expect("resolve");
2947 14 : assert_eq!(
2948 14 : ctx.expand_key("observedAt"),
2949 2 : "https://uri.etsi.org/ngsi-ld/observedAt"
2950 2 : );
2951 2 : }
2952 2 : }
2953 :
2954 : /// 4.4, V1.9.1: "The NGSI-LD Core @context is publicly available at
2955 : /// `https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.9.jsonld` and
2956 : /// shall contain all the terms as mandated by annex B." A request that
2957 : /// names no @context of its own is answered under that document alone,
2958 : /// so the terms V1.9 added have to expand from the implicit core — not
2959 : /// only when a client spells the v1.9 URL out. Under an older core they
2960 : /// fall through @vocab into the default context, and a member the broker
2961 : /// itself renders (`Snapshot`, `valueType`, `orderBy`) then carries a
2962 : /// name no other implementation reads as the core term.
2963 : #[tokio::test]
2964 2 : async fn the_implicit_core_context_is_the_one_v1_9_1_names() {
2965 2 : assert_eq!(
2966 : CORE_CONTEXT,
2967 : "https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.9.jsonld"
2968 : );
2969 2 : let core = Loader::new().core();
2970 12 : for (term, iri) in [
2971 2 : ("Snapshot", "https://uri.etsi.org/ngsi-ld/Snapshot"),
2972 2 : ("orderBy", "https://uri.etsi.org/ngsi-ld/orderBy"),
2973 2 : ("collation", "https://uri.etsi.org/ngsi-ld/collation"),
2974 2 : ("valueType", "https://uri.etsi.org/ngsi-ld/hasValueType"),
2975 2 : ("objectLists", "https://uri.etsi.org/ngsi-ld/hasObjectLists"),
2976 2 : ("aggrMethods", "https://uri.etsi.org/ngsi-ld/aggrMethods"),
2977 2 : ] {
2978 12 : assert_eq!(
2979 12 : core.expand_key(term),
2980 2 : iri,
2981 2 : "{term} must expand to its Annex B IRI, not through @vocab"
2982 2 : );
2983 2 : }
2984 2 : }
2985 :
2986 : /// Annex B (normative), V1.9.1: the core @context served for the v1.9
2987 : /// URL is that version's document. The terms V1.9 added are the test —
2988 : /// a v1.8 document under the v1.9 name expands every one of them
2989 : /// through the @vocab fallback into the default context instead, and
2990 : /// the broker then does not see a core member at all.
2991 : #[tokio::test]
2992 2 : async fn the_v1_9_core_context_carries_the_terms_v1_9_added() {
2993 2 : let l = Loader::new();
2994 2 : let ctx = l
2995 2 : .resolve(&Value::String(
2996 2 : "https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.9.jsonld".into(),
2997 2 : ))
2998 2 : .await
2999 2 : .expect("resolve");
3000 34 : for term in [
3001 2 : "Snapshot",
3002 2 : "SnapshotNotification",
3003 2 : "ExecutionResultDetails",
3004 2 : "aggrMethods",
3005 2 : "aggrParams",
3006 2 : "aggrPeriodDuration",
3007 2 : "collation",
3008 2 : "lastUsedAt",
3009 2 : "ngsildproof",
3010 2 : "orderBy",
3011 2 : "ordering",
3012 2 : "problemDetails",
3013 2 : "resultStatus",
3014 2 : "snapshotId",
3015 2 : "snapshotLifetime",
3016 2 : "snapshotPriority",
3017 2 : "snapshotStatus",
3018 2 : ] {
3019 34 : assert_eq!(
3020 34 : ctx.expand_key(term),
3021 34 : format!("https://uri.etsi.org/ngsi-ld/{term}"),
3022 : "{term} must expand to its Annex B IRI, not through @vocab"
3023 : );
3024 : }
3025 : // renamed by the V1.9 annex, and the rename is only visible if the
3026 : // document under the v1.9 name really is the v1.9 one.
3027 2 : assert_eq!(
3028 2 : ctx.expand_key("objectLists"),
3029 : "https://uri.etsi.org/ngsi-ld/hasObjectLists"
3030 : );
3031 4 : let terms = |url: &str| {
3032 4 : pinned(url)
3033 4 : .and_then(|c| c.as_object().cloned())
3034 4 : .expect("pinned document")
3035 4 : };
3036 2 : let v19 = terms("https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.9.jsonld");
3037 2 : let v18 = terms("https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.8.jsonld");
3038 4 : for gone in ["objectsLists", "geometryProperty"] {
3039 4 : assert!(v18.contains_key(gone), "{gone} belongs to the v1.8 annex");
3040 4 : assert!(
3041 4 : !v19.contains_key(gone),
3042 2 : "{gone} is not in the V1.9 annex; the v1.9 document is a copy of v1.8"
3043 2 : );
3044 2 : }
3045 2 : assert!(!v18.contains_key("Snapshot"), "v1.8 predates Snapshot");
3046 2 : }
3047 : }
|
