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 55190 : fn evict_oldest<V>(map: &mut HashMap<String, V>, key: &str, stamp: impl Fn(&V) -> Option<Instant>) {
55 55190 : let prefix = format!("{}\u{1f}", key.split('\u{1f}').next().unwrap_or(""));
56 61194 : 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 55190 : }
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 28 : pub(crate) fn reg_key(tenant: &str, reg_id: &str) -> String {
99 28 : format!("{tenant}\u{1f}{reg_id}")
100 28 : }
101 :
102 : impl Egress {
103 3224 : pub fn new(policy: antares_jsonld::EgressPolicy) -> Self {
104 3224 : Self {
105 3224 : policy,
106 3224 : breakers: Mutex::new(HashMap::new()),
107 3224 : reg_failures: Mutex::new(HashMap::new()),
108 3224 : }
109 3224 : }
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 16 : pub(crate) fn reg_in_cooldown(&self, reg_key: &str, cooldown_ms: u64) -> bool {
116 16 : self.reg_failures
117 16 : .lock()
118 16 : .unwrap_or_else(std::sync::PoisonError::into_inner)
119 16 : .get(reg_key)
120 16 : .is_some_and(|t| t.elapsed() < Duration::from_millis(cooldown_ms))
121 16 : }
122 :
123 : /// 5.2.34 cooldown bookkeeping: a failed forward stamps the window, a
124 : /// successful one clears it.
125 18396 : pub fn reg_record(&self, reg_key: &str, ok: bool) {
126 18396 : let mut m = self
127 18396 : .reg_failures
128 18396 : .lock()
129 18396 : .unwrap_or_else(std::sync::PoisonError::into_inner);
130 18396 : if ok {
131 4 : m.remove(reg_key);
132 4 : } else {
133 18392 : if !m.contains_key(reg_key) {
134 8192000 : evict_oldest(&mut m, reg_key, |t: &Instant| Some(*t));
135 0 : }
136 18392 : m.insert(reg_key.to_owned(), Instant::now());
137 : }
138 18396 : }
139 :
140 : /// scheme allowlist + private-range deny. `Err` is a reason
141 : /// string for the caller\'s log/207 detail.
142 477 : pub async fn check_url(&self, url: &str) -> Result<(), String> {
143 477 : let scheme = reqwest::Url::parse(url)
144 477 : .map(|u| u.scheme().to_owned())
145 477 : .map_err(|e| format!("bad URL {}: {e}", redacted(url)))?;
146 469 : match scheme.as_str() {
147 469 : "http" | "https" => {}
148 8 : other => return Err(format!("scheme {other:?} is not allowed for egress")),
149 : }
150 461 : self.check_destination(url).await
151 477 : }
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 887 : pub(crate) async fn check_destination(&self, url: &str) -> Result<(), String> {
166 887 : let parsed =
167 887 : reqwest::Url::parse(url).map_err(|e| format!("bad URL {}: {e}", redacted(url)))?;
168 887 : let host = parsed
169 887 : .host_str()
170 887 : .filter(|h| !h.is_empty())
171 887 : .ok_or_else(|| format!("endpoint {} names no host", redacted(url)))?
172 879 : .to_owned();
173 879 : let port = parsed.port_or_known_default().unwrap_or(443);
174 879 : self.policy.check_host(&host, port).await
175 887 : }
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 38500 : fn key(tenant: &str, url: &str) -> String {
192 38500 : let dest = reqwest::Url::parse(url)
193 38500 : .ok()
194 38500 : .map(|u| {
195 38500 : format!(
196 : "{}://{}:{}",
197 38500 : u.scheme(),
198 38500 : u.host_str().unwrap_or_default(),
199 38500 : u.port_or_known_default().unwrap_or(0)
200 : )
201 38500 : })
202 38500 : .unwrap_or_else(|| url.to_owned());
203 38500 : format!("{tenant}\u{1f}{dest}")
204 38500 : }
205 :
206 : /// Is this destination currently open-circuit FOR THIS TENANT? A tripped
207 : /// destination admits ONE probe per cooldown window (half-open).
208 873 : pub(crate) fn is_open(&self, tenant: &str, url: &str) -> bool {
209 873 : let mut map = self
210 873 : .breakers
211 873 : .lock()
212 873 : .unwrap_or_else(std::sync::PoisonError::into_inner);
213 873 : let Some(b) = map.get_mut(&Self::key(tenant, url)) else {
214 805 : 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 873 : }
225 :
226 701 : pub(crate) fn record_success(&self, tenant: &str, url: &str) {
227 701 : self.breakers
228 701 : .lock()
229 701 : .unwrap_or_else(std::sync::PoisonError::into_inner)
230 701 : .remove(&Self::key(tenant, url));
231 701 : }
232 :
233 36926 : pub(crate) fn record_failure(&self, tenant: &str, url: &str) {
234 36926 : let mut map = self
235 36926 : .breakers
236 36926 : .lock()
237 36926 : .unwrap_or_else(std::sync::PoisonError::into_inner);
238 36926 : let k = Self::key(tenant, url);
239 36926 : if !map.contains_key(&k) {
240 36798 : evict_oldest(&mut map, &k, |b: &Breaker| b.touched_at);
241 128 : }
242 36926 : let b = map.entry(k).or_default();
243 36926 : b.failures += 1;
244 36926 : b.touched_at = Some(Instant::now());
245 36926 : if b.failures >= TRIP_AFTER {
246 32 : b.tripped_at = Some(Instant::now());
247 36894 : }
248 36926 : }
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 : });
264 8 : for url in [
265 4 : "http://alice:s3cret@[not-an-ip]/x",
266 4 : "http://alice:s3cret@/x",
267 4 : ] {
268 8 : let err = e.check_url(url).await.expect_err("refused");
269 8 : assert!(!err.contains("s3cret"), "userinfo in the reason: {err}");
270 4 : }
271 4 : }
272 :
273 : #[tokio::test]
274 4 : async fn scheme_allowlist_and_private_deny() {
275 4 : let e = Egress::new(antares_jsonld::EgressPolicy {
276 4 : allow_private: false,
277 4 : });
278 4 : assert!(e.check_url("file:///etc/passwd").await.is_err());
279 4 : assert!(e.check_url("http://127.0.0.1:9090/x").await.is_err());
280 4 : assert!(e
281 4 : .check_url("http://169.254.169.254/latest/meta-data")
282 4 : .await
283 4 : .is_err());
284 4 : let allow = Egress::new(antares_jsonld::EgressPolicy {
285 4 : allow_private: true,
286 4 : });
287 4 : assert!(allow.check_url("http://127.0.0.1:9090/x").await.is_ok());
288 4 : assert!(
289 4 : allow.check_url("mqtt://localhost:1883/t").await.is_err(),
290 : "@context fetches and federation forwards are HTTP; a notification \
291 : binding's own scheme goes through check_destination"
292 : );
293 : // A binding's own scheme is the sink's business, but its host is
294 : // still the policy's: a plugin binding cannot reach a denied host,
295 : // and an endpoint with no host is refused outright.
296 4 : assert!(allow
297 4 : .check_destination("wss://localhost:9000/n")
298 4 : .await
299 4 : .is_ok());
300 4 : assert!(e.check_destination("wss://127.0.0.1:9000/n").await.is_err());
301 4 : assert!(e
302 4 : .check_destination("wss://169.254.169.254/latest")
303 4 : .await
304 4 : .is_err());
305 4 : assert!(allow.check_destination("file:///etc/passwd").await.is_err());
306 4 : assert!(allow.check_destination("memory://").await.is_err());
307 4 : }
308 :
309 : /// The metadata denial does not depend on `ANTARES_EGRESS_ALLOW_PRIVATE`
310 : /// — but this check judges the destination as WRITTEN. A literal
311 : /// metadata address is refused in every spelling with private egress
312 : /// allowed; a host written as a NAME is not resolved here under that
313 : /// switch, and the classifier below is what the transports apply to the
314 : /// addresses the name turns out to stand for.
315 : #[tokio::test]
316 4 : async fn a_literal_metadata_address_is_refused_with_private_egress_allowed() {
317 4 : let allow = Egress::new(antares_jsonld::EgressPolicy {
318 4 : allow_private: true,
319 4 : });
320 24 : for u in [
321 4 : "http://169.254.169.254/latest/meta-data",
322 4 : "http://100.100.100.200/latest",
323 4 : "http://[fd00:ec2::254]/latest",
324 4 : "http://[::ffff:169.254.169.254]/latest",
325 4 : "http://[64:ff9b::a9fe:a9fe]/latest",
326 4 : // 6to4 (RFC 3056): 2002:169.254.169.254::
327 4 : "http://[2002:a9fe:a9fe::]/latest",
328 4 : ] {
329 24 : assert!(
330 24 : allow.check_url(u).await.is_err(),
331 4 : "{u} reached the instance-metadata range"
332 4 : );
333 24 : assert!(
334 24 : allow.check_destination(u).await.is_err(),
335 4 : "{u} reached the instance-metadata range as a binding endpoint"
336 4 : );
337 4 : }
338 4 : // The same addresses, as an answer a NAME resolves to: this is the
339 4 : // classifier `PolicyResolver` and the MQTT connect run before they
340 4 : // dial, and it is what covers the case the check above cannot see.
341 28 : for ip in [
342 4 : "169.254.169.254",
343 4 : "100.100.100.200",
344 4 : "fd00:ec2::254",
345 4 : "::ffff:169.254.169.254",
346 4 : "64:ff9b::a9fe:a9fe",
347 4 : "2002:a9fe:a9fe::",
348 4 : "2002:a9fe:a9fe:1:2:3:4:5",
349 4 : ] {
350 28 : assert!(
351 28 : antares_jsonld::EgressPolicy::ip_is_metadata(ip.parse().expect("address")),
352 4 : "{ip} not classified as instance metadata"
353 4 : );
354 4 : }
355 4 : }
356 :
357 : /// 5.2.34 + 5.5.10: the cooldown a failing registration earns belongs to
358 : /// ITS tenant. Another tenant's registration under the same client-chosen
359 : /// id keeps being contacted.
360 : #[test]
361 4 : fn cooldown_is_scoped_to_the_tenant_that_earned_it() {
362 4 : let e = Egress::default();
363 4 : let id = "urn:ngsi-ld:ContextSourceRegistration:shared";
364 4 : e.reg_record(®_key("tenant-a", id), false);
365 4 : assert!(
366 4 : e.reg_in_cooldown(®_key("tenant-a", id), 60_000),
367 : "the failing tenant's registration is in its window"
368 : );
369 4 : assert!(
370 4 : !e.reg_in_cooldown(®_key("tenant-b", id), 60_000),
371 : "one tenant's failing registration must not put another tenant's \
372 : same-id registration into timeout"
373 : );
374 : // and a success clears only its own tenant's stamp
375 4 : e.reg_record(®_key("tenant-b", id), false);
376 4 : e.reg_record(®_key("tenant-a", id), true);
377 4 : assert!(!e.reg_in_cooldown(®_key("tenant-a", id), 60_000));
378 4 : assert!(e.reg_in_cooldown(®_key("tenant-b", id), 60_000));
379 4 : }
380 :
381 : #[test]
382 4 : fn breaker_trips_after_consecutive_failures() {
383 4 : let e = Egress::default();
384 4 : let t = "tenant-a";
385 4 : let url = "http://dead.example:9090/notify";
386 4 : for _ in 0..(TRIP_AFTER - 1) {
387 16 : e.record_failure(t, url);
388 16 : assert!(!e.is_open(t, url), "not tripped before the threshold");
389 : }
390 4 : e.record_failure(t, url);
391 4 : assert!(e.is_open(t, url), "tripped at the threshold");
392 : // per-destination, not global
393 4 : assert!(!e.is_open(t, "http://healthy.example:9090/notify"));
394 4 : e.record_success(t, url);
395 4 : assert!(!e.is_open(t, url), "success clears the breaker");
396 4 : }
397 :
398 : /// 4.14 + 5.5.10: the breaker a failing endpoint earns belongs to the
399 : /// tenant whose delivery earned it. Tenants share destinations, and
400 : /// whether a destination answers inside the deadline is a property of the
401 : /// pair, not of the host: the same host is a timeout for a subscription
402 : /// at the 6.3.8 100 ms floor and healthy for one that allows 5 s.
403 : #[test]
404 4 : fn a_tripped_destination_is_tripped_only_for_the_tenant_that_tripped_it() {
405 4 : let e = Egress::default();
406 4 : let url = "http://shared-consumer.example:8080/notify";
407 20 : for _ in 0..TRIP_AFTER {
408 20 : e.record_failure("tenant-a", url);
409 20 : }
410 4 : assert!(e.is_open("tenant-a", url), "the failing tenant is tripped");
411 4 : assert!(
412 4 : !e.is_open("tenant-b", url),
413 : "one tenant's failing endpoint must not suppress another \
414 : tenant's notifications to the same host:port"
415 : );
416 : // and clearing one tenant's breaker leaves the other's alone
417 20 : for _ in 0..TRIP_AFTER {
418 20 : e.record_failure("tenant-b", url);
419 20 : }
420 4 : e.record_success("tenant-a", url);
421 4 : assert!(!e.is_open("tenant-a", url));
422 4 : assert!(e.is_open("tenant-b", url));
423 4 : }
424 :
425 : /// The ceiling is shared; the isolation must not be. A tenant churning
426 : /// destinations past it evicts its OWN oldest entry — another tenant's
427 : /// tripped breaker survives, or every later notification to that
428 : /// tenant's dead endpoint goes back to spending a full timeout.
429 : #[test]
430 4 : fn filling_the_ceiling_leaves_another_tenants_breaker_tripped() {
431 4 : let e = Egress::default();
432 4 : let victim = "http://dead-peer.example:9090/notify";
433 20 : for _ in 0..TRIP_AFTER {
434 20 : e.record_failure("b", victim);
435 20 : }
436 4 : assert!(e.is_open("b", victim), "the breaker starts tripped");
437 18384 : for i in 0..(MAX_TRACKED + 500) {
438 18384 : e.record_failure("a", &format!("http://churn-{i}.example:9090/notify"));
439 18384 : }
440 4 : assert!(
441 4 : e.is_open("b", victim),
442 : "one tenant's churn cleared another tenant's breaker"
443 : );
444 4 : }
445 :
446 : /// Both maps are keyed by client-supplied strings (notification endpoints,
447 : /// registration ids), so neither may grow without a ceiling: a client that
448 : /// points subscriptions at thousands of dead hosts must not be able to
449 : /// spend the broker's memory one entry at a time.
450 : #[test]
451 4 : fn destination_maps_stay_bounded_under_distinct_keys() {
452 4 : let e = Egress::default();
453 18384 : for i in 0..(MAX_TRACKED + 500) {
454 18384 : e.record_failure("t", &format!("http://dead-{i}.example:9090/notify"));
455 18384 : e.reg_record(&format!("urn:ngsi-ld:CSR:{i}"), false);
456 18384 : }
457 4 : assert!(
458 4 : e.breakers.lock().expect("breaker lock").len() <= MAX_TRACKED,
459 : "breaker map grew past the ceiling"
460 : );
461 4 : assert!(
462 4 : e.reg_failures.lock().expect("reg_failures lock").len() <= MAX_TRACKED,
463 : "registration cooldown map grew past the ceiling"
464 : );
465 : // The ceiling must not cost correctness for a live destination: the
466 : // most recently recorded failure is still tracked.
467 4 : let live = format!("http://dead-{}.example:9090/notify", MAX_TRACKED + 499);
468 20 : for _ in 0..TRIP_AFTER {
469 20 : e.record_failure("t", &live);
470 20 : }
471 4 : assert!(e.is_open("t", &live), "recent destination still trips");
472 4 : }
473 : }
|
