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