Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Outbound safety for the request-path egress classes:
3 : //! notification delivery and federation forwarding. The third class,
4 : //! @context fetching, enforces the same policy inside `antares-jsonld`
5 : //! (that is where the fetch happens) — this module governs the two that
6 : //! leave from `antares-api`.
7 : //!
8 : //! Per-destination circuit breakers matter at federation scale: a dead
9 : //! peer must not spend its full timeout on every request.
10 :
11 : use std::collections::HashMap;
12 : use std::sync::Mutex;
13 : use std::time::Duration;
14 : // Clock rule: 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 :
21 : /// Ceiling on tracked destinations and registrations, held with the other
22 : /// published ceilings so `/q/health` reports it.
23 : use crate::bounds::MAX_TRACKED_DESTINATIONS as MAX_TRACKED;
24 : /// Consecutive failures before a destination is tripped.
25 : pub(crate) const TRIP_AFTER: u32 = 5;
26 : /// How long a tripped destination stays open-circuit before one probe.
27 : const COOLDOWN: Duration = Duration::from_secs(30);
28 :
29 : #[derive(Default)]
30 : struct Breaker {
31 : failures: u32,
32 : tripped_at: Option<Instant>,
33 : /// When this entry was last written — the eviction order at the ceiling.
34 : touched_at: Option<Instant>,
35 : }
36 :
37 : /// Drop the least recently written entry once the map is at its ceiling, so a
38 : /// new key always has room. Called before inserting, never on lookup.
39 : ///
40 : /// The ceiling is shared by every tenant, so the eviction stays inside the
41 : /// tenant that is filling it (both maps are keyed `tenant\u{1f}rest`): one
42 : /// tenant pointing subscriptions at thousands of dead hosts would otherwise
43 : /// drop another tenant's tripped breaker, and that tenant's notifications go
44 : /// back to spending a full timeout on a destination already known dead. A key
45 : /// whose tenant holds no entry yet takes the globally oldest one, so a tenant
46 : /// arriving at a full map still gets in.
47 : // ponytail: a linear scan per eviction, which happens only at the ceiling on
48 : // a new key. Measured on this shape at MAX_TRACKED: 10-15 us when the
49 : // arriving tenant already holds entries, 25 us in the worst case, where it
50 : // holds none and both scans run over the whole map. That is paid only by a
51 : // destination the map has never seen while it is full, so a per-tenant LRU
52 : // list buys back microseconds in a shape no subscription set produces; add
53 : // one if a profile ever disagrees.
54 55196 : fn evict_oldest<V>(map: &mut HashMap<String, V>, key: &str, stamp: impl Fn(&V) -> Option<Instant>) {
55 55196 : let prefix = format!("{}\u{1f}", key.split('\u{1f}').next().unwrap_or(""));
56 61200 : while map.len() >= MAX_TRACKED {
57 8004 : let oldest = |mine: bool| {
58 8004 : map.iter()
59 32784384 : .filter(|(k, _)| k.starts_with(&prefix) == mine)
60 24590380 : .min_by_key(|(_, v)| stamp(v))
61 8004 : .map(|(k, _)| k.clone())
62 8004 : };
63 6004 : let Some(victim) = oldest(true).or_else(|| oldest(false)) else {
64 0 : return;
65 : };
66 6004 : map.remove(&victim);
67 : }
68 55196 : }
69 :
70 : /// Egress gate shared by the notification and federation paths.
71 : pub struct Egress {
72 : policy: antares_jsonld::EgressPolicy,
73 : breakers: Mutex<HashMap<String, Breaker>>,
74 : /// 5.2.34 cooldown: instant of the last failed forward per registration
75 : /// id — only consulted for registrations that DECLARE management.cooldown.
76 : reg_failures: Mutex<HashMap<String, Instant>>,
77 : }
78 :
79 : impl Default for Egress {
80 20 : fn default() -> Self {
81 20 : Self::new(antares_jsonld::EgressPolicy::from_env())
82 20 : }
83 : }
84 :
85 : /// A URI as it may be repeated back to a caller. Every reason string this
86 : /// module returns is interpolated into a log line beside a URI the caller
87 : /// already redacted, so it carries the same redaction: 5.2.9 puts no limit on
88 : /// a registered endpoint's URI and reqwest sends its userinfo as credentials.
89 16 : fn redacted(url: &str) -> String {
90 16 : antares_notifier::redact_userinfo(url)
91 16 : }
92 :
93 : /// The 5.2.34 cooldown key. The registration id is client-chosen PER TENANT
94 : /// (5.5.10), so the bare id would let one tenant's failing registration put
95 : /// another tenant's same-id registration into timeout. The unit separator
96 : /// cannot appear in either part (TenantId and EntityId both refuse C0
97 : /// controls).
98 34 : pub(crate) fn reg_key(tenant: &str, reg_id: &str) -> String {
99 34 : format!("{tenant}\u{1f}{reg_id}")
100 34 : }
101 :
102 : impl Egress {
103 4860 : pub fn new(policy: antares_jsonld::EgressPolicy) -> Self {
104 4860 : Self {
105 4860 : policy,
106 4860 : breakers: Mutex::new(HashMap::new()),
107 4860 : reg_failures: Mutex::new(HashMap::new()),
108 4860 : }
109 4860 : }
110 :
111 : /// 5.2.34 cooldown: "If requests are received before the cooldown
112 : /// period has expired, a timeout error response for the registration is
113 : /// automatically returned." True while the per-registration window is
114 : /// still open.
115 20 : pub(crate) fn reg_in_cooldown(&self, reg_key: &str, cooldown_ms: u64) -> bool {
116 20 : self.reg_failures
117 20 : .lock()
118 20 : .unwrap_or_else(std::sync::PoisonError::into_inner)
119 20 : .get(reg_key)
120 20 : .is_some_and(|t| t.elapsed() < Duration::from_millis(cooldown_ms))
121 20 : }
122 :
123 : /// 5.2.34 cooldown bookkeeping: a failed forward stamps the window, a
124 : /// successful one clears it.
125 18398 : pub fn reg_record(&self, reg_key: &str, ok: bool) {
126 18398 : let mut m = self
127 18398 : .reg_failures
128 18398 : .lock()
129 18398 : .unwrap_or_else(std::sync::PoisonError::into_inner);
130 18398 : if ok {
131 4 : m.remove(reg_key);
132 4 : } else {
133 18394 : if !m.contains_key(reg_key) {
134 8192000 : evict_oldest(&mut m, reg_key, |t: &Instant| Some(*t));
135 0 : }
136 18394 : m.insert(reg_key.to_owned(), Instant::now());
137 : }
138 18398 : }
139 :
140 : /// scheme allowlist + private-range deny. `Err` is a reason
141 : /// string for the caller\'s log/207 detail.
142 1456 : pub async fn check_url(&self, url: &str) -> Result<(), String> {
143 1456 : let scheme = reqwest::Url::parse(url)
144 1456 : .map(|u| u.scheme().to_owned())
145 1456 : .map_err(|e| format!("bad URL {}: {e}", redacted(url)))?;
146 1448 : match scheme.as_str() {
147 1448 : "http" | "https" => {}
148 8 : other => return Err(format!("scheme {other:?} is not allowed for egress")),
149 : }
150 1440 : self.check_destination(url).await
151 1456 : }
152 :
153 : /// The host policy for the destination of any notification binding. The
154 : /// scheme belongs to the sink (6.3.8, clause 7, or one a deployment
155 : /// registered); the host and port belong here. A URI with no host names
156 : /// no destination that can be cleared, so it is refused.
157 : ///
158 : /// This is the verdict on the destination as WRITTEN. A destination
159 : /// written as a name, under the default `ANTARES_EGRESS_ALLOW_PRIVATE`,
160 : /// is not resolved here — the addresses a name stands for are judged by
161 : /// the transport that dials them, and a binding that opens its own
162 : /// socket owes that filter (`EgressPolicy::ip_is_metadata` and
163 : /// `ip_is_private` over the resolved answer, as `checked_addr` does for
164 : /// MQTT and `PolicyResolver` for every reqwest client).
165 6440 : pub(crate) async fn check_destination(&self, url: &str) -> Result<(), String> {
166 6440 : let parsed =
167 6440 : reqwest::Url::parse(url).map_err(|e| format!("bad URL {}: {e}", redacted(url)))?;
168 6440 : let host = parsed
169 6440 : .host_str()
170 6440 : .filter(|h| !h.is_empty())
171 6440 : .ok_or_else(|| format!("endpoint {} names no host", redacted(url)))?
172 6432 : .to_owned();
173 6432 : let port = parsed.port_or_known_default().unwrap_or(443);
174 6432 : self.policy.check_host(&host, port).await
175 6440 : }
176 :
177 : /// The breaker key: one destination, within one tenant. 4.14 puts the
178 : /// tenant in it — "the NGSI-LD API operations for managing, retrieving
179 : /// and subscribing to entity information, but also any context source
180 : /// related operations only apply to the information of the specified
181 : /// `Tenant` in isolation and never have any effect on the information of
182 : /// other `Tenants`". Tenants share destinations (one consumer host, one
183 : /// MQTT broker), so a destination-only key lets one tenant's failing
184 : /// endpoint suppress another tenant's notifications to the same
185 : /// host:port, and the victim sees no evidence: a suppressed delivery
186 : /// deliberately does not move `timesSent`, `lastNotification` or
187 : /// `status`. Same reasoning, and the same separator, as `reg_key`.
188 : ///
189 : /// Userinfo, path and topic stay out: they are the credentials and the
190 : /// destination WITHIN a peer, and it is the peer that goes unresponsive.
191 49509 : fn key(tenant: &str, url: &str) -> String {
192 49509 : let dest = reqwest::Url::parse(url)
193 49509 : .ok()
194 49509 : .map(|u| {
195 49509 : format!(
196 : "{}://{}:{}",
197 49509 : u.scheme(),
198 49509 : u.host_str().unwrap_or_default(),
199 49509 : u.port_or_known_default().unwrap_or(0)
200 : )
201 49509 : })
202 49509 : .unwrap_or_else(|| url.to_owned());
203 49509 : format!("{tenant}\u{1f}{dest}")
204 49509 : }
205 :
206 : /// Is this destination currently open-circuit FOR THIS TENANT? A tripped
207 : /// destination admits ONE probe per cooldown window (half-open).
208 6422 : pub(crate) fn is_open(&self, tenant: &str, url: &str) -> bool {
209 6422 : let mut map = self
210 6422 : .breakers
211 6422 : .lock()
212 6422 : .unwrap_or_else(std::sync::PoisonError::into_inner);
213 6422 : let Some(b) = map.get_mut(&Self::key(tenant, url)) else {
214 6354 : return false;
215 : };
216 36 : match b.tripped_at {
217 36 : Some(t) if t.elapsed() >= COOLDOWN => {
218 0 : b.tripped_at = Some(Instant::now()); // this call IS the probe
219 0 : false
220 : }
221 36 : Some(_) => true,
222 32 : None => false,
223 : }
224 6422 : }
225 :
226 6157 : pub(crate) fn record_success(&self, tenant: &str, url: &str) {
227 6157 : self.breakers
228 6157 : .lock()
229 6157 : .unwrap_or_else(std::sync::PoisonError::into_inner)
230 6157 : .remove(&Self::key(tenant, url));
231 6157 : }
232 :
233 36930 : pub(crate) fn record_failure(&self, tenant: &str, url: &str) {
234 36930 : let mut map = self
235 36930 : .breakers
236 36930 : .lock()
237 36930 : .unwrap_or_else(std::sync::PoisonError::into_inner);
238 36930 : let k = Self::key(tenant, url);
239 36930 : if !map.contains_key(&k) {
240 36802 : evict_oldest(&mut map, &k, |b: &Breaker| b.touched_at);
241 128 : }
242 36930 : let b = map.entry(k).or_default();
243 36930 : b.failures = b.failures.saturating_add(1);
244 36930 : b.touched_at = Some(Instant::now());
245 36930 : if b.failures >= TRIP_AFTER {
246 32 : b.tripped_at = Some(Instant::now());
247 36898 : }
248 36930 : }
249 : }
250 :
251 : #[cfg(test)]
252 : mod tests {
253 : use super::*;
254 :
255 : /// The reason string is interpolated into a caller's log line next to a
256 : /// URI that caller redacted (`notify.rs`, `federation.rs`), so it may not
257 : /// smuggle back what the redaction removed: 5.2.9 allows any URI as a
258 : /// registered endpoint and reqwest sends its userinfo as basic auth.
259 : #[tokio::test]
260 4 : async fn a_refused_url_never_repeats_its_userinfo() {
261 4 : let e = Egress::new(antares_jsonld::EgressPolicy {
262 4 : allow_private: false,
263 4 : private_hosts: &[],
264 4 : });
265 8 : for url in [
266 4 : "http://alice:s3cret@[not-an-ip]/x",
267 4 : "http://alice:s3cret@/x",
268 4 : ] {
269 8 : let err = e.check_url(url).await.expect_err("refused");
270 8 : assert!(!err.contains("s3cret"), "userinfo in the reason: {err}");
271 4 : }
272 4 : }
273 :
274 : #[tokio::test]
275 4 : async fn scheme_allowlist_and_private_deny() {
276 4 : let e = Egress::new(antares_jsonld::EgressPolicy {
277 4 : allow_private: false,
278 4 : private_hosts: &[],
279 4 : });
280 4 : assert!(e.check_url("file:///etc/passwd").await.is_err());
281 4 : assert!(e.check_url("http://127.0.0.1:9090/x").await.is_err());
282 4 : assert!(e
283 4 : .check_url("http://169.254.169.254/latest/meta-data")
284 4 : .await
285 4 : .is_err());
286 4 : let allow = Egress::new(antares_jsonld::EgressPolicy {
287 4 : allow_private: true,
288 4 : private_hosts: &[],
289 4 : });
290 4 : assert!(allow.check_url("http://127.0.0.1:9090/x").await.is_ok());
291 4 : assert!(
292 4 : allow.check_url("mqtt://localhost:1883/t").await.is_err(),
293 : "@context fetches and federation forwards are HTTP; a notification \
294 : binding's own scheme goes through check_destination"
295 : );
296 : // A binding's own scheme is the sink's business, but its host is
297 : // still the policy's: a plugin binding cannot reach a denied host,
298 : // and an endpoint with no host is refused outright.
299 4 : assert!(allow
300 4 : .check_destination("wss://localhost:9000/n")
301 4 : .await
302 4 : .is_ok());
303 4 : assert!(e.check_destination("wss://127.0.0.1:9000/n").await.is_err());
304 4 : assert!(e
305 4 : .check_destination("wss://169.254.169.254/latest")
306 4 : .await
307 4 : .is_err());
308 4 : assert!(allow.check_destination("file:///etc/passwd").await.is_err());
309 4 : assert!(allow.check_destination("memory://").await.is_err());
310 4 : }
311 :
312 : /// The metadata denial does not depend on `ANTARES_EGRESS_ALLOW_PRIVATE`
313 : /// — but this check judges the destination as WRITTEN. A literal
314 : /// metadata address is refused in every spelling with private egress
315 : /// allowed; a host written as a NAME is not resolved here under that
316 : /// switch, and the classifier below is what the transports apply to the
317 : /// addresses the name turns out to stand for.
318 : #[tokio::test]
319 4 : async fn a_literal_metadata_address_is_refused_with_private_egress_allowed() {
320 4 : let allow = Egress::new(antares_jsonld::EgressPolicy {
321 4 : allow_private: true,
322 4 : private_hosts: &[],
323 4 : });
324 24 : for u in [
325 4 : "http://169.254.169.254/latest/meta-data",
326 4 : "http://100.100.100.200/latest",
327 4 : "http://[fd00:ec2::254]/latest",
328 4 : "http://[::ffff:169.254.169.254]/latest",
329 4 : "http://[64:ff9b::a9fe:a9fe]/latest",
330 4 : // 6to4 (RFC 3056): 2002:169.254.169.254::
331 4 : "http://[2002:a9fe:a9fe::]/latest",
332 4 : ] {
333 24 : assert!(
334 24 : allow.check_url(u).await.is_err(),
335 4 : "{u} reached the instance-metadata range"
336 4 : );
337 24 : assert!(
338 24 : allow.check_destination(u).await.is_err(),
339 4 : "{u} reached the instance-metadata range as a binding endpoint"
340 4 : );
341 4 : }
342 4 : // The same addresses, as an answer a NAME resolves to: this is the
343 4 : // classifier `PolicyResolver` and the MQTT connect run before they
344 4 : // dial, and it is what covers the case the check above cannot see.
345 28 : for ip in [
346 4 : "169.254.169.254",
347 4 : "100.100.100.200",
348 4 : "fd00:ec2::254",
349 4 : "::ffff:169.254.169.254",
350 4 : "64:ff9b::a9fe:a9fe",
351 4 : "2002:a9fe:a9fe::",
352 4 : "2002:a9fe:a9fe:1:2:3:4:5",
353 4 : ] {
354 28 : assert!(
355 28 : antares_jsonld::EgressPolicy::ip_is_metadata(ip.parse().expect("address")),
356 4 : "{ip} not classified as instance metadata"
357 4 : );
358 4 : }
359 4 : }
360 :
361 : /// 5.2.34 + 5.5.10: the cooldown a failing registration earns belongs to
362 : /// ITS tenant. Another tenant's registration under the same client-chosen
363 : /// id keeps being contacted.
364 : #[test]
365 4 : fn cooldown_is_scoped_to_the_tenant_that_earned_it() {
366 4 : let e = Egress::default();
367 4 : let id = "urn:ngsi-ld:ContextSourceRegistration:shared";
368 4 : e.reg_record(®_key("tenant-a", id), false);
369 4 : assert!(
370 4 : e.reg_in_cooldown(®_key("tenant-a", id), 60_000),
371 : "the failing tenant's registration is in its window"
372 : );
373 4 : assert!(
374 4 : !e.reg_in_cooldown(®_key("tenant-b", id), 60_000),
375 : "one tenant's failing registration must not put another tenant's \
376 : same-id registration into timeout"
377 : );
378 : // and a success clears only its own tenant's stamp
379 4 : e.reg_record(®_key("tenant-b", id), false);
380 4 : e.reg_record(®_key("tenant-a", id), true);
381 4 : assert!(!e.reg_in_cooldown(®_key("tenant-a", id), 60_000));
382 4 : assert!(e.reg_in_cooldown(®_key("tenant-b", id), 60_000));
383 4 : }
384 :
385 : #[test]
386 4 : fn breaker_trips_after_consecutive_failures() {
387 4 : let e = Egress::default();
388 4 : let t = "tenant-a";
389 4 : let url = "http://dead.example:9090/notify";
390 4 : for _ in 0..(TRIP_AFTER - 1) {
391 16 : e.record_failure(t, url);
392 16 : assert!(!e.is_open(t, url), "not tripped before the threshold");
393 : }
394 4 : e.record_failure(t, url);
395 4 : assert!(e.is_open(t, url), "tripped at the threshold");
396 : // per-destination, not global
397 4 : assert!(!e.is_open(t, "http://healthy.example:9090/notify"));
398 4 : e.record_success(t, url);
399 4 : assert!(!e.is_open(t, url), "success clears the breaker");
400 4 : }
401 :
402 : /// 4.14 + 5.5.10: the breaker a failing endpoint earns belongs to the
403 : /// tenant whose delivery earned it. Tenants share destinations, and
404 : /// whether a destination answers inside the deadline is a property of the
405 : /// pair, not of the host: the same host is a timeout for a subscription
406 : /// at the 6.3.8 100 ms floor and healthy for one that allows 5 s.
407 : #[test]
408 4 : fn a_tripped_destination_is_tripped_only_for_the_tenant_that_tripped_it() {
409 4 : let e = Egress::default();
410 4 : let url = "http://shared-consumer.example:8080/notify";
411 20 : for _ in 0..TRIP_AFTER {
412 20 : e.record_failure("tenant-a", url);
413 20 : }
414 4 : assert!(e.is_open("tenant-a", url), "the failing tenant is tripped");
415 4 : assert!(
416 4 : !e.is_open("tenant-b", url),
417 : "one tenant's failing endpoint must not suppress another \
418 : tenant's notifications to the same host:port"
419 : );
420 : // and clearing one tenant's breaker leaves the other's alone
421 20 : for _ in 0..TRIP_AFTER {
422 20 : e.record_failure("tenant-b", url);
423 20 : }
424 4 : e.record_success("tenant-a", url);
425 4 : assert!(!e.is_open("tenant-a", url));
426 4 : assert!(e.is_open("tenant-b", url));
427 4 : }
428 :
429 : /// The ceiling is shared; the isolation must not be. A tenant churning
430 : /// destinations past it evicts its OWN oldest entry — another tenant's
431 : /// tripped breaker survives, or every later notification to that
432 : /// tenant's dead endpoint goes back to spending a full timeout.
433 : #[test]
434 4 : fn filling_the_ceiling_leaves_another_tenants_breaker_tripped() {
435 4 : let e = Egress::default();
436 4 : let victim = "http://dead-peer.example:9090/notify";
437 20 : for _ in 0..TRIP_AFTER {
438 20 : e.record_failure("b", victim);
439 20 : }
440 4 : assert!(e.is_open("b", victim), "the breaker starts tripped");
441 18384 : for i in 0..(MAX_TRACKED + 500) {
442 18384 : e.record_failure("a", &format!("http://churn-{i}.example:9090/notify"));
443 18384 : }
444 4 : assert!(
445 4 : e.is_open("b", victim),
446 : "one tenant's churn cleared another tenant's breaker"
447 : );
448 4 : }
449 :
450 : /// Both maps are keyed by client-supplied strings (notification endpoints,
451 : /// registration ids), so neither may grow without a ceiling: a client that
452 : /// points subscriptions at thousands of dead hosts must not be able to
453 : /// spend the broker's memory one entry at a time.
454 : #[test]
455 4 : fn destination_maps_stay_bounded_under_distinct_keys() {
456 4 : let e = Egress::default();
457 18384 : for i in 0..(MAX_TRACKED + 500) {
458 18384 : e.record_failure("t", &format!("http://dead-{i}.example:9090/notify"));
459 18384 : e.reg_record(&format!("urn:ngsi-ld:CSR:{i}"), false);
460 18384 : }
461 4 : assert!(
462 4 : e.breakers.lock().expect("breaker lock").len() <= MAX_TRACKED,
463 : "breaker map grew past the ceiling"
464 : );
465 4 : assert!(
466 4 : e.reg_failures.lock().expect("reg_failures lock").len() <= MAX_TRACKED,
467 : "registration cooldown map grew past the ceiling"
468 : );
469 : // The ceiling must not cost correctness for a live destination: the
470 : // most recently recorded failure is still tracked.
471 4 : let live = format!("http://dead-{}.example:9090/notify", MAX_TRACKED + 499);
472 20 : for _ in 0..TRIP_AFTER {
473 20 : e.record_failure("t", &live);
474 20 : }
475 4 : assert!(e.is_open("t", &live), "recent destination still trips");
476 4 : }
477 : }
|
