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