Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! HTTP notification binding — CIM 009 clause 6.3.8.
3 : //!
4 : //! A notification is an HTTP POST to `notification.endpoint.uri`. The MIME
5 : //! type is `endpoint.accept`, defaulting to `"application/json"`; for
6 : //! `"application/json"` (and, as this broker serves it, for
7 : //! `"application/geo+json"`) the request carries a Link header naming the
8 : //! JSON-LD `@context`. Each `endpoint.receiverInfo` pair becomes one custom
9 : //! header.
10 :
11 : use crate::{DeliveryError, DeliveryFuture, NotificationSink, Outbound};
12 : use antares_model::NgsiError;
13 : use std::time::Duration;
14 :
15 : /// The HTTP(S) binding over one shared outbound client. The client carries
16 : /// the deployment's egress policy (resolver pinning, connect timeouts); the
17 : /// per-destination policy check and breaker stay in the caller.
18 : pub struct HttpSink {
19 : client: antares_jsonld::HttpClient,
20 : }
21 :
22 : impl HttpSink {
23 : /// Bind to an already-configured outbound client.
24 4824 : pub fn new(client: antares_jsonld::HttpClient) -> Self {
25 4824 : Self { client }
26 4824 : }
27 : }
28 :
29 : /// An absolute http(s) URL with an authority and a host, the only endpoint
30 : /// this binding can send to. Every other form is BadRequestData, whose
31 : /// message names the URI with any userinfo stripped: it travels back in
32 : /// `detail` (5.5.3). `http:user:pass@host` (no `//`) is refused here, and
33 : /// its credential is not echoed.
34 1316 : pub fn check_http_endpoint(uri: &str) -> Result<(), NgsiError> {
35 1316 : let safe = crate::redact_userinfo(uri);
36 : // IETF RFC 3986 3.1: scheme names are case-insensitive, and
37 : // `SinkRegistry::scheme_of` already lowercases to pick this sink.
38 1316 : let rest = uri
39 1316 : .split_once("://")
40 1316 : .filter(|(s, _)| s.eq_ignore_ascii_case("http") || s.eq_ignore_ascii_case("https"))
41 1316 : .map(|(_, rest)| rest)
42 1316 : .ok_or_else(|| {
43 24 : NgsiError::BadRequestData(format!("not an http(s) endpoint URI: {safe:?}"))
44 24 : })?;
45 1292 : let authority_end = rest.find(['/', '?', '#']).unwrap_or(rest.len());
46 1292 : let host = rest[..authority_end]
47 1292 : .rsplit_once('@')
48 1292 : .map_or(&rest[..authority_end], |(_, h)| h);
49 1292 : if host.is_empty() {
50 4 : return Err(NgsiError::BadRequestData(format!(
51 4 : "http endpoint {safe:?} has no host"
52 4 : )));
53 1288 : }
54 1288 : Ok(())
55 1316 : }
56 :
57 : impl NotificationSink for HttpSink {
58 34822 : fn schemes(&self) -> &'static [&'static str] {
59 34822 : &["http", "https"]
60 34822 : }
61 :
62 : /// 5.2.15: the endpoint URI has to be dereferenceable — for this binding,
63 : /// an absolute http(s) URL with an authority.
64 1272 : fn parse_endpoint(&self, uri: &str, _notifier_info: &[(&str, &str)]) -> Result<(), NgsiError> {
65 1272 : check_http_endpoint(uri)
66 1272 : }
67 :
68 4958 : fn deliver<'a>(
69 4958 : &'a self,
70 4958 : uri: &'a str,
71 4958 : out: &'a Outbound,
72 4958 : timeout: Duration,
73 4958 : ) -> DeliveryFuture<'a> {
74 4958 : Box::pin(async move {
75 4958 : let bytes = antares_model::ordered_vec(&out.body);
76 : // Wasm: the page sink takes matching endpoints — a page cannot
77 : // listen on a socket, so this IS its delivery channel.
78 : #[cfg(target_arch = "wasm32")]
79 : if page_sink::try_deliver(uri, &bytes) {
80 : return Ok(());
81 : }
82 4958 : let mut req = self.client.post(uri);
83 10096 : for (k, v) in headers(out) {
84 10096 : req = req.header(k, v);
85 10096 : }
86 : // endpoint.timeout rides on the request natively (the client's
87 : // own total alone would let a stalled endpoint eat the full cap
88 : // per delivery); stretched under the sanitizer like the client's
89 : // other deadlines.
90 : #[cfg(not(target_arch = "wasm32"))]
91 4958 : let req =
92 4958 : req.timeout(timeout.saturating_mul(
93 4958 : u32::try_from(antares_jsonld::slow_factor()).unwrap_or(u32::MAX),
94 : ));
95 4958 : let deadline_ms = u32::try_from(timeout.as_millis()).unwrap_or(u32::MAX);
96 : // One Send unit so the admin replay handler stays Send on wasm32.
97 4958 : antares_jsonld::http_interaction(async move {
98 4958 : match antares_jsonld::io_deadline(req.body(bytes).send(), deadline_ms).await {
99 4706 : Some(Ok(r)) if r.status().is_success() => Ok(()),
100 50 : Some(Ok(r)) => Err(DeliveryError::failed(format!(
101 50 : "HTTP {}",
102 50 : r.status().as_u16()
103 50 : ))),
104 105 : Some(Err(e)) => Err(DeliveryError {
105 105 : timed_out: e.is_timeout(),
106 105 : message: crate::redact_userinfo(&e.to_string()),
107 105 : }),
108 0 : None => Err(DeliveryError::timeout("timeout")),
109 : }
110 4811 : })
111 4958 : .await
112 4811 : })
113 4958 : }
114 : }
115 :
116 : /// 6.3.8: the headers one notification POST carries. `application/ld+json`
117 : /// holds its `@context` in the payload body and takes no Link header; the
118 : /// other two MIME types carry it in the header. Every `receiverInfo` pair
119 : /// (and the tenant/snapshot markers the caller appended to it) becomes one
120 : /// custom header.
121 4972 : fn headers(out: &Outbound) -> Vec<(String, String)> {
122 4972 : let mut h = Vec::with_capacity(out.receiver_info.len() + 2);
123 4972 : h.push(("Content-Type".to_owned(), out.accept.clone()));
124 4972 : if out.accept != "application/ld+json" {
125 4954 : h.push(("Link".to_owned(), out.link.clone()));
126 4954 : }
127 : // The client APPENDS every header it is handed, so a receiverInfo pair
128 : // keyed like one of the two above would travel beside it rather than
129 : // replace it, and this clause names the source of both: the MIME type
130 : // comes from endpoint.accept, the Link from the served @context. The
131 : // colliding pair is dropped, not doubled.
132 4972 : let own = h.len();
133 4972 : for (k, v) in &out.receiver_info {
134 424 : if h[..own].iter().any(|(n, _)| n.eq_ignore_ascii_case(k)) {
135 12 : continue;
136 204 : }
137 204 : h.push((k.clone(), v.clone()));
138 : }
139 4972 : h
140 4972 : }
141 :
142 : /// The browser build has no inbound socket to receive notification callbacks
143 : /// on, so a subscription whose endpoint matches the registered URL prefix is
144 : /// delivered to page JS instead of the network. Endpoints outside the prefix
145 : /// still leave via fetch — the Node tier registers nothing and keeps pure
146 : /// HTTP delivery.
147 : #[cfg(target_arch = "wasm32")]
148 : pub mod page_sink {
149 : use std::sync::OnceLock;
150 :
151 : type Sink = Box<dyn Fn(&str, &[u8]) -> bool + Send + Sync>;
152 : type Hook = (String, Sink);
153 : static HOOK: OnceLock<Hook> = OnceLock::new();
154 :
155 : /// Register the sink (once per module instance).
156 : pub fn set(prefix: String, h: Sink) {
157 : let _ = HOOK.set((prefix, h));
158 : }
159 :
160 : /// True when the page sink claimed (and thus delivered) this endpoint.
161 : pub fn try_deliver(url: &str, body: &[u8]) -> bool {
162 : match HOOK.get() {
163 : Some((prefix, h)) if url.starts_with(prefix.as_str()) => h(url, body),
164 : _ => false,
165 : }
166 : }
167 : }
168 :
169 : #[cfg(test)]
170 : mod tests {
171 : use super::*;
172 : use serde_json::json;
173 :
174 14 : fn out(accept: &str, receiver_info: &[(&str, &str)]) -> Outbound {
175 : Outbound {
176 14 : body: json!({"type": "Notification"}),
177 14 : accept: accept.to_owned(),
178 14 : link: "<https://ctx>; rel=\"http://www.w3.org/ns/json-ld#context\"".to_owned(),
179 14 : receiver_info: receiver_info
180 14 : .iter()
181 18 : .map(|(k, v)| ((*k).to_owned(), (*v).to_owned()))
182 14 : .collect(),
183 14 : notifier_info: Vec::new(),
184 : }
185 14 : }
186 :
187 8 : fn sink() -> HttpSink {
188 8 : HttpSink::new(
189 8 : antares_jsonld::client_builder(antares_jsonld::EgressPolicy {
190 8 : allow_private: true,
191 8 : private_hosts: &[],
192 8 : })
193 8 : .build()
194 8 : .expect("client"),
195 : )
196 8 : }
197 :
198 : /// 6.3.8: json and geo+json carry the @context in a Link header,
199 : /// ld+json carries it in the body and takes none.
200 : #[test]
201 2 : fn link_header_follows_the_target_mime_type() {
202 6 : let names = |o: &Outbound| {
203 6 : headers(o)
204 6 : .into_iter()
205 6 : .map(|(k, _)| k)
206 6 : .collect::<Vec<String>>()
207 6 : };
208 2 : assert_eq!(
209 2 : names(&out("application/json", &[])),
210 : ["Content-Type", "Link"]
211 : );
212 2 : assert_eq!(
213 2 : names(&out("application/geo+json", &[])),
214 : ["Content-Type", "Link"]
215 : );
216 2 : assert_eq!(names(&out("application/ld+json", &[])), ["Content-Type"]);
217 2 : }
218 :
219 : /// Every receiverInfo pair becomes one custom header, in order, after
220 : /// the binding's own.
221 : #[test]
222 2 : fn receiver_info_becomes_custom_headers() {
223 2 : let h = headers(&out(
224 2 : "application/json",
225 2 : &[("Authorization", "Bearer t"), ("NGSILD-Tenant", "acme")],
226 2 : ));
227 2 : assert_eq!(h[0].0, "Content-Type");
228 2 : assert_eq!(h[0].1, "application/json");
229 2 : assert_eq!(h[2], ("Authorization".to_owned(), "Bearer t".to_owned()));
230 2 : assert_eq!(h[3], ("NGSILD-Tenant".to_owned(), "acme".to_owned()));
231 2 : }
232 :
233 : /// 6.3.8 fixes the MIME type of the POST to `endpoint.accept` and, for
234 : /// the other two types, mandates the Link header carrying the @context
235 : /// reference. The client appends each header it is handed, so a
236 : /// receiverInfo pair keyed like either one travels BESIDE the
237 : /// binding's: two Content-Type fields make the request malformed (IETF
238 : /// RFC 9110 clause 5.5.1 makes it a singleton field) and a second Link
239 : /// leaves the receiver two @context references to choose between.
240 : #[test]
241 2 : fn receiver_info_cannot_double_the_headers_the_binding_owns() {
242 4 : for accept in ["application/json", "application/geo+json"] {
243 4 : let h = headers(&out(
244 4 : accept,
245 4 : &[
246 4 : ("content-type", "text/plain"),
247 4 : ("LINK", "<https://evil>"),
248 4 : ("X-Kept", "yes"),
249 4 : ],
250 4 : ));
251 20 : let named = |n: &str| {
252 20 : h.iter()
253 60 : .filter(|(k, _)| k.eq_ignore_ascii_case(n))
254 20 : .collect::<Vec<_>>()
255 20 : };
256 4 : assert_eq!(named("Content-Type").len(), 1, "{accept}");
257 4 : assert_eq!(named("Content-Type")[0].1, accept);
258 4 : assert_eq!(named("Link").len(), 1, "{accept}");
259 4 : assert!(named("Link")[0].1.contains("json-ld#context"), "{accept}");
260 4 : assert_eq!(named("X-Kept").len(), 1, "an ordinary pair still travels");
261 : }
262 : // ld+json sets no Link of its own, so a pair keyed Link is ordinary
263 2 : let h = headers(&out("application/ld+json", &[("Link", "<https://x>")]));
264 2 : assert_eq!(
265 2 : h.iter()
266 4 : .filter(|(k, _)| k.eq_ignore_ascii_case("Link"))
267 2 : .count(),
268 : 1
269 : );
270 2 : }
271 :
272 : /// 5.2.15 dereferenceable URI: this binding needs an absolute http(s)
273 : /// URL with a host.
274 : #[test]
275 2 : fn endpoint_validation_needs_an_absolute_url_with_a_host() {
276 2 : let s = sink();
277 2 : assert!(s.parse_endpoint("http://example.org/notify", &[]).is_ok());
278 2 : assert!(s.parse_endpoint("https://example.org", &[]).is_ok());
279 2 : assert!(s.parse_endpoint("https://u:p@example.org/n", &[]).is_ok());
280 10 : for bad in [
281 2 : "example.org/notify",
282 2 : "http:/notify",
283 2 : "ftp://example.org/n",
284 2 : "http:///notify",
285 2 : "https://@/n",
286 2 : ] {
287 10 : let err = s.parse_endpoint(bad, &[]).expect_err(bad);
288 10 : assert_eq!(err.status(), 400, "{bad}");
289 : }
290 2 : }
291 :
292 : /// IETF RFC 3986 3.1: "schemes are case-insensitive ... an implementation
293 : /// should accept uppercase letters as equivalent to lowercase in scheme
294 : /// names". `SinkRegistry::scheme_of` lowercases before it picks a sink, so
295 : /// `HTTP://…` is routed HERE — and this is the layer that decides, so a
296 : /// case-sensitive check turns a legal endpoint URI into a 400 the
297 : /// registry's own contract says it should not be.
298 : #[test]
299 2 : fn the_scheme_is_matched_case_insensitively() {
300 2 : let s = sink();
301 6 : for uri in [
302 2 : "HTTP://example.org/notify",
303 2 : "HTTPS://example.org/n",
304 2 : "HttP://example.org/n",
305 2 : ] {
306 6 : assert!(s.parse_endpoint(uri, &[]).is_ok(), "{uri}");
307 : }
308 2 : }
309 :
310 : /// The HTTP binding hands a DRIVER ERROR to `redact_userinfo`, not a bare
311 : /// URI: reqwest prints the failing URL and `url::Url`'s Display serializes
312 : /// the password with it. Every message shape the driver produces has to
313 : /// lose the credential.
314 : #[test]
315 2 : fn a_driver_error_carrying_the_url_loses_the_password() {
316 8 : for msg in [
317 2 : "error sending request for url (https://alice:s3cret@host/notify)",
318 2 : "error sending request for url (https://alice:s3cret@host)",
319 2 : "error sending request for url (https://alice:s3cret@host): error trying to connect",
320 2 : "error sending request for url (https://alice:s3cret@host/n): dns error: no such host",
321 2 : ] {
322 8 : let red = crate::redact_userinfo(msg);
323 8 : assert!(!red.contains("s3cret"), "{msg} -> {red}");
324 8 : assert!(!red.contains("alice"), "{msg} -> {red}");
325 8 : assert!(
326 8 : red.contains("host"),
327 : "the destination stays readable: {red}"
328 : );
329 : }
330 2 : }
331 :
332 : /// A rejected endpoint travels back to the client in `detail` (5.5.3)
333 : /// and into the logs: the userinfo credentials never ride along.
334 : #[test]
335 2 : fn rejection_message_carries_no_credentials() {
336 2 : let err = sink()
337 2 : .parse_endpoint("ftp://user:hunter2@example.org/n", &[])
338 2 : .expect_err("not http");
339 2 : let text = format!("{err}");
340 2 : assert!(!text.contains("hunter2"), "{text}");
341 2 : assert!(text.contains("example.org"), "{text}");
342 2 : }
343 :
344 : #[test]
345 2 : fn serves_exactly_the_two_http_schemes() {
346 2 : assert_eq!(sink().schemes(), &["http", "https"]);
347 2 : }
348 : }
|
